AMQP-172: move ChannelListener to base class connection factory

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java

@@ -31,8 +31,11 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 	protected final Log logger = LogFactory.getLog(getClass());
 
 	private final com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory;
+
 	private final CompositeConnectionListener connectionListener = new CompositeConnectionListener();
 
+	private final CompositeChannelListener channelListener = new CompositeChannelListener();
+
 	/**
 	 * Create a new SingleConnectionFactory for the given target ConnectionFactory.
 	 * @param rabbitConnectionFactory the target ConnectionFactory
@@ -79,9 +82,18 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 	 * 
 	 * @return the connection listener
 	 */
-	protected CompositeConnectionListener getConnectionListener() {
+	protected ConnectionListener getConnectionListener() {
 		return connectionListener;
 	}
+	
+	/**
+	 * A composite channel listener to be used by subclasses when creating and closing channels.
+	 * 
+	 * @return the channel listener
+	 */
+	protected ChannelListener getChannelListener() {
+		return channelListener;
+	}
 
 	public void setConnectionListeners(List<? extends ConnectionListener> listeners) {
 		this.connectionListener.setDelegates(listeners);
@@ -91,6 +103,14 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 		this.connectionListener.addDelegate(listener);
 	}
 
+	public void setChannelListeners(List<? extends ChannelListener> listeners) {
+		this.channelListener.setDelegates(listeners);
+	}
+
+	public void addChannelListener(ChannelListener listener) {
+		this.channelListener.addDelegate(listener);
+	}
+
 	final protected Connection createBareConnection() {
 		try {
 			return new SimpleConnection(this.rabbitConnectionFactory.newConnection());

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java

@@ -21,7 +21,6 @@ import java.util.LinkedList;
 import java.util.List;
 
 import org.springframework.amqp.AmqpException;
-import org.springframework.beans.factory.DisposableBean;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -46,7 +45,7 @@ import com.rabbitmq.client.Channel;
  * @author Mark Fisher
  * @author Dave Syer
  */
-public class CachingConnectionFactory extends AbstractConnectionFactory implements DisposableBean {
+public class CachingConnectionFactory extends AbstractConnectionFactory {
 
 	private int channelCacheSize = 1;
 
@@ -61,8 +60,6 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
 	/** Synchronization monitor for the shared Connection */
 	private final Object connectionMonitor = new Object();
 
-	private final CompositeChannelListener channelListener = new CompositeChannelListener();
-
 	/**
 	 * Create a new CachingConnectionFactory initializing the hostname to be the value returned from
 	 * InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.
@@ -112,14 +109,6 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
 		super(rabbitConnectionFactory);
 	}
 
-	public void setChannelListeners(List<? extends ChannelListener> listeners) {
-		this.channelListener.setDelegates(listeners);
-	}
-
-	public void addChannelListener(ChannelListener listener) {
-		this.channelListener.addDelegate(listener);
-	}
-
 	public void setChannelCacheSize(int sessionCacheSize) {
 		Assert.isTrue(sessionCacheSize >= 1, "Channel cache size must be 1 or higher");
 		this.channelCacheSize = sessionCacheSize;
@@ -129,6 +118,22 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
 		return this.channelCacheSize;
 	}
 
+	public void setConnectionListeners(List<? extends ConnectionListener> listeners) {
+		super.setConnectionListeners(listeners);
+		// If the connection is already alive we assume that the new listeners want to be notified
+		if (this.connection != null) {
+			this.getConnectionListener().onCreate(this.connection);
+		}
+	}
+
+	public void addConnectionListener(ConnectionListener listener) {
+		super.addConnectionListener(listener);
+		// If the connection is already alive we assume that the new listener wants to be notified
+		if (this.connection != null) {
+			listener.onCreate(this.connection);
+		}
+	}
+
 	private Channel getChannel(boolean transactional) {
 		LinkedList<ChannelProxy> channelList = transactional ? this.cachedChannelsTransactional
 				: this.cachedChannelsNonTransactional;
@@ -153,14 +158,14 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
 		if (logger.isDebugEnabled()) {
 			logger.debug("Creating cached Rabbit Channel from " + targetChannel);
 		}
-		channelListener.onCreate(targetChannel, transactional);
+		getChannelListener().onCreate(targetChannel, transactional);
 		return (ChannelProxy) Proxy.newProxyInstance(ChannelProxy.class.getClassLoader(),
 				new Class[] { ChannelProxy.class }, new CachedChannelInvocationHandler(targetChannel, channelList,
 						transactional));
 	}
 
 	private Channel createBareChannel(boolean transactional) {
-		if (this.connection==null || !this.connection.isOpen()) {
+		if (this.connection == null || !this.connection.isOpen()) {
 			this.connection = null;
 			// Use createConnection here not doCreateConnection so that the old one is properly disposed
 			createConnection();
@@ -280,9 +285,12 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
 				return this.target;
 			}
 			try {
-				synchronized (targetMonitor) {
-					if (this.target == null) {
-						this.target = createBareChannel(transactional);
+				if (this.target == null || !this.target.isOpen()) {
+					this.target = null;
+					synchronized (targetMonitor) {
+						if (this.target == null) {
+							this.target = createBareChannel(transactional);
+						}
 					}
 				}
 				return method.invoke(this.target, args);
@@ -306,9 +314,9 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
 		 * @param proxy the channel to close
 		 */
 		private void logicalClose(ChannelProxy proxy) throws Exception {
-			if (this.target!=null && !this.target.isOpen()) {
+			if (this.target != null && !this.target.isOpen()) {
 				synchronized (targetMonitor) {
-					if (this.target!=null && !this.target.isOpen()) {
+					if (this.target != null && !this.target.isOpen()) {
 						this.target = null;
 						return;
 					}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeChannelListener.java

@@ -28,9 +28,9 @@ public class CompositeChannelListener implements ChannelListener {
 
 	private List<ChannelListener> delegates = new ArrayList<ChannelListener>();
 
-	public void onCreate(Channel connection, boolean transactional) {
+	public void onCreate(Channel channel, boolean transactional) {
 		for (ChannelListener delegate : delegates) {
-			delegate.onCreate(connection, transactional);
+			delegate.onCreate(channel, transactional);
 		}	
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java

@@ -165,6 +165,7 @@ public class SingleConnectionFactory extends AbstractConnectionFactory {
 				}
 			}
 			Channel channel = target.createChannel(transactional);
+			getChannelListener().onCreate(channel, transactional);
 			return channel;
 		}
 

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

@@ -1,17 +1,14 @@
 /*
  * Copyright 2002-2011 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.
+ * 
+ * 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;
@@ -87,7 +84,8 @@ public class BlockingQueueConsumer {
 	 * 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, MessagePropertiesConverter messagePropertiesConverter,
+	public BlockingQueueConsumer(ConnectionFactory connectionFactory,
+			MessagePropertiesConverter messagePropertiesConverter,
 			ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter, AcknowledgeMode acknowledgeMode,
 			boolean transactional, int prefetchCount, String... queues) {
 		this.connectionFactory = connectionFactory;
@@ -172,15 +170,20 @@ public class BlockingQueueConsumer {
 	}
 
 	public void start() throws AmqpException {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Starting consumer " + this);
+		}
 		this.channel = ConnectionFactoryUtils.getTransactionalResourceHolder(connectionFactory, transactional)
 				.getChannel();
 		this.consumer = new InternalConsumer(channel);
-		this.activeObjectCounter.add(this);
 		this.deliveryTags.clear();
+		this.activeObjectCounter.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)
-			channel.basicQos(prefetchCount);
+			if (!acknowledgeMode.isAutoAck()) {
+				// Set basicQos before calling basicConsume (otherwise if we are not acking the broker
+				// will send blocks of 100 messages)
+				channel.basicQos(prefetchCount);
+			}
 			for (int i = 0; i < queues.length; i++) {
 				channel.queueDeclarePassive(queues[i]);
 			}
@@ -193,7 +196,7 @@ public class BlockingQueueConsumer {
 			for (int i = 0; i < queues.length; i++) {
 				channel.basicConsume(queues[i], acknowledgeMode.isAutoAck(), consumer);
 				if (logger.isDebugEnabled()) {
-					logger.debug("Started " + this);
+					logger.debug("Started on queue '" + queues[i] + "': " + this);
 				}
 			}
 		} catch (IOException e) {
@@ -341,7 +344,7 @@ public class BlockingQueueConsumer {
 		try {
 
 			boolean ackRequired = !acknowledgeMode.isAutoAck() && !acknowledgeMode.isManual();
-			
+
 			if (ackRequired) {
 
 				if (transactional && !locallyTransacted) {
@@ -350,13 +353,12 @@ public class BlockingQueueConsumer {
 					// could be synchronized with an external transaction
 					for (Long deliveryTag : deliveryTags) {
 						ConnectionFactoryUtils.registerDeliveryTag(connectionFactory, channel, deliveryTag);
-					}					
+					}
 
-				
 				} else {
 
 					if (!deliveryTags.isEmpty()) {
-						long deliveryTag = new ArrayList<Long>(deliveryTags).get(deliveryTags.size()-1);
+						long deliveryTag = new ArrayList<Long>(deliveryTags).get(deliveryTags.size() - 1);
 						channel.basicAck(deliveryTag, true);
 					}
 

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

@@ -106,11 +106,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 	private ContainerDelegate proxy = delegate;
 
-
 	/**
 	 * Default constructor for convenient dependency injection via setters.
 	 */
-	public SimpleMessageListenerContainer() { }
+	public SimpleMessageListenerContainer() {
+	}
 
 	/**
 	 * Create a listener container from the connection factory (mandatory).
@@ -121,7 +121,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		this.setConnectionFactory(connectionFactory);
 	}
 
-
 	/**
 	 * <p>
 	 * Public setter for the {@link Advice} to apply to listener executions. If {@link #setTxSize(int) txSize>1} then
@@ -378,7 +377,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 				try {
 					// Need to recycle the channel in this consumer
 					consumer.stop();
-					// Ensure consumer counts are correct (another is not going
+					// Ensure consumer counts are correct (another is going
 					// to start because of the exception, but
 					// we haven't counted down yet)
 					this.cancellationLock.release(consumer);
@@ -444,7 +443,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 			}
 
 		}
-		
+
 		return consumer.commitIfNecessary(isChannelLocallyTransacted(channel));
 
 	}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java

@@ -0,0 +1,135 @@
+package org.springframework.amqp.rabbit.connection;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import com.rabbitmq.client.ConnectionFactory;
+
+/**
+ * @author Dave Syer
+ */
+public abstract class AbstractConnectionFactoryTests {
+
+	protected abstract AbstractConnectionFactory createConnectionFactory(ConnectionFactory mockConnectionFactory);
+	@Test
+	public void testWithListener() throws IOException {
+
+		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
+		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
+
+		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
+
+		final AtomicInteger called = new AtomicInteger(0);
+		AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
+		connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() {
+			public void onCreate(Connection connection) {
+				called.incrementAndGet();
+			}
+			public void onClose(Connection connection) {
+				called.decrementAndGet();
+			}
+		}));
+
+		Connection con = connectionFactory.createConnection();
+		assertEquals(1, called.get());
+
+		con.close();
+		assertEquals(1, called.get());
+		verify(mockConnection, never()).close();
+		
+		connectionFactory.createConnection();
+		assertEquals(1, called.get());
+
+		connectionFactory.destroy();
+		assertEquals(0, called.get());
+		verify(mockConnection, atLeastOnce()).close();
+
+		verify(mockConnectionFactory, times(1)).newConnection();
+
+	}
+	
+	@Test
+	public void testWithListenerRegisteredAfterOpen() throws IOException {
+
+		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
+		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
+
+		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
+
+		final AtomicInteger called = new AtomicInteger(0);
+		AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
+		Connection con = connectionFactory.createConnection();
+		assertEquals(0, called.get());
+
+		connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() {
+			public void onCreate(Connection connection) {
+				called.incrementAndGet();
+			}
+			public void onClose(Connection connection) {
+				called.decrementAndGet();
+			}
+		}));
+		assertEquals(1, called.get());
+
+		con.close();
+		assertEquals(1, called.get());
+		verify(mockConnection, never()).close();
+		
+		connectionFactory.createConnection();
+		assertEquals(1, called.get());
+
+		connectionFactory.destroy();
+		assertEquals(0, called.get());
+		verify(mockConnection, atLeastOnce()).close();
+
+		verify(mockConnectionFactory, times(1)).newConnection();
+
+	}
+
+	@Test
+	public void testCloseInvalidConnection() throws Exception {
+
+		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
+		com.rabbitmq.client.Connection mockConnection1 = mock(com.rabbitmq.client.Connection.class);
+		com.rabbitmq.client.Connection mockConnection2 = mock(com.rabbitmq.client.Connection.class);
+
+		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection1).thenReturn(mockConnection2);
+		// simulate a dead connection
+		when(mockConnection1.isOpen()).thenReturn(false);
+
+		AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
+		
+		Connection connection = connectionFactory.createConnection();
+		// the dead connection should be discarded
+		connection.createChannel(false);
+		verify(mockConnectionFactory, times(2)).newConnection();
+		verify(mockConnection2, times(1)).createChannel();
+		
+		connectionFactory.destroy();
+		verify(mockConnection2, times(1)).close();
+
+	}
+
+	@Test
+	public void testDestroyBeforeUsed() throws Exception {
+
+		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
+
+		AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
+		connectionFactory.destroy();
+
+		verify(mockConnectionFactory, never()).newConnection();
+	}
+
+}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java

@@ -19,15 +19,21 @@ import org.junit.Test;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.GetResponse;
 
 /**
  * @author Mark Pollack
+ * @author Dave Syer
  */
-public class CachingConnectionFactoryTests {
+public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTests {
+
+	@Override
+	protected AbstractConnectionFactory createConnectionFactory(ConnectionFactory connectionFactory) {
+		return new CachingConnectionFactory(connectionFactory);
+	}
 
 	@Test
-
 	public void testWithConnectionFactoryDefaults() throws IOException {
 		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
 		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
@@ -341,60 +347,26 @@ public class CachingConnectionFactoryTests {
 		Assert.assertNotSame(channel3, channel2);
 	}
 
-	@Test
-	public void testWithConnectionListener() throws IOException {
-
-		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
-		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
-
-		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
-
-		final AtomicInteger called = new AtomicInteger(0);
-		CachingConnectionFactory connectionFactory = new CachingConnectionFactory(mockConnectionFactory);
-		connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() {
-			public void onCreate(Connection connection) {
-				called.incrementAndGet();
-			}
-			public void onClose(Connection connection) {
-				called.decrementAndGet();
-			}
-		}));
-
-		Connection con = connectionFactory.createConnection();
-		assertEquals(1, called.get());
-
-		con.close();
-		assertEquals(1, called.get());
-		verify(mockConnection, never()).close();
-		
-		connectionFactory.createConnection();
-		assertEquals(1, called.get());
-
-		connectionFactory.destroy();
-		assertEquals(0, called.get());
-		verify(mockConnection, atLeastOnce()).close();
-
-		verify(mockConnectionFactory).newConnection();
-
-	}
-	
 	@Test
 	public void testWithChannelListener() throws IOException {
 
 		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
 		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
+		Channel mockChannel = mock(Channel.class);
 
 		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
 		when(mockConnection.isOpen()).thenReturn(true);
+		when(mockChannel.isOpen()).thenReturn(true);
+		when(mockConnection.createChannel()).thenReturn(mockChannel);
 
 		final AtomicInteger called = new AtomicInteger(0);
-		CachingConnectionFactory connectionFactory = new CachingConnectionFactory(mockConnectionFactory);
+		AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
 		connectionFactory.setChannelListeners(Arrays.asList(new ChannelListener() {
 			public void onCreate(Channel channel, boolean transactional) {
 				called.incrementAndGet();
 			}
 		}));
-		connectionFactory.setChannelCacheSize(1);
+		((CachingConnectionFactory)connectionFactory).setChannelCacheSize(1);
 
 		Connection con = connectionFactory.createConnection();
 		Channel channel = con.createChannel(false);
@@ -414,5 +386,4 @@ public class CachingConnectionFactoryTests {
 		verify(mockConnectionFactory).newConnection();
 
 	}
-
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java

@@ -4,7 +4,6 @@ import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -14,119 +13,55 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConnectionFactory;
+
 /**
  * @author Dave Syer
  */
-public class SingleConnectionFactoryTests {
-
-	@Test
-	public void testWithListener() throws IOException {
-
-		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
-		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
-
-		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
-
-		final AtomicInteger called = new AtomicInteger(0);
-		SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory);
-		connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() {
-			public void onCreate(Connection connection) {
-				called.incrementAndGet();
-			}
-			public void onClose(Connection connection) {
-				called.decrementAndGet();
-			}
-		}));
-
-		Connection con = connectionFactory.createConnection();
-		assertEquals(1, called.get());
-
-		con.close();
-		assertEquals(1, called.get());
-		verify(mockConnection, never()).close();
-		
-		connectionFactory.createConnection();
-		assertEquals(1, called.get());
-
-		connectionFactory.destroy();
-		assertEquals(0, called.get());
-		verify(mockConnection, atLeastOnce()).close();
-
-		verify(mockConnectionFactory, times(1)).newConnection();
-
-	}
+public class SingleConnectionFactoryTests extends AbstractConnectionFactoryTests {
 	
+	@Override
+	protected AbstractConnectionFactory createConnectionFactory(ConnectionFactory connectionFactory) {
+		return new SingleConnectionFactory(connectionFactory);
+	}
+
 	@Test
-	public void testWithListenerRegisteredAfterOpen() throws IOException {
+	public void testWithChannelListener() throws IOException {
 
 		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
 		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
+		Channel mockChannel = mock(Channel.class);
 
 		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
+		when(mockConnection.isOpen()).thenReturn(true);
+		when(mockConnection.createChannel()).thenReturn(mockChannel);
 
 		final AtomicInteger called = new AtomicInteger(0);
-		SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory);
-		Connection con = connectionFactory.createConnection();
-		assertEquals(0, called.get());
-
-		connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() {
-			public void onCreate(Connection connection) {
+		AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
+		connectionFactory.setChannelListeners(Arrays.asList(new ChannelListener() {
+			public void onCreate(Channel channel, boolean transactional) {
 				called.incrementAndGet();
 			}
-			public void onClose(Connection connection) {
-				called.decrementAndGet();
-			}
 		}));
-		assertEquals(1, called.get());
 
-		con.close();
+		Connection con = connectionFactory.createConnection();
+		Channel channel = con.createChannel(false);
 		assertEquals(1, called.get());
+		channel.close();
+		
+		con.close();
 		verify(mockConnection, never()).close();
 		
 		connectionFactory.createConnection();
-		assertEquals(1, called.get());
+		con.createChannel(false);
+		assertEquals(2, called.get());
 
 		connectionFactory.destroy();
-		assertEquals(0, called.get());
 		verify(mockConnection, atLeastOnce()).close();
 
-		verify(mockConnectionFactory, times(1)).newConnection();
+		verify(mockConnectionFactory).newConnection();
 
 	}
 
-	@Test
-	public void testCloseInvalidConnection() throws Exception {
-
-		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
-		com.rabbitmq.client.Connection mockConnection1 = mock(com.rabbitmq.client.Connection.class);
-		com.rabbitmq.client.Connection mockConnection2 = mock(com.rabbitmq.client.Connection.class);
-
-		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection1).thenReturn(mockConnection2);
-		// simulate a dead connection
-		when(mockConnection1.isOpen()).thenReturn(false);
-
-		SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory);
-		
-		Connection connection = connectionFactory.createConnection();
-		// the dead connection should be discarded
-		connection.createChannel(false);
-		verify(mockConnectionFactory, times(2)).newConnection();
-		verify(mockConnection2, times(1)).createChannel();
-		
-		connectionFactory.destroy();
-		verify(mockConnection2, times(1)).close();
-
-	}
-
-	@Test
-	public void testDestroyBeforeUsed() throws Exception {
-
-		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
-
-		SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory);
-		connectionFactory.destroy();
-
-		verify(mockConnectionFactory, never()).newConnection();
-	}
-
 }

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

@@ -160,7 +160,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 
 			assertEquals(concurrentConsumers, container.getActiveConsumerCount());
 			container.stop();
-			Thread.sleep(500L);
+			Thread.sleep(1000L);
 			assertEquals(0, container.getActiveConsumerCount());
 			if (!transactional) {
 

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

@@ -53,8 +53,8 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
 	private SimpleMessageListenerContainer container;
 
 	@Rule
-	public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.DEBUG, RabbitTemplate.class,
-			SimpleMessageListenerContainer.class, BlockingQueueConsumer.class);
+	public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.INFO, RabbitTemplate.class,
+			SimpleMessageListenerContainer.class, BlockingQueueConsumer.class, CachingConnectionFactory.class);
 
 	@Rule
 	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue, sendQueue);