diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java index a5a65cd9..756f0fc2 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2011 the original author or authors. + * Copyright 2010-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -21,6 +21,7 @@ import org.w3c.dom.Element; /** * @author Dave Syer + * @author Gary Russell */ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { @@ -38,6 +39,8 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { private static final String PASSWORD_ATTRIBUTE = "password"; + private static final String EXECUTOR_ATTRIBUTE = "executor"; + @Override protected Class getBeanClass(Element element) { return CachingConnectionFactory.class; @@ -63,7 +66,7 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { NamespaceUtils.setValueIfAttributeDefined(builder, element, USER_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, PASSWORD_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, VIRTUAL_HOST_ATTRIBUTE); - + NamespaceUtils.setReferenceIfAttributeDefined(builder, element, EXECUTOR_ATTRIBUTE); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index 98a24205..65058fa8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -16,14 +16,18 @@ import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.Assert; /** * @author Dave Syer + * @author Gary Russell * */ public abstract class AbstractConnectionFactory implements ConnectionFactory, DisposableBean { @@ -36,6 +40,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di private final CompositeChannelListener channelListener = new CompositeChannelListener(); + private volatile ExecutorService executorService; + /** * Create a new SingleConnectionFactory for the given target ConnectionFactory. * @param rabbitConnectionFactory the target ConnectionFactory @@ -111,9 +117,28 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di this.channelListener.addDelegate(listener); } + /** + * Provide an Executor for + * use by the Rabbit ConnectionFactory when creating connections. + * Can either be an ExecutorService or a Spring + * ThreadPoolTaskExecutor, as defined by a <task:executor/> element. + * @param executor The executor. + */ + public void setExecutor(Executor executor) { + boolean isExecutorService = executor instanceof ExecutorService; + boolean isThreadPoolTaskExecutor = executor instanceof ThreadPoolTaskExecutor; + Assert.isTrue(isExecutorService || isThreadPoolTaskExecutor); + if (isExecutorService) { + this.executorService = (ExecutorService) executor; + } + else { + this.executorService = ((ThreadPoolTaskExecutor) executor).getThreadPoolExecutor(); + } + } + final protected Connection createBareConnection() { try { - return new SimpleConnection(this.rabbitConnectionFactory.newConnection()); + return new SimpleConnection(this.rabbitConnectionFactory.newConnection(this.executorService)); } catch (IOException e) { throw RabbitUtils.convertRabbitAccessException(e); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index ed5ac442..0f431d82 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -379,6 +379,9 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { MessageProperties messageProperties = messagePropertiesConverter.toMessageProperties( properties, envelope, encoding); Message reply = new Message(body, messageProperties); + if (logger.isTraceEnabled()) { + logger.trace("Message received " + reply); + } try { replyHandoff.put(reply); } catch (InterruptedException e) { diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd index ad08e113..03d44624 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd @@ -774,6 +774,20 @@ + + + + element). Passed to the Rabbit library when creating the connection. When not supplied, the + Rabbit library currently uses a fixed thread pool ExecutorService with 5 threads. + ]]> + + + + + + + diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java index d90711a6..e7c65ae6 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2011 the original author or authors. + * Copyright 2010-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -15,16 +15,23 @@ package org.springframework.amqp.rabbit.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; + +import java.util.concurrent.ExecutorService; import org.junit.Before; import org.junit.Test; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.xml.XmlBeanFactory; import org.springframework.core.io.ClassPathResource; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; /** * * @author Dave Syer + * @author Gary Russell * */ public final class ConnectionFactoryParserTests { @@ -41,6 +48,7 @@ public final class ConnectionFactoryParserTests { CachingConnectionFactory connectionFactory = beanFactory.getBean("kitchenSink", CachingConnectionFactory.class); assertNotNull(connectionFactory); assertEquals(10, connectionFactory.getChannelCacheSize()); + assertNull(new DirectFieldAccessor(connectionFactory).getPropertyValue("executorService")); } @Test @@ -49,5 +57,27 @@ public final class ConnectionFactoryParserTests { assertNotNull(connectionFactory); assertEquals(10, connectionFactory.getChannelCacheSize()); } - + + @Test + public void testWithExecutor() throws Exception { + CachingConnectionFactory connectionFactory = beanFactory.getBean("withExecutor", CachingConnectionFactory.class); + assertNotNull(connectionFactory); + assertEquals(10, connectionFactory.getChannelCacheSize()); + Object executor = new DirectFieldAccessor(connectionFactory).getPropertyValue("executorService"); + assertNotNull(executor); + ThreadPoolTaskExecutor exec = beanFactory.getBean("exec", ThreadPoolTaskExecutor.class); + assertSame(exec.getThreadPoolExecutor(), executor); + } + + @Test + public void testWithExecutorService() throws Exception { + CachingConnectionFactory connectionFactory = beanFactory.getBean("withExecutorService", CachingConnectionFactory.class); + assertNotNull(connectionFactory); + assertEquals(10, connectionFactory.getChannelCacheSize()); + Object executor = new DirectFieldAccessor(connectionFactory).getPropertyValue("executorService"); + assertNotNull(executor); + ExecutorService exec = beanFactory.getBean("execService", ExecutorService.class); + assertSame(exec, executor); + } + } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java index 747b640d..0716c8e1 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java @@ -10,6 +10,7 @@ import static org.mockito.Mockito.when; import java.io.IOException; import java.util.Arrays; +import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; @@ -28,7 +29,7 @@ public abstract class AbstractConnectionFactoryTests { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); final AtomicInteger called = new AtomicInteger(0); AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory); @@ -55,7 +56,7 @@ public abstract class AbstractConnectionFactoryTests { assertEquals(0, called.get()); verify(mockConnection, atLeastOnce()).close(); - verify(mockConnectionFactory, times(1)).newConnection(); + verify(mockConnectionFactory, times(1)).newConnection((ExecutorService) null); } @@ -65,7 +66,7 @@ public abstract class AbstractConnectionFactoryTests { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); final AtomicInteger called = new AtomicInteger(0); AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory); @@ -93,7 +94,7 @@ public abstract class AbstractConnectionFactoryTests { assertEquals(0, called.get()); verify(mockConnection, atLeastOnce()).close(); - verify(mockConnectionFactory, times(1)).newConnection(); + verify(mockConnectionFactory, times(1)).newConnection((ExecutorService) null); } @@ -104,7 +105,7 @@ public abstract class AbstractConnectionFactoryTests { com.rabbitmq.client.Connection mockConnection1 = mock(com.rabbitmq.client.Connection.class); com.rabbitmq.client.Connection mockConnection2 = mock(com.rabbitmq.client.Connection.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection1).thenReturn(mockConnection2); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection1).thenReturn(mockConnection2); // simulate a dead connection when(mockConnection1.isOpen()).thenReturn(false); @@ -113,7 +114,7 @@ public abstract class AbstractConnectionFactoryTests { Connection connection = connectionFactory.createConnection(); // the dead connection should be discarded connection.createChannel(false); - verify(mockConnectionFactory, times(2)).newConnection(); + verify(mockConnectionFactory, times(2)).newConnection((ExecutorService) null); verify(mockConnection2, times(1)).createChannel(); connectionFactory.destroy(); @@ -129,7 +130,7 @@ public abstract class AbstractConnectionFactoryTests { AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory); connectionFactory.destroy(); - verify(mockConnectionFactory, never()).newConnection(); + verify(mockConnectionFactory, never()).newConnection((ExecutorService) null); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java index 7d3a52fd..3a340b27 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java @@ -11,6 +11,7 @@ import static org.mockito.Mockito.when; import java.io.IOException; import java.util.Arrays; import java.util.List; +import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicInteger; import junit.framework.Assert; @@ -39,7 +40,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); Channel mockChannel = mock(Channel.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.createChannel()).thenReturn(mockChannel); when(mockChannel.isOpen()).thenReturn(true); when(mockConnection.isOpen()).thenReturn(true); @@ -72,7 +73,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest Channel mockChannel1 = mock(Channel.class); Channel mockChannel2 = mock(Channel.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2); @@ -125,7 +126,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest Channel mockChannel2 = mock(Channel.class); Channel mockChannel3 = mock(Channel.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2).thenReturn(mockChannel3); when(mockConnection.isOpen()).thenReturn(true); @@ -180,7 +181,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest Channel mockChannel1 = mock(Channel.class); Channel mockChannel2 = mock(Channel.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2); when(mockConnection.isOpen()).thenReturn(true); @@ -227,7 +228,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest Channel mockChannel1 = mock(Channel.class); Channel mockChannel2 = mock(Channel.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2); when(mockConnection.isOpen()).thenReturn(true); @@ -287,7 +288,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest Assert.assertNotSame(mockChannel1, mockChannel2); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); // You can't repeat 'when' statements for stubbing consecutive calls to // the same method to returning different // values. @@ -354,7 +355,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); Channel mockChannel = mock(Channel.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); when(mockChannel.isOpen()).thenReturn(true); when(mockConnection.createChannel()).thenReturn(mockChannel); @@ -383,7 +384,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest connectionFactory.destroy(); verify(mockConnection, atLeastOnce()).close(); - verify(mockConnectionFactory).newConnection(); + verify(mockConnectionFactory).newConnection((ExecutorService) null); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java index a9c82dfd..45afd6d3 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java @@ -9,6 +9,7 @@ import static org.mockito.Mockito.when; import java.io.IOException; import java.util.Arrays; +import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; @@ -33,7 +34,7 @@ public class SingleConnectionFactoryTests extends AbstractConnectionFactoryTests com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); Channel mockChannel = mock(Channel.class); - when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); when(mockConnection.createChannel()).thenReturn(mockChannel); @@ -60,7 +61,7 @@ public class SingleConnectionFactoryTests extends AbstractConnectionFactoryTests connectionFactory.destroy(); verify(mockConnection, atLeastOnce()).close(); - verify(mockConnectionFactory).newConnection(); + verify(mockConnectionFactory).newConnection((ExecutorService) null); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 48542d3e..ff7d6239 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2011 the original author or authors. + * Copyright 2010-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -17,17 +17,24 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.when; +import java.lang.reflect.Field; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.commons.logging.Log; import org.junit.Before; import org.junit.Rule; import org.junit.Test; - +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; @@ -39,6 +46,7 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.rabbit.test.BrokerRunning; import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionException; import org.springframework.transaction.TransactionStatus; @@ -46,6 +54,9 @@ import org.springframework.transaction.support.AbstractPlatformTransactionManage import org.springframework.transaction.support.DefaultTransactionStatus; import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.ReflectionUtils.FieldCallback; +import org.springframework.util.ReflectionUtils.FieldFilter; import com.rabbitmq.client.Channel; import com.rabbitmq.client.GetResponse; @@ -319,6 +330,76 @@ public class RabbitTemplateIntegrationTests { assertEquals(null, reply); } + @Test + public void testAtomicSendAndReceiveExternalExecutor() throws Exception { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor(); + final String execName = "make-sure-exec-passed-in"; + exec.setBeanName(execName); + exec.afterPropertiesSet(); + connectionFactory.setExecutor(exec); + final Field[] fields = new Field[1]; + ReflectionUtils.doWithFields(RabbitTemplate.class, new FieldCallback() { + public void doWith(Field field) throws IllegalArgumentException, + IllegalAccessException { + field.setAccessible(true); + fields[0] = field; + } + }, new FieldFilter() { + public boolean matches(Field field) { + return field.getName().equals("logger"); + } + }); + Log logger = Mockito.mock(Log.class); + when(logger.isTraceEnabled()).thenReturn(true); + + final AtomicBoolean execConfiguredOk = new AtomicBoolean(); + + doAnswer(new Answer(){ + public Object answer(InvocationOnMock invocation) throws Throwable { + String log = (String) invocation.getArguments()[0]; + if (log.startsWith("Message received") && + Thread.currentThread().getName().startsWith(execName)) { + execConfiguredOk.set(true); + } + return null; + } + }).when(logger).trace(Mockito.anyString()); + final RabbitTemplate template = new RabbitTemplate(connectionFactory); + ReflectionUtils.setField(fields[0], template, logger); + template.setRoutingKey(ROUTE); + template.setQueue(ROUTE); + ExecutorService executor = Executors.newFixedThreadPool(1); + // Set up a consumer to respond to our producer + Future received = executor.submit(new Callable() { + + public Message call() throws Exception { + Message message = null; + for (int i = 0; i < 10; i++) { + message = template.receive(); + if (message != null) { + break; + } + Thread.sleep(100L); + } + assertNotNull("No message received", message); + template.send(message.getMessageProperties().getReplyTo(), message); + return message; + } + + }); + Message message = new Message("test-message".getBytes(), new MessageProperties()); + Message reply = template.sendAndReceive(message); + assertEquals(new String(message.getBody()), new String(received.get(1000, TimeUnit.MILLISECONDS).getBody())); + assertNotNull("Reply is expected", reply); + assertEquals(new String(message.getBody()), new String(reply.getBody())); + // Message was consumed so nothing left on queue + reply = template.receive(); + assertEquals(null, reply); + + assertTrue(execConfiguredOk.get()); + } + @Test public void testAtomicSendAndReceiveWithRoutingKey() throws Exception { final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory()); diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml index 4ed92b34..0d9dd2c7 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml @@ -1,8 +1,14 @@ - + @@ -11,4 +17,16 @@ + + + + + + + +