AMQP-125: add custom lock manager and tests

- Add test for transactional send in listener
- Add ActiveObjectCounter and use it in SMLC and consumer

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2002-2010 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
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package org.springframework.amqp.rabbit.listener;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class ActiveObjectCounter<T> {
+
+	private final ConcurrentMap<T, CountDownLatch> locks = new ConcurrentHashMap<T, CountDownLatch>();
+
+	public void add(T object) {
+		CountDownLatch lock = new CountDownLatch(1);
+		locks.putIfAbsent(object, lock);
+	}
+
+	public void release(T object) {
+		CountDownLatch remove = locks.remove(object);
+		if (remove != null) {
+			remove.countDown();
+		}
+	}
+
+	public boolean await(Long timeout, TimeUnit timeUnit) throws InterruptedException {
+		long t0 = System.currentTimeMillis();
+		long t1 = t0 + TimeUnit.MILLISECONDS.convert(timeout, timeUnit);
+		while (System.currentTimeMillis() <= t1) {
+			if (locks.isEmpty()) {
+				return true;
+			}
+			Collection<T> objects = new HashSet<T>(locks.keySet());
+			for (T object : objects) {
+				CountDownLatch lock = locks.get(object);
+				if (lock==null) {
+					continue;
+				}
+				t0 = System.currentTimeMillis();
+				if (lock.await(t1 - t0, TimeUnit.MILLISECONDS)) {
+					locks.remove(object);
+				}
+			}
+		}
+		return false;
+	}
+
+	public int getCount() {
+		return locks.size();
+	}
+
+	public void reset() {
+		locks.clear();
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java

@@ -57,13 +57,16 @@ public class BlockingQueueConsumer {
 
 	private final ConnectionFactory connectionFactory;
 
+	private final ActiveObjectCounter<BlockingQueueConsumer> stopped;
+
 	/**
 	 * Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker
 	 * until it is started.
 	 */
-	public BlockingQueueConsumer(ConnectionFactory connectionFactory, AcknowledgeMode acknowledgeMode,
-			boolean transactional, int prefetchCount, String... queues) {
+	public BlockingQueueConsumer(ConnectionFactory connectionFactory, ActiveObjectCounter<BlockingQueueConsumer> stopped,
+			AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, String... queues) {
 		this.connectionFactory = connectionFactory;
+		this.stopped = stopped;
 		this.acknowledgeMode = acknowledgeMode;
 		this.transactional = transactional;
 		this.prefetchCount = prefetchCount;
@@ -145,6 +148,7 @@ public class BlockingQueueConsumer {
 		this.channel = ConnectionFactoryUtils.getTransactionalResourceHolder(connectionFactory, transactional)
 				.getChannel();
 		this.consumer = new InternalConsumer(channel);
+		this.stopped.add(this);
 		try {
 			// Set basicQos before calling basicConsume (it is ignored if we are not transactional and the broker will
 			// send blocks of 100 messages)
@@ -170,10 +174,10 @@ public class BlockingQueueConsumer {
 
 	public void stop() {
 		cancelled.set(true);
-		logger.debug("Closing Rabbit Channel: " + channel);
 		if (consumer != null && consumer.getChannel() != null && consumer.getConsumerTag() != null) {
 			RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), transactional);
 		}
+		logger.debug("Closing Rabbit Channel: " + channel);
 		// This one never throws exceptions...
 		RabbitUtils.closeChannel(channel);
 	}
@@ -193,6 +197,15 @@ public class BlockingQueueConsumer {
 			// TODO: interrupt?
 		}
 
+		@Override
+		public void handleCancelOk(String consumerTag) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Received cancellation notice for " + BlockingQueueConsumer.this);
+			}
+			// Signal to the container that we have been cancelled
+			stopped.release(BlockingQueueConsumer.this);
+		}
+
 		@Override
 		public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
 				throws IOException {
@@ -201,8 +214,10 @@ public class BlockingQueueConsumer {
 					return;
 				}
 			}
-			// TODO: do we want to pass on 'consumerTag'?
-			logger.debug("Storing delivery for " + BlockingQueueConsumer.this);
+			if (logger.isDebugEnabled()) {
+				// TODO: do we want to pass on 'consumerTag'?
+				logger.debug("Storing delivery for " + BlockingQueueConsumer.this);
+			}
 			try {
 				// TODO: If transactional we could use a bounded queue and offer() here with a timeout
 				// in which case if it fails we could nack the message and have it requeued.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java

@@ -33,6 +33,8 @@ 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.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.support.MetricType;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
 import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
@@ -82,7 +84,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 	private TransactionAttribute transactionAttribute = new DefaultTransactionAttribute();
 
-	private CountDownLatch cancellationLock;
+	private ActiveObjectCounter<BlockingQueueConsumer> cancellationLock = new ActiveObjectCounter<BlockingQueueConsumer>();
 
 	public static interface ContainerDelegate {
 		boolean receiveAndExecute(BlockingQueueConsumer consumer) throws Throwable;
@@ -276,8 +278,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		initializeProxy();
 	}
 
+	@ManagedMetric(metricType = MetricType.GAUGE)
 	public int getActiveConsumerCount() {
-		return (int) cancellationLock.getCount();
+		return cancellationLock.getCount();
 	}
 
 	/**
@@ -288,17 +291,15 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	 */
 	protected void doStart() throws Exception {
 		super.doStart();
-		initializeConsumers();
 		synchronized (this.consumersMonitor) {
+			initializeConsumers();
 			if (this.consumers == null) {
 				logger.info("Consumers were initialized and then cleared (presumably the container was stopped concurrently)");
 				return;
 			}
-			cancellationLock = new CountDownLatch(this.consumers.size());
 			Set<AsyncMessageProcessingConsumer> processors = new HashSet<AsyncMessageProcessingConsumer>();
 			for (BlockingQueueConsumer consumer : this.consumers) {
-				AsyncMessageProcessingConsumer processor = new AsyncMessageProcessingConsumer(consumer,
-						cancellationLock);
+				AsyncMessageProcessingConsumer processor = new AsyncMessageProcessingConsumer(consumer);
 				processors.add(processor);
 				this.taskExecutor.execute(processor);
 			}
@@ -345,6 +346,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	protected void initializeConsumers() {
 		synchronized (this.consumersMonitor) {
 			if (this.consumers == null) {
+				cancellationLock.reset();
 				this.consumers = new HashSet<BlockingQueueConsumer>(this.concurrentConsumers);
 				for (int i = 0; i < this.concurrentConsumers; i++) {
 					BlockingQueueConsumer consumer = createBlockingQueueConsumer();
@@ -361,8 +363,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	protected BlockingQueueConsumer createBlockingQueueConsumer() {
 		BlockingQueueConsumer consumer;
 		String[] queues = getRequiredQueueNames();
-		consumer = new BlockingQueueConsumer(getConnectionFactory(), getAcknowledgeMode(), isChannelTransacted(),
-				prefetchCount, queues);
+		consumer = new BlockingQueueConsumer(getConnectionFactory(), cancellationLock, getAcknowledgeMode(),
+				isChannelTransacted(), prefetchCount, queues);
 		return consumer;
 	}
 
@@ -372,6 +374,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 				try {
 					// Need to recycle the channel in this consumer
 					consumer.stop();
+					this.cancellationLock.release(consumer);
 					this.consumers.remove(consumer);
 					consumer = createBlockingQueueConsumer();
 					this.consumers.add(consumer);
@@ -380,12 +383,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 					// 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.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, cancellationLock));
+				this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer));
 			}
 		}
 	}
@@ -426,17 +428,14 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 		private final BlockingQueueConsumer consumer;
 
-		private final CountDownLatch latch;
-
 		private final CountDownLatch start;
 
 		private volatile ListenerStartupFatalException startupException;
 
 		private AtomicBoolean started = new AtomicBoolean(false);
 
-		public AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer, CountDownLatch latch) {
+		public AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) {
 			this.consumer = consumer;
-			this.latch = latch;
 			this.start = new CountDownLatch(1);
 		}
 
@@ -497,8 +496,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 				// Fatal, but no point re-throwing, so just abort.
 				aborted = true;
 			} catch (Throwable t) {
-				if (logger.isInfoEnabled()) {
-					logger.info(
+				if (logger.isDebugEnabled()) {
+					logger.warn(
 							"Consumer raised exception, processing can restart if the connection factory supports it",
 							t);
 				} else {
@@ -511,17 +510,18 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 			if (!isActive() || aborted) {
 				logger.debug("Cancelling " + consumer);
-				latch.countDown();
 				try {
 					consumer.stop();
 				} catch (AmqpException e) {
-					logger.info("Could not stop message consumer on shutdown", e);
+					logger.info("Could not cancel message consumer", e);
+					// TODO: should we release the cancellationLock here?
 				}
 				if (aborted) {
 					stop();
 				}
 			} else {
 				logger.info("Restarting " + consumer);
+				// cancellationLock is not decremented on restart
 				restart(consumer);
 			}
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java

@@ -14,6 +14,7 @@ import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.RabbitAccessor;
+import org.springframework.amqp.rabbit.listener.ActiveObjectCounter;
 import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
@@ -232,7 +233,7 @@ public class RabbitBindingIntegrationTests {
 	}
 
 	private BlockingQueueConsumer createConsumer(RabbitAccessor accessor) {
-		BlockingQueueConsumer consumer = new BlockingQueueConsumer(accessor.getConnectionFactory(), AcknowledgeMode.AUTO, true, 1, queue.getName());
+		BlockingQueueConsumer consumer = new BlockingQueueConsumer(accessor.getConnectionFactory(), new ActiveObjectCounter<BlockingQueueConsumer>(), AcknowledgeMode.AUTO, true, 1, queue.getName());
 		consumer.start();
 		return consumer;
 	}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounterTests.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2002-2010 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.amqp.rabbit.listener;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class ActiveObjectCounterTests {
+	
+	private ActiveObjectCounter<Object> counter = new ActiveObjectCounter<Object>();
+	
+	@Test
+	public void testActiveCount() throws Exception {
+		final Object object1 = new Object();
+		final Object object2 = new Object();
+		counter.add(object1);
+		counter.add(object2);
+		assertEquals(2, counter.getCount());
+		counter.release(object2);
+		assertEquals(1, counter.getCount());
+		counter.release(object1);
+		counter.release(object1);
+		assertEquals(0, counter.getCount());
+	}
+
+	@Test
+	public void testWaitForLocks() throws Exception {
+		final Object object1 = new Object();
+		final Object object2 = new Object();
+		counter.add(object1);
+		counter.add(object2);
+		Future<Boolean> future = Executors.newSingleThreadExecutor().submit(new Callable<Boolean>() {
+			public Boolean call() throws Exception {
+				counter.release(object1);
+				counter.release(object2);
+				counter.release(object2);
+				return true;
+			}
+		});
+		assertEquals(true, counter.await(1000L, TimeUnit.MILLISECONDS));
+		assertEquals(true, future.get());
+	}
+
+	@Test
+	public void testTimeoutWaitForLocks() throws Exception {
+		final Object object1 = new Object();
+		counter.add(object1);
+		assertEquals(false, counter.await(200L, TimeUnit.MILLISECONDS));
+	}
+
+}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java

@@ -183,12 +183,12 @@ 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)");
 			waited = latch.await(timeout, TimeUnit.SECONDS);
 			logger.info("All messages received after start: " + waited);
+			assertEquals(concurrentConsumers, container.getActiveConsumerCount());
 			if (transactional) {
 				assertTrue("Timed out waiting for message", waited);
 			} else {
@@ -203,7 +203,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 		} finally {
 			// Wait for broker communication to finish before trying to stop
 			// container
-			Thread.sleep(300L);
+			Thread.sleep(500L);
 			container.shutdown();
 			assertEquals(0, container.getActiveConsumerCount());
 		}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java

@@ -23,6 +23,7 @@ import org.springframework.amqp.rabbit.connection.Connection;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionProxy;
 import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.MessageListenerBrokerInterruptionIntegrationTests.VanillaListener;
 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
@@ -74,6 +75,63 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 		}
 	}
 
+	@Test
+	public void testListenerSendsMessageAndThenCommit() throws Exception {
+
+		ConnectionFactory connectionFactory = createConnectionFactory();
+		RabbitTemplate template = new RabbitTemplate(connectionFactory);
+		Queue sendQueue = new Queue("test.send");
+		new RabbitAdmin(connectionFactory).declareQueue(sendQueue);
+
+		acknowledgeMode = AcknowledgeMode.AUTO;
+		transactional = true;
+
+		CountDownLatch latch = new CountDownLatch(1);
+		container = createContainer(queue.getName(), new ChannelSenderListener(sendQueue.getName(), latch, false),
+				connectionFactory);
+		template.convertAndSend(queue.getName(), "foo");
+
+		int timeout = getTimeout();
+		logger.debug("Waiting for messages with timeout = " + timeout + " (s)");
+		boolean waited = latch.await(timeout, TimeUnit.SECONDS);
+		assertTrue("Timed out waiting for message", waited);
+
+		// All messages committed
+		assertEquals("bar", new String((byte[])template.receiveAndConvert(sendQueue.getName())));
+		assertNull(template.receiveAndConvert(queue.getName()));
+
+	}
+
+	@Test
+	public void testListenerSendsMessageAndThenRollback() throws Exception {
+
+		ConnectionFactory connectionFactory = createConnectionFactory();
+		RabbitTemplate template = new RabbitTemplate(connectionFactory);
+		Queue sendQueue = new Queue("test.send");
+		new RabbitAdmin(connectionFactory).declareQueue(sendQueue);
+
+		acknowledgeMode = AcknowledgeMode.AUTO;
+		transactional = true;
+
+		CountDownLatch latch = new CountDownLatch(1);
+		container = createContainer(queue.getName(), new ChannelSenderListener(sendQueue.getName(), latch, true),
+				connectionFactory);
+		template.convertAndSend(queue.getName(), "foo");
+
+		int timeout = getTimeout();
+		logger.debug("Waiting for messages with timeout = " + timeout + " (s)");
+		boolean waited = latch.await(timeout, TimeUnit.SECONDS);
+		assertTrue("Timed out waiting for message", waited);
+		
+		container.stop();
+		
+		// Foo message is redelivered
+		assertEquals("foo", template.receiveAndConvert(queue.getName()));
+		// Sending of bar message is also rolled back
+		assertNull(template.receiveAndConvert(sendQueue.getName()));
+
+	}
+
 	@Test
 	public void testListenerRecoversFromBogusDoubleAck() throws Exception {
 
@@ -87,7 +145,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 			template.convertAndSend(queue.getName(), i + "foo");
 		}
 
-		int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30);
+		int timeout = getTimeout();
 		logger.debug("Waiting for messages with timeout = " + timeout + " (s)");
 		boolean waited = latch.await(timeout, TimeUnit.SECONDS);
 		assertTrue("Timed out waiting for message", waited);
@@ -107,7 +165,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 			template.convertAndSend(queue.getName(), i + "foo");
 		}
 
-		int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30);
+		int timeout = getTimeout();
 		logger.debug("Waiting for messages with timeout = " + timeout + " (s)");
 		boolean waited = latch.await(timeout, TimeUnit.SECONDS);
 		assertTrue("Timed out waiting for message", waited);
@@ -123,17 +181,18 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 		CountDownLatch latch = new CountDownLatch(messageCount);
 		container = createContainer(queue.getName(), new AbortChannelListener(latch), createConnectionFactory());
+		Thread.sleep(500L);
 		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);
+		int timeout = getTimeout();
 		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());
@@ -149,8 +208,9 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 
 		CountDownLatch latch = new CountDownLatch(messageCount);
 		ConnectionFactory connectionFactory = createConnectionFactory();
-		container = createContainer(queue.getName(), new CloseConnectionListener((ConnectionProxy) connectionFactory.createConnection(),
-						latch), connectionFactory);
+		container = createContainer(queue.getName(),
+				new CloseConnectionListener((ConnectionProxy) connectionFactory.createConnection(), latch),
+				connectionFactory);
 		for (int i = 0; i < messageCount; i++) {
 			template.convertAndSend(queue.getName(), i + "foo");
 		}
@@ -178,7 +238,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 			template.convertAndSend(queue.getName(), i + "foo");
 		}
 
-		int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30);
+		int timeout = getTimeout();
 		logger.debug("Waiting for messages with timeout = " + timeout + " (s)");
 		boolean waited = latch.await(timeout, TimeUnit.SECONDS);
 		assertTrue("Timed out waiting for message", waited);
@@ -187,8 +247,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 		assertNull(template.receiveAndConvert(queue.getName()));
 
 	}
-
-	@Test(expected=AmqpIllegalStateException.class)
+	@Test(expected = AmqpIllegalStateException.class)
 	public void testListenerDoesNotRecoverFromMissingQueue() throws Exception {
 		// TODO: with only 1 this test tends to fail
 		concurrentConsumers = 3;
@@ -196,7 +255,12 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 		container = createContainer("nonexistent", new VanillaListener(latch), createConnectionFactory());
 	}
 
-	private SimpleMessageListenerContainer createContainer(String queueName, Object listener, ConnectionFactory connectionFactory) {
+	private int getTimeout() {
+		return Math.min(1 + messageCount / (4 * concurrentConsumers), 30);
+	}
+	
+	private SimpleMessageListenerContainer createContainer(String queueName, Object listener,
+			ConnectionFactory connectionFactory) {
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
 		container.setMessageListener(new MessageListenerAdapter(listener));
 		container.setQueueNames(queueName);
@@ -235,6 +299,36 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 		}
 	}
 
+	public static class ChannelSenderListener implements ChannelAwareMessageListener {
+
+		private final CountDownLatch latch;
+
+		private final boolean fail;
+
+		private final String queueName;
+
+		public ChannelSenderListener(String queueName, CountDownLatch latch, boolean fail) {
+			this.queueName = queueName;
+			this.latch = latch;
+			this.fail = fail;
+		}
+
+		public void onMessage(Message message, Channel channel) throws Exception {
+			String value = new String(message.getBody());
+			try {
+				logger.debug("Received: " + value + " Sending: bar");
+				channel.basicPublish("", queueName, null, "bar".getBytes());
+				if (fail) {
+					logger.debug("Failing (planned)");
+					// intentional error (causes exception on connection thread):
+					throw new RuntimeException("Planned");
+				}
+			} finally {
+				latch.countDown();
+			}
+		}
+	}
+
 	public static class AbortChannelListener implements ChannelAwareMessageListener {
 
 		private AtomicBoolean failed = new AtomicBoolean(false);