From 5ea317798f31ac3fb77dfc0cc3f952e8a7bb4981 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 20 Apr 2012 14:58:33 -0400 Subject: [PATCH] AMQP-225 Add Support for External Executor If no ExecutorService is provided when creating a new connection, Rabbit uses a default fixed thread pool executor with 5 threads. You can now specify an executor on the . Due to limitations in the Rabbit API, this must either reference a ExecutorService (such as one returned by Executors.new...Executor() methods, or a Spring ThreadPoolTaskExecutor, such as one defined using the element. The underlying API (newConnection()) with no arg calls (newConnection(null)), so it is safe to always use newConnection(exec), and not test for null. --- .../config/ConnectionFactoryParser.java | 7 +- .../connection/AbstractConnectionFactory.java | 29 ++++++- .../amqp/rabbit/core/RabbitTemplate.java | 3 + .../amqp/rabbit/config/spring-rabbit-1.0.xsd | 14 +++ .../config/ConnectionFactoryParserTests.java | 34 +++++++- .../AbstractConnectionFactoryTests.java | 15 ++-- .../CachingConnectionFactoryTests.java | 17 ++-- .../SingleConnectionFactoryTests.java | 5 +- .../core/RabbitTemplateIntegrationTests.java | 85 ++++++++++++++++++- .../ConnectionFactoryParserTests-context.xml | 26 +++++- 10 files changed, 206 insertions(+), 29 deletions(-) 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 @@ + + + + + + + +