AMQP-44: refactor SimpleMessageListenerContainer to use Advice for transactions - opens up possibility to apply retry advice

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();
 	}

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.
-	 * <p>
-	 * This implementation always returns <code>true</code>; the default 'running' state is purely determined by
-	 * {@link #start()} / {@link #stop()}.
-	 * <p>
-	 * 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 <code>false</code>
-	 * if such a restriction prevents listeners from running.
-	 */
-	protected boolean runningAllowed() {
-		return true;
-	}
-
 	// -------------------------------------------------------------------------
 	// Management of a shared Rabbit Connection
 	// -------------------------------------------------------------------------

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?

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<BlockingQueueConsumer> consumers;
+	private Set<BlockingQueueConsumer> 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;
+
+	/**
+	 * <p>
+	 * 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.
+	 * </p>
+	 * <p>
+	 * 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).
+	 * </p>
+	 * 
+	 * @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<Boolean>() {
-							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;
-
-		}
 	}
 
 }

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 <code>null</code>)
 	 */
 	public static void closeChannel(Channel channel) {
-		if (channel != null) {
+		if (channel != null && channel.isOpen()) {
 			try {
 				channel.close();
 			} catch (IOException ex) {

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<QueueInfo> 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<QueueInfo> queues = brokerAdmin.getQueues();

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));

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<QueueInfo> 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();
+		}
+	}
+}

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()));

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 {
 

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();

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;
 
 /**
  * <p>
@@ -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();

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;
+	}
+
 }