diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java index 0c12ff33..228534db 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java @@ -87,6 +87,10 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea this.rabbitConnectionFactory.setPassword(password); } + public void setHost(String host) { + this.rabbitConnectionFactory.setHost(host); + } + public String getHost() { return this.rabbitConnectionFactory.getHost(); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 90152b1b..ca270f97 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -356,7 +356,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im return this.active; } } - + /** * Start this container. * @see #doStart @@ -432,24 +432,10 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im */ public final boolean isRunning() { synchronized (this.lifecycleMonitor) { - return (this.running && runningAllowed()); + return (this.running); } } - /** - * Check whether this container's listeners are generally allowed to run. - *

- * This implementation always returns true; the default 'running' state is purely determined by - * {@link #start()} / {@link #stop()}. - *

- * Subclasses may override this method to check against temporary conditions that prevent listeners from actually - * running. In other words, they may apply further restrictions to the 'running' state, returning false - * if such a restriction prevents listeners from running. - */ - protected boolean runningAllowed() { - return true; - } - // ------------------------------------------------------------------------- // Management of a shared Rabbit Connection // ------------------------------------------------------------------------- diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 500c31ee..f1f1d572 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -167,7 +167,7 @@ public class BlockingQueueConsumer { @Override public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) { if (logger.isDebugEnabled()) { - logger.debug("Received shutdown for consumer tag=" + consumerTag, sig); + logger.debug("Received shutdown signal for consumer tag=" + consumerTag, sig); } shutdown = sig; // TODO: interrupt? diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 4e27b13d..ec510baa 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java @@ -20,19 +20,22 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; -import org.springframework.amqp.AmqpException; +import org.aopalliance.aop.Advice; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils; import org.springframework.amqp.rabbit.connection.RabbitResourceHolder; import org.springframework.amqp.rabbit.listener.adapter.ListenerExecutionFailedException; +import org.springframework.aop.Pointcut; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.support.DefaultPointcutAdvisor; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.TransactionStatus; -import org.springframework.transaction.support.DefaultTransactionDefinition; -import org.springframework.transaction.support.TransactionCallback; -import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.transaction.interceptor.DefaultTransactionAttribute; +import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionAttribute; +import org.springframework.transaction.interceptor.TransactionInterceptor; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -63,16 +66,47 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT; - private volatile Set consumers; + private Set consumers; private final Object consumersMonitor = new Object(); private PlatformTransactionManager transactionManager; - private DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition(); + private TransactionAttribute transactionAttribute = new DefaultTransactionAttribute(); private CountDownLatch cancellationLock; + public static interface ContainerDelegate { + boolean receiveAndExecute(BlockingQueueConsumer consumer) throws Throwable; + } + + private Advice[] advices = new Advice[0]; + + private ContainerDelegate delegate = new ContainerDelegate() { + public boolean receiveAndExecute(BlockingQueueConsumer consumer) throws Throwable { + return SimpleMessageListenerContainer.this.receiveAndExecute(consumer); + } + }; + + private ContainerDelegate proxy = delegate; + + /** + *

+ * Public setter for the {@link Advice} to apply to listener executions. If {@link #setTxSize(int) txSize>1} then + * multiple listener executions will all be wrapped in the same advice up to that limit. + *

+ *

+ * If a {@link #setTransactionManager(PlatformTransactionManager) transactionManager} is provided as well, then + * separate advice is created for the transaction and applied first in the chain. In that case the advice chain + * provided here should not contain a transaction interceptor (otherwise two transactions would be be applied). + *

+ * + * @param advices the advice chain to set + */ + public void setAdviceChain(Advice[] advices) { + this.advices = advices; + } + public SimpleMessageListenerContainer() { } @@ -137,6 +171,13 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta this.transactionManager = transactionManager; } + /** + * @param transactionAttribute the transaction attribute to set + */ + public void setTransactionAttribute(TransactionAttribute transactionAttribute) { + this.transactionAttribute = transactionAttribute; + } + /** * Avoid the possibility of not configuring the CachingConnectionFactory in sync with the number of concurrent * consumers. @@ -172,6 +213,26 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } + public void initializeProxy() { + if (advices.length == 0 && transactionManager == null) { + return; + } + ProxyFactory factory = new ProxyFactory(); + if (transactionManager != null) { + MatchAlwaysTransactionAttributeSource txAttributeSource = new MatchAlwaysTransactionAttributeSource(); + txAttributeSource.setTransactionAttribute(transactionAttribute); + Advice txAdvice = new TransactionInterceptor(transactionManager, txAttributeSource); + factory.addAdvisor(new DefaultPointcutAdvisor(Pointcut.TRUE, txAdvice)); + } + for (Advice advice : advices) { + factory.addAdvisor(new DefaultPointcutAdvisor(Pointcut.TRUE, advice)); + } + factory.setProxyTargetClass(false); + factory.addInterface(ContainerDelegate.class); + factory.setTarget(delegate); + proxy = (ContainerDelegate) factory.getProxy(); + } + // ------------------------------------------------------------------------- // Implementation of AbstractMessageListenerContainer's template methods // ------------------------------------------------------------------------- @@ -191,6 +252,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta */ protected void doInitialize() throws Exception { establishSharedConnection(); + initializeProxy(); + } + + public int getActiveConsumerCount() { + return (int) cancellationLock.getCount(); } /** @@ -209,8 +275,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } cancellationLock = new CountDownLatch(this.consumers.size()); for (BlockingQueueConsumer consumer : this.consumers) { - this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, this.txSize, this, - cancellationLock)); + this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, cancellationLock)); } } } @@ -231,13 +296,13 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta logger.debug("Waiting for workers to finish."); boolean finished = cancellationLock.await(shutdownTimeout, TimeUnit.MILLISECONDS); if (finished) { - logger.debug("Successfully waited for workers to finish."); + logger.info("Successfully waited for workers to finish."); } else { - logger.debug("Workers not finished. Forcing connections to close."); + logger.info("Workers not finished. Forcing connections to close."); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); - logger.debug("Interrupted waiting for workers. Continuing with shutdown."); + logger.warn("Interrupted waiting for workers. Continuing with shutdown."); } synchronized (this.consumersMonitor) { @@ -278,41 +343,60 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta try { // Need to recycle the channel in this consumer consumer.stop(); - } catch (Exception e) { - // Ignore + this.consumers.remove(consumer); + Channel channel = getTransactionalResourceHolder().getChannel(); + consumer = createBlockingQueueConsumer(channel); + this.consumers.add(consumer); + } catch (RuntimeException e) { + // Ensure consumer counts are correct (another is not going to start because of the exception, but + // we haven't counted down yet) + logger.warn("Consumer died on restart. " + e.getClass() + ": " + e.getMessage()); + cancellationLock.countDown(); + // Thrown into the void (probably) in a background thread. Oh well, here goes... + throw e; } - this.consumers.remove(consumer); - Channel channel = getTransactionalResourceHolder().getChannel(); - consumer = createBlockingQueueConsumer(channel); - this.consumers.add(consumer); - this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, this.txSize, this, - cancellationLock)); + this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, cancellationLock)); } } } + private boolean receiveAndExecute(BlockingQueueConsumer consumer) throws Throwable { + + Channel channel = consumer.getChannel(); + + int totalMsgCount = 0; + + ConnectionFactory connectionFactory = getConnectionFactory(); + if (getAcknowledgeMode().isTransactionAllowed()) { + ConnectionFactoryUtils + .bindResourceToTransaction(new RabbitResourceHolder(channel), connectionFactory, true); + } + + for (int i = 0; i < txSize; i++) { + + logger.debug("Waiting for message from consumer."); + Message message = consumer.nextMessage(receiveTimeout); + if (message == null) { + return false; + } + totalMsgCount++; + executeListener(channel, message); + + } + + return true; + + } + private class AsyncMessageProcessingConsumer implements Runnable { private final BlockingQueueConsumer consumer; - private int txSize; - - private final PlatformTransactionManager transactionManager; - - private final DefaultTransactionDefinition transactionDefinition; - - private long receiveTimeout; - private final CountDownLatch latch; - public AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer, int txSize, - SimpleMessageListenerContainer messageListenerContainer, CountDownLatch latch) { + public AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer, CountDownLatch latch) { this.consumer = consumer; - this.txSize = txSize; this.latch = latch; - this.transactionManager = messageListenerContainer.transactionManager; - this.transactionDefinition = messageListenerContainer.transactionDefinition; - this.receiveTimeout = messageListenerContainer.receiveTimeout; } public void run() { @@ -325,26 +409,22 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta boolean continuable = false; while (isActive() || continuable) { try { - if (this.transactionManager != null) { - // Execute within transaction. - transactionalReceiveAndExecute(); - } else { - // Will come back false when the queue is drained - continuable = receiveAndExecute() && !isChannelTransacted(); - } + // Will come back false when the queue is drained + continuable = proxy.receiveAndExecute(consumer) && !isChannelTransacted(); } catch (ListenerExecutionFailedException ex) { // Continue to process, otherwise re-throw } } + } catch (InterruptedException e) { logger.debug("Consumer thread interrupted, processing stopped."); Thread.currentThread().interrupt(); } catch (Throwable t) { logger.debug("Consumer received fatal exception, processing stopped.", t); } finally { - latch.countDown(); if (!isActive()) { logger.debug("Cancelling " + consumer); + latch.countDown(); consumer.stop(); } else { logger.debug("Restarting " + consumer); @@ -354,61 +434,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } - private boolean transactionalReceiveAndExecute() throws Exception { - try { - return new TransactionTemplate(this.transactionManager, this.transactionDefinition) - .execute(new TransactionCallback() { - public Boolean doInTransaction(TransactionStatus status) { - try { - return receiveAndExecute(); - } catch (ListenerExecutionFailedException ex) { - // These are expected - throw ex; - } catch (Exception ex) { - throw new AmqpException("Unexpected exception on listener execution", ex); - } catch (Error err) { - throw err; - } catch (Throwable t) { - throw new AmqpException(t); - } - } - }); - } catch (Exception ex) { - throw ex; - } catch (Error err) { - throw err; - } catch (Throwable t) { - throw new AmqpException(t); - } - } - - private boolean receiveAndExecute() throws Throwable { - - Channel channel = consumer.getChannel(); - - int totalMsgCount = 0; - - ConnectionFactory connectionFactory = getConnectionFactory(); - if (getAcknowledgeMode().isTransactionAllowed()) { - ConnectionFactoryUtils.bindResourceToTransaction(new RabbitResourceHolder(channel), connectionFactory, - true); - } - - for (int i = 0; i < txSize; i++) { - - logger.debug("Waiting for message from consumer."); - Message message = consumer.nextMessage(receiveTimeout); - if (message == null) { - return false; - } - totalMsgCount++; - executeListener(channel, message); - - } - - return true; - - } } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitUtils.java index 2b608632..7663f5fe 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitUtils.java @@ -69,7 +69,7 @@ public abstract class RabbitUtils { * @param channel the RabbitMQ Channel to close (may be null) */ public static void closeChannel(Channel channel) { - if (channel != null) { + if (channel != null && channel.isOpen()) { try { channel.close(); } catch (IOException ex) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java index 53e9a30f..991f514d 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java @@ -28,6 +28,7 @@ import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.test.BrokerPanic; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; /** @@ -38,10 +39,6 @@ import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; */ public class RabbitBrokerAdminIntegrationTests { - private static final int PORT = 15672; - - private static final String NODE_NAME = "spring@localhost"; - @Rule public Log4jLevelAdjuster logLevel = new Log4jLevelAdjuster(Level.INFO, RabbitBrokerAdmin.class); @@ -56,10 +53,7 @@ public class RabbitBrokerAdminIntegrationTests { @BeforeClass public static void start() throws Exception { // Set up broker admin for non-root user - brokerAdmin = new RabbitBrokerAdmin(NODE_NAME, PORT); - brokerAdmin.setRabbitLogBaseDirectory("target/rabbitmq/log"); - brokerAdmin.setRabbitMnesiaBaseDirectory("target/rabbitmq/mnesia"); - brokerAdmin.setStartupTimeout(10000L); + brokerAdmin = BrokerTestUtils.getRabbitBrokerAdmin(); brokerAdmin.startNode(); panic.setBrokerAdmin(brokerAdmin); } @@ -113,10 +107,16 @@ public class RabbitBrokerAdminIntegrationTests { } + @Test + public void testGetEmptyQueues() throws Exception { + List queues = brokerAdmin.getQueues(); + assertEquals(0, queues.size()); + } + @Test public void testGetQueues() throws Exception { SingleConnectionFactory connectionFactory = new SingleConnectionFactory(); - connectionFactory.setPort(PORT); + connectionFactory.setPort(BrokerTestUtils.getAdminPort()); Queue queue = new RabbitAdmin(connectionFactory).declareQueue(); assertEquals("/", connectionFactory.getVirtualHost()); List queues = brokerAdmin.getQueues(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminLifecycleIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminLifecycleIntegrationTests.java index d10c5cce..e5915da0 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminLifecycleIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminLifecycleIntegrationTests.java @@ -26,6 +26,7 @@ import org.apache.log4j.Level; import org.junit.Before; import org.junit.Rule; import org.junit.Test; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; import org.springframework.erlang.OtpException; @@ -37,7 +38,7 @@ public class RabbitBrokerAdminLifecycleIntegrationTests { private static Log logger = LogFactory.getLog(RabbitBrokerAdminLifecycleIntegrationTests.class); - private static final String NODE_NAME = "spring@localhost"; + private static final String NODE_NAME = "spring@localhost"; @Rule public Log4jLevelAdjuster logLevel = new Log4jLevelAdjuster(Level.INFO, RabbitBrokerAdmin.class); @@ -51,12 +52,7 @@ public class RabbitBrokerAdminLifecycleIntegrationTests { public void testStartNode() throws Exception { // Set up broker admin for non-root user - final RabbitBrokerAdmin brokerAdmin = new RabbitBrokerAdmin(NODE_NAME, 15672); - brokerAdmin.setRabbitLogBaseDirectory("target/rabbitmq/log"); - brokerAdmin.setRabbitMnesiaBaseDirectory("target/rabbitmq/mnesia"); - - brokerAdmin.setStartupTimeout(10000L); - + final RabbitBrokerAdmin brokerAdmin = BrokerTestUtils.getRabbitBrokerAdmin(NODE_NAME); RabbitStatus status = brokerAdmin.getStatus(); try { // Stop it if it is already running @@ -93,18 +89,14 @@ public class RabbitBrokerAdminLifecycleIntegrationTests { public void testStopAndStartBroker() throws Exception { // Set up broker admin for non-root user - final RabbitBrokerAdmin brokerAdmin = new RabbitBrokerAdmin(NODE_NAME, 15672); - brokerAdmin.setRabbitLogBaseDirectory("target/rabbitmq/log"); - brokerAdmin.setRabbitMnesiaBaseDirectory("target/rabbitmq/mnesia"); - - brokerAdmin.setStartupTimeout(10000L); + final RabbitBrokerAdmin brokerAdmin = BrokerTestUtils.getRabbitBrokerAdmin(NODE_NAME); RabbitStatus status = brokerAdmin.getStatus(); status = brokerAdmin.getStatus(); if (!status.isRunning()) { brokerAdmin.startBrokerApplication(); } - + brokerAdmin.stopBrokerApplication(); status = brokerAdmin.getStatus(); @@ -127,10 +119,10 @@ public class RabbitBrokerAdminLifecycleIntegrationTests { } } - /** - * Asserts that the named-node is running. - * @param status - */ + /** + * Asserts that the named-node is running. + * @param status + */ private void assertBrokerAppRunning(RabbitStatus status) { assertEquals(1, status.getRunningNodes().size()); assertTrue(status.getRunningNodes().get(0).getName().contains(NODE_NAME)); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerBrokerInterruptionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerBrokerInterruptionIntegrationTests.java new file mode 100644 index 00000000..5ade0354 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerBrokerInterruptionIntegrationTests.java @@ -0,0 +1,165 @@ +package org.springframework.amqp.rabbit.listener; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.io.FileUtils; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.admin.QueueInfo; +import org.springframework.amqp.rabbit.admin.RabbitBrokerAdmin; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; +import org.springframework.amqp.rabbit.test.BrokerPanic; +import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; + +import com.rabbitmq.client.Channel; + +public class MessageListenerBrokerInterruptionIntegrationTests { + + private static Log logger = LogFactory.getLog(MessageListenerBrokerInterruptionIntegrationTests.class); + + private Queue queue = new Queue("test.queue"); + + private int concurrentConsumers = 1; + + private int messageCount = 10; + + private int txSize = 1; + + private boolean transactional = false; + + private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO; + + private SimpleMessageListenerContainer container; + + /* + * Ensure broker dies if a test fails (otherwise the erl process might have to be killed manually) + */ + @Rule + public static BrokerPanic panic = new BrokerPanic(); + + @Rule + public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue); + + private ConnectionFactory connectionFactory; + + private RabbitBrokerAdmin brokerAdmin; + + public MessageListenerBrokerInterruptionIntegrationTests() throws Exception { + FileUtils.deleteDirectory(new File("target/rabbitmq")); + // Ensure queue is durable, or it won't survive the broker restart + queue.setDurable(true); + brokerIsRunning.setPort(BrokerTestUtils.getAdminPort()); + logger.debug("Setting up broker"); + brokerAdmin = BrokerTestUtils.getRabbitBrokerAdmin(); + panic.setBrokerAdmin(brokerAdmin); + brokerAdmin.startNode(); + } + + @Before + public void createConnectionFactory() { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + connectionFactory.setChannelCacheSize(concurrentConsumers); + connectionFactory.setPort(BrokerTestUtils.getAdminPort()); + this.connectionFactory = connectionFactory; + } + + @After + public void clear() throws Exception { + // Wait for broker communication to finish before trying to stop container + Thread.sleep(300L); + logger.debug("Shutting down at end of test"); + if (container != null) { + container.shutdown(); + } + brokerAdmin.stopNode(); + // Remove all trace of the durable queue... + FileUtils.deleteDirectory(new File("target/rabbitmq")); + } + + @Test + public void testListenerRecoversFromClosedConnection() throws Exception { + + List queues = brokerAdmin.getQueues(); + logger.info("Queues: " + queues); + assertEquals(1, queues.size()); + assertTrue(queues.get(0).isDurable()); + + RabbitTemplate template = new RabbitTemplate(connectionFactory); + + CountDownLatch latch = new CountDownLatch(messageCount); + container = createContainer(new VanillaListener(latch), connectionFactory); + for (int i = 0; i < messageCount; i++) { + template.convertAndSend(queue.getName(), i + "foo"); + } + + brokerAdmin.stopBrokerApplication(); + boolean waited = latch.await(500, TimeUnit.MILLISECONDS); + assertFalse("Did not time out waiting for message", waited); + + container.stop(); + assertEquals(0, container.getActiveConsumerCount()); + + brokerAdmin.startBrokerApplication(); + queues = brokerAdmin.getQueues(); + logger.info("Queues: " + queues); + container.start(); + assertEquals(concurrentConsumers, container.getActiveConsumerCount()); + + int timeout = Math.min(4 + messageCount / (4 * concurrentConsumers), 30); + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + waited = latch.await(timeout, TimeUnit.SECONDS); + assertTrue("Timed out waiting for message", waited); + + assertNull(template.receiveAndConvert(queue.getName())); + + } + + private SimpleMessageListenerContainer createContainer(Object listener, ConnectionFactory connectionFactory) { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); + container.setMessageListener(new MessageListenerAdapter(listener)); + container.setQueueName(queue.getName()); + container.setTxSize(txSize); + container.setPrefetchCount(txSize); + container.setConcurrentConsumers(concurrentConsumers); + container.setChannelTransacted(transactional); + container.setAcknowledgeMode(acknowledgeMode); + container.afterPropertiesSet(); + container.start(); + return container; + } + + public static class VanillaListener implements ChannelAwareMessageListener { + + private final CountDownLatch latch; + + public VanillaListener(CountDownLatch latch) { + this.latch = latch; + } + + public void onMessage(Message message, Channel channel) throws Exception { + String value = new String(message.getBody()); + logger.debug("Receiving: " + value); + latch.countDown(); + } + } +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java index 8dec2e9b..4bb244d7 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java @@ -158,8 +158,10 @@ public class MessageListenerContainerLifecycleIntegrationTests { assertFalse("Expected not to receive all messages before stop", waited); } + assertEquals(concurrentConsumers, container.getActiveConsumerCount()); container.stop(); Thread.sleep(500L); + assertEquals(0, container.getActiveConsumerCount()); if (!transactional) { int messagesReceivedAfterStop = listener.getCount(); @@ -181,6 +183,7 @@ public class MessageListenerContainerLifecycleIntegrationTests { int messagesReceivedBeforeStart = listener.getCount(); container.start(); + assertEquals(concurrentConsumers, container.getActiveConsumerCount()); int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); @@ -195,11 +198,14 @@ public class MessageListenerContainerLifecycleIntegrationTests { assertNull("Messages still available", template.receive(queue.getName())); } + assertEquals(concurrentConsumers, container.getActiveConsumerCount()); + } finally { // Wait for broker communication to finish before trying to stop // container Thread.sleep(300L); container.shutdown(); + assertEquals(0, container.getActiveConsumerCount()); } assertNull(template.receiveAndConvert(queue.getName())); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerCachingConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java similarity index 85% rename from spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerCachingConnectionIntegrationTests.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java index 5d39b0f4..70a8d442 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerCachingConnectionIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java @@ -1,5 +1,6 @@ package org.springframework.amqp.rabbit.listener; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; @@ -29,9 +30,9 @@ import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; import com.rabbitmq.client.Channel; -public class MessageListenerCachingConnectionIntegrationTests { +public class MessageListenerRecoveryCachingConnectionIntegrationTests { - private static Log logger = LogFactory.getLog(MessageListenerCachingConnectionIntegrationTests.class); + private static Log logger = LogFactory.getLog(MessageListenerRecoveryCachingConnectionIntegrationTests.class); private Queue queue = new Queue("test.queue"); @@ -113,6 +114,32 @@ public class MessageListenerCachingConnectionIntegrationTests { } + @Test + public void testListenerRecoversFromClosedChannelAndStop() throws Exception { + + RabbitTemplate template = new RabbitTemplate(createConnectionFactory()); + + CountDownLatch latch = new CountDownLatch(messageCount); + container = createContainer(new AbortChannelListener(latch), createConnectionFactory()); + assertEquals(concurrentConsumers, container.getActiveConsumerCount()); + + for (int i = 0; i < messageCount; i++) { + template.convertAndSend(queue.getName(), i + "foo"); + } + + int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + boolean waited = latch.await(timeout, TimeUnit.SECONDS); + assertTrue("Timed out waiting for message", waited); + + assertNull(template.receiveAndConvert(queue.getName())); + + assertEquals(concurrentConsumers, container.getActiveConsumerCount()); + container.stop(); + assertEquals(0, container.getActiveConsumerCount()); + + } + @Test public void testListenerRecoversFromClosedConnection() throws Exception { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerSingleConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoverySingleConnectionIntegrationTests.java similarity index 76% rename from spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerSingleConnectionIntegrationTests.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoverySingleConnectionIntegrationTests.java index 45d4fb62..c6a97716 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerSingleConnectionIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoverySingleConnectionIntegrationTests.java @@ -4,7 +4,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; import org.springframework.amqp.rabbit.test.BrokerTestUtils; -public class MessageListenerSingleConnectionIntegrationTests extends MessageListenerCachingConnectionIntegrationTests { +public class MessageListenerRecoverySingleConnectionIntegrationTests extends MessageListenerRecoveryCachingConnectionIntegrationTests { protected ConnectionFactory createConnectionFactory() { SingleConnectionFactory connectionFactory = new SingleConnectionFactory(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java index b3033566..c7172a72 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerRunning.java @@ -10,6 +10,7 @@ import org.junit.runners.model.Statement; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.util.StringUtils; /** *

@@ -56,6 +57,10 @@ public class BrokerRunning extends TestWatchman { private Queue queue; + private int port = BrokerTestUtils.DEFAULT_PORT; + + private String hostName = null; + /** * Ensure the broker is running and has an empty queue with the specified name in the default exchange. * @@ -101,6 +106,20 @@ public class BrokerRunning extends TestWatchman { private BrokerRunning(boolean assumeOnline) { this(assumeOnline, new Queue(DEFAULT_QUEUE_NAME)); } + + /** + * @param port the port to set + */ + public void setPort(int port) { + this.port = port; + } + + /** + * @param hostName the hostName to set + */ + public void setHostName(String hostName) { + this.hostName = hostName; + } @Override public Statement apply(Statement base, FrameworkMethod method, Object target) { @@ -115,6 +134,10 @@ public class BrokerRunning extends TestWatchman { try { CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + connectionFactory.setPort(port); + if (StringUtils.hasText(hostName)) { + connectionFactory.setHost(hostName); + } RabbitAdmin admin = new RabbitAdmin(connectionFactory); String queueName = queue.getName(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java index 0d617daa..06d192ee 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java @@ -12,6 +12,9 @@ */ package org.springframework.amqp.rabbit.test; +import org.springframework.amqp.rabbit.admin.RabbitBrokerAdmin; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; + /** * Global convenience class for all integration tests, carrying constants and other utilities for broker set up. * @@ -19,13 +22,70 @@ package org.springframework.amqp.rabbit.test; * */ public class BrokerTestUtils { - + public static final int DEFAULT_PORT = 5672; public static final int TRACER_PORT = 5673; + + public static final String ADMIN_NODE_NAME = "spring@localhost"; + + /** + * The port that the broker is listening on (e.g. as input for a {@link ConnectionFactory}). + * + * @return a port number + */ public static int getPort() { return DEFAULT_PORT; } + /** + * An alternative port number than can safely be used to stop and start a broker, even when one is already running + * on the standard port as a privileged user. Useful for tests involving {@link RabbitBrokerAdmin} on UN*X. + * + * @return a port number + */ + public static int getAdminPort() { + return 15672; + } + + /** + * Convenience factory for a {@link RabbitBrokerAdmin} instance that will usually start and stop cleanly on all + * systems. + * + * @return a {@link RabbitBrokerAdmin} instance + */ + public static RabbitBrokerAdmin getRabbitBrokerAdmin() { + return getRabbitBrokerAdmin(ADMIN_NODE_NAME, getAdminPort()); + } + + /** + * Convenience factory for a {@link RabbitBrokerAdmin} instance that will usually start and stop cleanly on all + * systems. + * + * @param nodeName the name of the node + * + * @return a {@link RabbitBrokerAdmin} instance + */ + public static RabbitBrokerAdmin getRabbitBrokerAdmin(String nodeName) { + return getRabbitBrokerAdmin(nodeName, getAdminPort()); + } + + /** + * Convenience factory for a {@link RabbitBrokerAdmin} instance that will usually start and stop cleanly on all + * systems. + * + * @param nodeName the name of the node + * @param port the port to listen on + * + * @return a {@link RabbitBrokerAdmin} instance + */ + public static RabbitBrokerAdmin getRabbitBrokerAdmin(String nodeName, int port) { + RabbitBrokerAdmin brokerAdmin = new RabbitBrokerAdmin(nodeName, port); + brokerAdmin.setRabbitLogBaseDirectory("target/rabbitmq/log"); + brokerAdmin.setRabbitMnesiaBaseDirectory("target/rabbitmq/mnesia"); + brokerAdmin.setStartupTimeout(10000L); + return brokerAdmin; + } + }