AMQP-44: implement some basic re-connect logic mainly to support async use cases

- re-connect in listener container if channel is closed
- re-connect in SingleConnectionFactory if connection is closed
Both of these features are immediate (no back off) because they apply to common
scenarios where the broker is fine, but there was a client problem (e.g. misuse
of API) that caused the connection to be closed by the broker.  Clients using
RabbitTemplate will also benefit: if an operation fails they can retry it immediately
and if the only problem was a closed Channel or Connection it will succeed.

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

@@ -13,7 +13,6 @@
 
 package org.springframework.amqp.rabbit.connection;
 
-import java.io.IOException;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
@@ -22,7 +21,8 @@ import java.util.LinkedList;
 
 import javax.naming.OperationNotSupportedException;
 
-import org.springframework.amqp.AmqpIOException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.util.Assert;
 
@@ -54,6 +54,8 @@ import com.rabbitmq.client.Channel;
 // TODO are there heartbeats and/or exception thrown if a connection is broken?
 public class CachingConnectionFactory extends SingleConnectionFactory implements DisposableBean {
 
+	private final Log logger = LogFactory.getLog(getClass());
+
 	private int channelCacheSize = 1;
 
 	private final LinkedList<ChannelProxy> cachedChannelsNonTransactional = new LinkedList<ChannelProxy>();
@@ -62,6 +64,8 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
 
 	private volatile boolean active = true;
 
+	private ChannelCachingConnectionProxy targetConnection;
+
 	/**
 	 * Create a new CachingConnectionFactory initializing the hostname to be the value returned from
 	 * InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.
@@ -97,8 +101,9 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
 		return this.channelCacheSize;
 	}
 
-	protected Channel getChannel(Connection connection, boolean transactional) throws IOException {
-		LinkedList<ChannelProxy> channelList = transactional ? this.cachedChannelsTransactional : this.cachedChannelsNonTransactional;
+	private Channel getChannel(boolean transactional) {
+		LinkedList<ChannelProxy> channelList = transactional ? this.cachedChannelsTransactional
+				: this.cachedChannelsNonTransactional;
 		Channel channel = null;
 		synchronized (channelList) {
 			if (!channelList.isEmpty()) {
@@ -110,33 +115,35 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
 				logger.trace("Found cached Rabbit Channel");
 			}
 		} else {
-			channel = getCachedChannelProxy(connection, channelList, transactional);
+			channel = getCachedChannelProxy(channelList, transactional);
 		}
 		return channel;
 	}
 
-	protected ChannelProxy getCachedChannelProxy(Connection connection, LinkedList<ChannelProxy> channelList, boolean transactional) {
-		Channel targetChannel = createBareChannel(connection, transactional);
+	private ChannelProxy getCachedChannelProxy(LinkedList<ChannelProxy> channelList, boolean transactional) {
+		Channel targetChannel = createBareChannel(transactional);
 		if (logger.isDebugEnabled()) {
 			logger.debug("Creating cached Rabbit Channel from " + targetChannel);
 		}
 		return (ChannelProxy) Proxy.newProxyInstance(ChannelProxy.class.getClassLoader(),
-				new Class[] { ChannelProxy.class }, new CachedChannelInvocationHandler(connection, targetChannel,
-						channelList, transactional));
+				new Class[] { ChannelProxy.class }, new CachedChannelInvocationHandler(targetChannel, channelList,
+						transactional));
 	}
 
-	private Channel createBareChannel(Connection connection, boolean transactional) {
-		try {
-			return connection.createChannel(transactional);
-		} catch (IOException e) {
-			throw new AmqpIOException(e);
-		}
+	private Channel createBareChannel(boolean transactional) {
+		return this.targetConnection.createBareChannel(transactional);
+	}
+
+	@Override
+	protected Connection doCreateConnection() {
+		targetConnection = new ChannelCachingConnectionProxy(super.doCreateConnection());
+		return targetConnection;
 	}
 
 	/**
 	 * Reset the Channel cache and underlying shared Connection, to be reinitialized on next access.
 	 */
-	public void resetConnection() {
+	protected void reset() {
 		this.active = false;
 		synchronized (this.cachedChannelsNonTransactional) {
 			for (ChannelProxy channel : cachedChannelsNonTransactional) {
@@ -149,7 +156,7 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
 			this.cachedChannelsNonTransactional.clear();
 		}
 		this.active = true;
-		super.resetConnection();
+		super.reset();
 	}
 
 	@Override
@@ -164,15 +171,12 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
 
 		private final LinkedList<ChannelProxy> channelList;
 
-		private final Connection connection;
-
 		private final Object targetMonitor = new Object();
 
 		private final boolean transactional;
 
-		public CachedChannelInvocationHandler(Connection connection, Channel target,
-				LinkedList<ChannelProxy> channelList, boolean transactional) {
-			this.connection = connection;
+		public CachedChannelInvocationHandler(Channel target, LinkedList<ChannelProxy> channelList,
+				boolean transactional) {
 			this.target = target;
 			this.channelList = channelList;
 			this.transactional = transactional;
@@ -209,7 +213,13 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
 			} else if (methodName.equals("getTargetChannel")) {
 				// Handle getTargetChannel method: return underlying Channel.
 				return this.target;
-			} try {
+			}
+			try {
+				synchronized (targetMonitor) {
+					if (this.target == null) {
+						this.target = createBareChannel(transactional);
+					}
+				}
 				return method.invoke(this.target, args);
 			} catch (InvocationTargetException ex) {
 				if (!this.target.isOpen()) {
@@ -217,7 +227,7 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
 					logger.debug("Detected closed channel on exception.  Re-initializing: " + target);
 					synchronized (targetMonitor) {
 						if (!this.target.isOpen()) {
-							this.target = createBareChannel(connection, transactional);
+							this.target = createBareChannel(transactional);
 						}
 					}
 				}
@@ -231,6 +241,14 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
 		 * @param proxy the channel to close
 		 */
 		private void logicalClose(ChannelProxy proxy) throws Exception {
+			if (!this.target.isOpen()) {
+				synchronized (targetMonitor) {
+					if (!this.target.isOpen()) {
+						this.target = null;
+						return;
+					}
+				}
+			}
 			// Allow for multiple close calls...
 			if (!this.channelList.contains(proxy)) {
 				if (logger.isTraceEnabled()) {
@@ -244,15 +262,76 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
 			if (logger.isDebugEnabled()) {
 				logger.debug("Closing cached Channel: " + this.target);
 			}
+			if (this.target == null) {
+				return;
+			}
 			if (this.target.isOpen()) {
 				synchronized (targetMonitor) {
 					if (this.target.isOpen()) {
 						this.target.close();
 					}
+					this.target = null;
 				}
 			}
 		}
 
 	}
 
+	private class ChannelCachingConnectionProxy implements Connection, ConnectionProxy {
+
+		private volatile Connection target;
+
+		public ChannelCachingConnectionProxy(Connection target) {
+			this.target = target;
+		}
+
+		private Channel createBareChannel(boolean transactional) {
+			return target.createChannel(transactional);
+		}
+
+		public Channel createChannel(boolean transactional) {
+			Channel channel = getChannel(transactional);
+			return channel;
+		}
+
+		public void close() {
+			target.close();
+		}
+
+		public boolean isOpen() {
+			return target!=null && target.isOpen();
+		}
+
+		public Connection getTargetConnection() {
+			return target;
+		}
+
+		@Override
+		public int hashCode() {
+			return 31 + ((target == null) ? 0 : target.hashCode());
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (this == obj)
+				return true;
+			if (obj == null)
+				return false;
+			if (getClass() != obj.getClass())
+				return false;
+			ChannelCachingConnectionProxy other = (ChannelCachingConnectionProxy) obj;
+			if (target == null) {
+				if (other.target != null)
+					return false;
+			} else if (!target.equals(other.target))
+				return false;
+			return true;
+		}
+
+		@Override
+		public String toString() {
+			return "Shared Rabbit Connection: " + this.target;
+		}
+
+	}
 }

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

@@ -3,7 +3,7 @@
  */
 package org.springframework.amqp.rabbit.connection;
 
-import java.io.IOException;
+import org.springframework.amqp.AmqpException;
 
 import com.rabbitmq.client.Channel;
 
@@ -17,9 +17,9 @@ public interface Connection {
      * Create a new channel, using an internally allocated channel number.
      * @param transactional true if the channel should support transactions
      * @return a new channel descriptor, or null if none is available
-     * @throws IOException if an I/O problem is encountered
+     * @throws AmqpException if an I/O problem is encountered
      */
-    Channel createChannel(boolean transactional) throws IOException;
+    Channel createChannel(boolean transactional) throws AmqpException;
 
     /**
      * Close this connection and all its channels
@@ -28,9 +28,15 @@ public interface Connection {
      *
      * Waits for all the close operations to complete.
      *
-     * @throws IOException if an I/O problem is encountered
+     * @throws AmqpException if an I/O problem is encountered
      */
-    // TODO: throws AmqpException
-    void close() throws IOException;
+    void close() throws AmqpException;
+    
+    /**
+     * Flag to indicate the status of the connection.
+     * 
+     * @return true if the connection is open
+     */
+    boolean isOpen();
 
 }

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

@@ -1,17 +1,14 @@
 /*
  * 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.
+ * 
+ * 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.connection;
@@ -19,18 +16,22 @@ package org.springframework.amqp.rabbit.connection;
 import java.io.IOException;
 
 /**
- * An interface based ConnectionFactory for creating {@link com.rabbitmq.client.Connection}s.
+ * An interface based ConnectionFactory for creating {@link com.rabbitmq.client.Connection Connections}.
  * 
- * <p>NOTE: The Rabbit API contains a ConnectionFactory class (same name).
+ * <p>
+ * NOTE: The Rabbit API contains a ConnectionFactory class (same name).
  * 
  * @author Mark Fisher
+ * @author Dave Syer
  */
 public interface ConnectionFactory {
 
 	Connection createConnection() throws IOException;
 
 	String getHost();
-	
+
+	int getPort();
+
 	String getVirtualHost();
 
 }

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

@@ -0,0 +1,34 @@
+/*
+ * 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.connection;
+
+/**
+ * Subinterface of {@link Connection} to be implemented by 
+ * Connection proxies.  Allows access to the underlying target Connection
+ * 
+ * @author Dave Syer
+ * @see CachingConnectionFactory
+ */
+public interface ConnectionProxy extends Connection {
+
+	/**
+	 * Return the target Channel of this proxy.
+	 * <p>This will typically be the native provider Connection
+	 * @return the underlying Connection (never <code>null</code>)
+	 */
+	Connection getTargetConnection();
+}

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

@@ -17,6 +17,8 @@ package org.springframework.amqp.rabbit.connection;
 
 import java.io.IOException;
 
+import org.springframework.amqp.rabbit.support.RabbitUtils;
+
 import com.rabbitmq.client.Channel;
 
 public class SimpleConnection implements Connection {
@@ -27,12 +29,26 @@ public class SimpleConnection implements Connection {
 		this.delegate = delegate;	
 	}
 
-	public Channel createChannel(boolean transactional) throws IOException {
-		return delegate.createChannel();
+	// TODO: expose the transactional flag
+	public Channel createChannel(boolean transactional) {
+		try {
+			return delegate.createChannel();
+		} catch (IOException e) {
+			throw RabbitUtils.convertRabbitAccessException(e);
+		}
 	}
 
-	public void close() throws IOException {
-		delegate.close();
+	public void close() {
+		try {
+			delegate.close();
+		} catch (IOException e) {
+			throw RabbitUtils.convertRabbitAccessException(e);
+		}
 	}
 
+	public boolean isOpen() {
+		return delegate!=null && delegate.isOpen();
+	}
+
+
 }

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

@@ -1,17 +1,14 @@
 /*
  * 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.
+ * 
+ * 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.connection;
@@ -30,40 +27,35 @@ import org.springframework.util.StringUtils;
 import com.rabbitmq.client.Channel;
 
 /**
- * A {@link ConnectionFactory} implementation that returns the same Connections from all
- * {@link #createConnection()} calls, and ignores calls to {@link com.rabbitmq.client.Connection#close()}.
+ * A {@link ConnectionFactory} implementation that returns the same Connections from all {@link #createConnection()}
+ * calls, and ignores calls to {@link com.rabbitmq.client.Connection#close()}.
  * 
  * @author Mark Fisher
  * @author Mark Pollack
  * @author Dave Syer
  */
-//TODO are there heartbeats and/or exception thrown if a connection is broken?
+// TODO are there heartbeats and/or exception thrown if a connection is broken?
 public class SingleConnectionFactory implements ConnectionFactory, DisposableBean {
 
-	protected final Log logger = LogFactory.getLog(getClass());
-
-	private volatile int port = RabbitUtils.DEFAULT_PORT;
+	private final Log logger = LogFactory.getLog(getClass());
 
 	private final com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory;
-	
+
 	/** Raw Rabbit Connection */
 	private Connection targetConnection;
 
 	/** Proxy Connection */
 	private Connection connection;
-	
+
 	/** Synchronization monitor for the shared Connection */
 	private final Object connectionMonitor = new Object();
 
-
 	/**
-	 * Create a new SingleConnectionFactory initializing the hostname to be the 
-	 * value returned from InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws
-	 * an exception.
+	 * Create a new SingleConnectionFactory initializing the hostname to be the value returned from
+	 * InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.
 	 */
 	public SingleConnectionFactory() {
-		this.rabbitConnectionFactory = new com.rabbitmq.client.ConnectionFactory();
-		this.rabbitConnectionFactory.setHost(this.getDefaultHostName());
+		this((String) null);
 	}
 
 	/**
@@ -87,7 +79,6 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
 		this.rabbitConnectionFactory = rabbitConnectionFactory;
 	}
 
-
 	public void setUsername(String username) {
 		this.rabbitConnectionFactory.setUsername(username);
 	}
@@ -112,92 +103,65 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
 		this.rabbitConnectionFactory.setPort(port);
 	}
 
-	protected int getPort() {
-		return this.port;
+	public int getPort() {
+		return this.rabbitConnectionFactory.getPort();
 	}
 
-	protected Channel getChannel(Connection connection, boolean transactional) throws IOException {
-		return this.createChannel(connection);
-	}
-
-	private Channel createChannel(Connection connection) throws IOException {
-		//TODO overload with channel number.
-		return connection.createChannel(false);
-	}
-
-	public Connection createConnection() throws IOException {
+	public final Connection createConnection() throws IOException {
 		synchronized (this.connectionMonitor) {
 			if (this.connection == null) {
-				initConnection();
+				if (this.targetConnection != null) {
+					RabbitUtils.closeConnection(this.targetConnection);
+				}
+				this.targetConnection = doCreateConnection();
+				if (logger.isInfoEnabled()) {
+					logger.info("Established shared Rabbit Connection: " + this.targetConnection);
+				}
+				this.connection = new SharedConnectionProxy(this.targetConnection);
 			}
 			return this.connection;
 		}
 	}
 
-	public void initConnection() throws IOException {
+	/**
+	 * Close the underlying shared connection. The provider of this ConnectionFactory needs to care for proper shutdown.
+	 * <p>
+	 * As this bean implements DisposableBean, a bean factory will automatically invoke this on destruction of its
+	 * cached singletons.
+	 */
+	public final void destroy() {
 		synchronized (this.connectionMonitor) {
 			if (this.targetConnection != null) {
-				closeConnection(this.targetConnection);
-			}
-			this.targetConnection = doCreateConnection();
-			prepareConnection(this.targetConnection);
-			if (logger.isInfoEnabled()) {
-				logger.info("Established shared Rabbit Connection: " + this.targetConnection);
-			}
-			this.connection = getSharedConnectionProxy(this.targetConnection);
-		}
-	}
-
-	/**
-	 * Close the underlying shared connection.
-	 * The provider of this ConnectionFactory needs to care for proper shutdown.
-	 * <p>As this bean implements DisposableBean, a bean factory will
-	 * automatically invoke this on destruction of its cached singletons.
-	 */
-	public void destroy() {
-		resetConnection();
-	}	
-
-	/**
-	 * Reset the underlying shared Connection, to be reinitialized on next access.
-	 */
-	public void resetConnection() {
-		synchronized (this.connectionMonitor) {
-			if (this.targetConnection != null) {
-				closeConnection(this.targetConnection);
+				RabbitUtils.closeConnection(this.targetConnection);
 			}
 			this.targetConnection = null;
 			this.connection = null;
 		}
+		reset();
 	}
 
 	/**
-	 * Close the given Connection.
-	 * @param connection the Connection to close
+	 * Default implementation does nothing. Called on {@link #destroy()}.
 	 */
-	protected void closeConnection(Connection connection) {
-		if (logger.isDebugEnabled()) {
-			logger.debug("Closing shared Rabbit Connection: " + this.targetConnection);
-		}
+	protected void reset() {
+	}
+
+	/**
+	 * Create a Connection. This implementation just delegates to the underlying Rabbit ConnectionFactory. Subclasses
+	 * typically will decorate the result to provide additional features.
+	 * 
+	 * @return the new Connection
+	 */
+	protected Connection doCreateConnection() {
+		return createBareConnection();
+	}
+
+	private Connection createBareConnection() {
 		try {
-			//TODO there are other close overloads close(int closeCode, java.lang.String closeMessage, int timeout) 
-			connection.close();			
+			return new SimpleConnection(this.rabbitConnectionFactory.newConnection());
+		} catch (IOException e) {
+			throw RabbitUtils.convertRabbitAccessException(e);
 		}
-		catch (Throwable ex) {
-			logger.debug("Could not close shared Rabbit Connection", ex);
-		}
-	}
-
-	/**
-	 * Create a Rabbit Connection via this class's ConnectionFactory.
-	 * @return the new Rabbit Connection
-	 */
-	protected Connection doCreateConnection() throws IOException {	
-		return new SimpleConnection(this.rabbitConnectionFactory.newConnection());	
-	}
-
-	protected void prepareConnection(Connection con) throws IOException {
-		//TODO configure ShutdownListener, investigate reconnection exceptions
 	}
 
 	private String getDefaultHostName() {
@@ -206,45 +170,54 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
 			InetAddress localMachine = InetAddress.getLocalHost();
 			temp = localMachine.getHostName();
 			logger.debug("Using hostname [" + temp + "] for hostname.");
-		}
-		catch (UnknownHostException e) {
+		} catch (UnknownHostException e) {
 			logger.warn("Could not get host name, using 'localhost' as default value", e);
 			temp = "localhost";
 		}
 		return temp;
 	}
 
-	/**
-	 * Wrap the given Connection with a proxy that delegates every method call to it
-	 * but suppresses close calls. This is useful for allowing application code to
-	 * handle a special framework Connection just like an ordinary Connection from a
-	 * Rabbit ConnectionFactory.
-	 * @param target the original Connection to wrap
-	 * @return the wrapped Connection
-	 */
-	protected Connection getSharedConnectionProxy(Connection target) {
-		return new SharedConnectionProxy(target);
-	}
-
 	@Override
 	public String toString() {
-		return "SingleConnectionFactory [host=" + rabbitConnectionFactory.getHost() + ", port=" + port + "]";
+		return "SingleConnectionFactory [host=" + rabbitConnectionFactory.getHost() + ", port="
+				+ rabbitConnectionFactory.getPort() + "]";
 	}
 
-	private class SharedConnectionProxy implements Connection {
+	/**
+	 * Wrap a raw Connection with a proxy that delegates every method call to it but suppresses close calls. This is
+	 * useful for allowing application code to handle a special framework Connection just like an ordinary Connection
+	 * from a Rabbit ConnectionFactory.
+	 */
+	private class SharedConnectionProxy implements Connection, ConnectionProxy {
 
-		private final Connection target;
+		private volatile Connection target;
 
 		public SharedConnectionProxy(Connection target) {
 			this.target = target;
 		}
 
-		public Channel createChannel(boolean transactional) throws IOException {
-			Channel channel = getChannel(this.target, transactional);
+		public Channel createChannel(boolean transactional) {
+			if (!target.isOpen()) {
+				synchronized (this) {
+					if (!target.isOpen()) {
+						logger.debug("Detected closed connection. Opening a new one before creating Channel.");
+						target = createBareConnection();
+					}
+				}
+			}
+			Channel channel = target.createChannel(transactional);
 			return channel;
 		}
 
-		public void close() throws IOException {
+		public void close() {
+		}
+
+		public boolean isOpen() {
+			return target != null && target.isOpen();
+		}
+
+		public Connection getTargetConnection() {
+			return target;
 		}
 
 		@Override
@@ -264,12 +237,11 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
 			if (target == null) {
 				if (other.target != null)
 					return false;
-			}
-			else if (!target.equals(other.target))
+			} else if (!target.equals(other.target))
 				return false;
 			return true;
 		}
-		
+
 		@Override
 		public String toString() {
 			return "Shared Rabbit Connection: " + this.target;

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

@@ -691,6 +691,10 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
 			// Not locally transacted but it is transacted so it
 			// could be synchronized with an external transaction
 			ConnectionFactoryUtils.registerDeliveryTag(getConnectionFactory(), channel, deliveryTag);
+		} else if (ackRequired) {
+			if (ackRequired) {
+				channel.basicAck(deliveryTag, false);
+			}			
 		}
 
 	}
@@ -700,7 +704,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
 	 * @param channel the Rabbit Channel to roll back
 	 */
 	protected void rollbackIfNecessary(Channel channel) {
-		if (this.isChannelLocallyTransacted(channel)) {
+		boolean ackRequired = !getAcknowledgeMode().isAutoAck() && !getAcknowledgeMode().isManual();
+		if (ackRequired) {
 			/*
 			 * Re-queue messages and don't get them re-delivered to the same consumer, otherwise the broker just spins
 			 * trying to get us to accept the same message over and over
@@ -710,6 +715,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
 			} catch (IOException e) {
 				throw new AmqpIOException(e);
 			}
+		}
+		if (this.isChannelLocallyTransacted(channel)) {
 			// Transacted channel enabled by this container -> rollback.
 			RabbitUtils.rollbackIfNecessary(channel);
 		}
@@ -723,16 +730,22 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
 	 */
 	protected void rollbackOnExceptionIfNecessary(Channel channel, Message message, Throwable ex) throws Exception {
 
-		// TODO not sure if the exception at this point if only from the
-		// application or from Rabbit.
+		boolean ackRequired = !getAcknowledgeMode().isAutoAck() && !getAcknowledgeMode().isManual();
 		try {
 			if (this.isChannelTransacted()) {
 				if (logger.isDebugEnabled()) {
 					logger.debug("Initiating transaction rollback on application exception: " + ex);
 				}
 				RabbitUtils.rollbackIfNecessary(channel);
-				if (message != null) {
+			}
+			if (message != null) {
+				if (ackRequired) {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Rejecting message");
+					}
 					channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
+				}
+				if (this.isChannelTransacted()) {
 					// Need to commit the reject (=nack)
 					RabbitUtils.commitIfNecessary(channel);
 				}

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

@@ -44,6 +44,8 @@ public class BlockingQueueConsumer {
 
 	private final int prefetchCount;
 
+	private final boolean transactional;
+	
 	private final Channel channel;
 
 	private final AtomicBoolean cancelled = new AtomicBoolean(false);
@@ -52,9 +54,10 @@ public class BlockingQueueConsumer {
 
 	private final AcknowledgeMode acknowledgeMode;
 
-	public BlockingQueueConsumer(Channel channel, AcknowledgeMode acknowledgeMode, int prefetchCount, String... queues) {
+	public BlockingQueueConsumer(Channel channel, AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, String... queues) {
 		this.channel = channel;
 		this.acknowledgeMode = acknowledgeMode;
+		this.transactional = transactional;
 		this.prefetchCount = prefetchCount;
 		this.queues = queues;
 		this.consumer = new InternalConsumer(channel);
@@ -72,8 +75,9 @@ public class BlockingQueueConsumer {
 	 * Check if we are in shutdown mode and if so throw an exception.
 	 */
 	private void checkShutdown() {
-		if (shutdown != null)
+		if (shutdown != null) {
 			throw Utility.fixStackTrace(shutdown);
+		}
 	}
 
 	/**
@@ -150,7 +154,7 @@ public class BlockingQueueConsumer {
 	public void stop() {
 		cancelled.set(true);
 		logger.debug("Closing Rabbit Channel: " + channel);
-		RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), acknowledgeMode.isTransactionAllowed());
+		RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), transactional);
 		RabbitUtils.closeChannel(channel);
 	}
 
@@ -162,9 +166,11 @@ public class BlockingQueueConsumer {
 
 		@Override
 		public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Received shutdown for consumer tag=" + consumerTag, sig);
+			}
 			shutdown = sig;
 			// TODO: interrupt?
-			// TODO: is this ever used?
 		}
 
 		@Override
@@ -177,7 +183,6 @@ public class BlockingQueueConsumer {
 			}
 			// TODO: do we want to pass on 'consumerTag'?
 			logger.debug("Storing delivery for " + BlockingQueueConsumer.this);
-			checkShutdown();
 			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

@@ -14,8 +14,6 @@
 package org.springframework.amqp.rabbit.listener;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -39,7 +37,6 @@ import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.ShutdownSignalException;
 
 /**
  * @author Mark Pollack
@@ -60,7 +57,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 
 	private volatile Executor taskExecutor = new SimpleAsyncTaskExecutor();
 
-	private volatile int concurrentConsumers = 1;
+	private volatile int concurrentConsumers = 0;
 
 	private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
 
@@ -161,7 +158,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 						"CachingConnectionFactory's channelCacheSize can not be less than the number of concurrentConsumers");
 			}
 			// Default setting
-			if (concurrentConsumers == 1) {
+			if (concurrentConsumers < 1) {
+				concurrentConsumers = 1;
 				// Set concurrent consumers to size of connection factory
 				// channel cache.
 				if (cf.getChannelCacheSize() > 1) {
@@ -204,10 +202,16 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	protected void doStart() throws Exception {
 		super.doStart();
 		initializeConsumers();
-		cancellationLock = new CountDownLatch(this.consumers.size());
-		for (BlockingQueueConsumer consumer : this.consumers) {
-			this.taskExecutor
-					.execute(new AsyncMessageProcessingConsumer(consumer, this.txSize, this, cancellationLock));
+		synchronized (this.consumersMonitor) {
+			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());
+			for (BlockingQueueConsumer consumer : this.consumers) {
+				this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, this.txSize, this,
+						cancellationLock));
+			}
 		}
 	}
 
@@ -236,20 +240,18 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 			logger.debug("Interrupted waiting for workers.  Continuing with shutdown.");
 		}
 
-		this.consumers = null;
+		synchronized (this.consumersMonitor) {
+			this.consumers = null;
+		}
 
 	}
 
 	protected void initializeConsumers() throws IOException {
 		synchronized (this.consumersMonitor) {
 			if (this.consumers == null) {
-				Collection<Channel> channels = new ArrayList<Channel>();
+				this.consumers = new HashSet<BlockingQueueConsumer>(this.concurrentConsumers);
 				for (int i = 0; i < this.concurrentConsumers; i++) {
 					Channel channel = getTransactionalResourceHolder().getChannel();
-					channels.add(channel);
-				}
-				this.consumers = new HashSet<BlockingQueueConsumer>(this.concurrentConsumers);
-				for (Channel channel : channels) {
 					BlockingQueueConsumer consumer = createBlockingQueueConsumer(channel);
 					this.consumers.add(consumer);
 				}
@@ -261,14 +263,34 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		return super.isChannelLocallyTransacted(channel) && this.transactionManager == null;
 	}
 
-	protected BlockingQueueConsumer createBlockingQueueConsumer(final Channel channel) throws IOException {
+	protected BlockingQueueConsumer createBlockingQueueConsumer(final Channel channel) {
 		BlockingQueueConsumer consumer;
 		String queueNames = getRequiredQueueName();
 		String[] queues = StringUtils.commaDelimitedListToStringArray(queueNames);
-		consumer = new BlockingQueueConsumer(channel, getAcknowledgeMode(), prefetchCount, queues);
+		consumer = new BlockingQueueConsumer(channel, getAcknowledgeMode(), isChannelTransacted(), prefetchCount,
+				queues);
 		return consumer;
 	}
 
+	private void restart(BlockingQueueConsumer consumer) {
+		synchronized (this.consumersMonitor) {
+			if (this.consumers != null) {
+				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);
+				this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, this.txSize, this,
+						cancellationLock));
+			}
+		}
+	}
+
 	private class AsyncMessageProcessingConsumer implements Runnable {
 
 		private final BlockingQueueConsumer consumer;
@@ -317,8 +339,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 			} catch (InterruptedException e) {
 				logger.debug("Consumer thread interrupted, processing stopped.");
 				Thread.currentThread().interrupt();
-			} catch (ShutdownSignalException e) {
-				logger.debug("Consumer received ShutdownSignal, processing stopped.", e);
 			} catch (Throwable t) {
 				logger.debug("Consumer received fatal exception, processing stopped.", t);
 			} finally {
@@ -326,8 +346,12 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 				if (!isActive()) {
 					logger.debug("Cancelling " + consumer);
 					consumer.stop();
+				} else {
+					logger.debug("Restarting " + consumer);
+					restart(consumer);
 				}
 			}
+
 		}
 
 		private boolean transactionalReceiveAndExecute() throws Exception {

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

@@ -14,6 +14,7 @@ import org.springframework.amqp.rabbit.core.ChannelCallback;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 
 import com.rabbitmq.client.Channel;
 
@@ -29,13 +30,13 @@ public class CachingConnectionFactoryIntegrationTests {
 	
 	@Before
 	public void open() {
-		// connectionFactory.setPort(5673);
+		connectionFactory.setPort(BrokerTestUtils.getPort());
 	}
 
 	@After
 	public void close() {
 		// Release resources
-		connectionFactory.resetConnection();
+		connectionFactory.reset();
 	}
 
 	@Test
@@ -61,7 +62,7 @@ public class CachingConnectionFactoryIntegrationTests {
 		template.convertAndSend(queue.getName(), "message");
 
 		// Force a physical close of the channel
-		connectionFactory.resetConnection();
+		connectionFactory.destroy();
 		
 		// The queue was removed when the channel was closed 
 		exception.expect(AmqpIOException.class);

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

@@ -25,37 +25,38 @@ import com.rabbitmq.client.GetResponse;
 public class CachingConnectionFactoryTests {
 
 	@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);
 		Channel mockChannel = mock(Channel.class);
-
+		
 		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
 		when(mockConnection.createChannel()).thenReturn(mockChannel);
-
+		when(mockChannel.isOpen()).thenReturn(true);
+		when(mockConnection.isOpen()).thenReturn(true);
+		
 		CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
 		Connection con = ccf.createConnection();
-
+		
 		Channel channel = con.createChannel(false);
 		channel.close(); // should be ignored, and placed into channel cache.
 		con.close(); // should be ignored
-
+		
 		Connection con2 = ccf.createConnection();
-		Channel channel2 = con2.createChannel(false); // will retrieve same
-														// channel object that
-														// was just put into
-														// channel
-														// cache
+		/*
+		 * will retrieve same channel object that was just put into channel cache
+		 */
+		Channel channel2 = con2.createChannel(false);
 		channel2.close(); // should be ignored
 		con2.close(); // should be ignored
-
+		
 		Assert.assertSame(con, con2);
 		Assert.assertSame(channel, channel2);
 		verify(mockConnection, never()).close();
 		verify(mockChannel, never()).close();
-
+		
 	}
-
 	@Test
 	public void testWithConnectionFactoryCacheSize() throws IOException {
 		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
@@ -64,10 +65,13 @@ public class CachingConnectionFactoryTests {
 		Channel mockChannel2 = mock(Channel.class);
 
 		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
+		when(mockConnection.isOpen()).thenReturn(true);
 		when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
 
 		when(mockChannel1.basicGet("foo", false)).thenReturn(new GetResponse(null, null, null, 1));
 		when(mockChannel2.basicGet("bar", false)).thenReturn(new GetResponse(null, null, null, 1));
+		when(mockChannel1.isOpen()).thenReturn(true);
+		when(mockChannel2.isOpen()).thenReturn(true);
 
 		CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
 		ccf.setChannelCacheSize(2);
@@ -115,9 +119,12 @@ public class CachingConnectionFactoryTests {
 
 		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
 		when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2).thenReturn(mockChannel3);
+		when(mockConnection.isOpen()).thenReturn(true);
 
 		// Called during physical close
+		when(mockChannel1.isOpen()).thenReturn(true);
 		when(mockChannel2.isOpen()).thenReturn(true);
+		when(mockChannel3.isOpen()).thenReturn(true);
 
 		CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
 		ccf.setChannelCacheSize(1);
@@ -130,12 +137,15 @@ public class CachingConnectionFactoryTests {
 		Channel channel2 = con.createChannel(false);
 		Assert.assertNotSame(channel1, channel2);
 
-		channel1.close(); // should be ignored, and add last into channel cache.
-		channel2.close(); // should be physically closed
+		// should be ignored, and added last into channel cache.
+		channel1.close();
+		// should be physically closed
+		channel2.close();
 
-		Channel ch1 = con.createChannel(false); // remove first entry in cache
-												// (channel1)
-		Channel ch2 = con.createChannel(false); // create anew channel
+		// remove first entry in cache (channel1)
+		Channel ch1 = con.createChannel(false);
+		// create a new channel
+		Channel ch2 = con.createChannel(false);
 
 		Assert.assertNotSame(ch1, ch2);
 		Assert.assertSame(ch1, channel1);
@@ -151,7 +161,7 @@ public class CachingConnectionFactoryTests {
 		verify(mockConnection, never()).close();
 		verify(mockChannel1, never()).close();
 		verify(mockChannel2, atLeastOnce()).close();
-		verify(mockChannel3, never()).close();
+		verify(mockChannel3, atLeastOnce()).close();
 
 	}
 
@@ -164,8 +174,10 @@ public class CachingConnectionFactoryTests {
 
 		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
 		when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
+		when(mockConnection.isOpen()).thenReturn(true);
 
 		// Called during physical close
+		when(mockChannel1.isOpen()).thenReturn(true);
 		when(mockChannel2.isOpen()).thenReturn(true);
 
 		CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
@@ -209,8 +221,10 @@ public class CachingConnectionFactoryTests {
 
 		when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
 		when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
+		when(mockConnection.isOpen()).thenReturn(true);
 
 		// Called during physical close
+		when(mockChannel1.isOpen()).thenReturn(true);
 		when(mockChannel2.isOpen()).thenReturn(true);
 
 		CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
@@ -245,7 +259,7 @@ public class CachingConnectionFactoryTests {
 		verify(mockConnection, never()).close();
 		verify(mockChannel1, never()).close();
 		verify(mockChannel2, never()).close();
-		
+
 		@SuppressWarnings("unchecked")
 		List<Channel> notxlist = (List<Channel>) ReflectionTestUtils.getField(ccf, "cachedChannelsNonTransactional");
 		assertEquals(1, notxlist.size());
@@ -270,6 +284,10 @@ public class CachingConnectionFactoryTests {
 		// the same method to returning different
 		// values.
 		when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
+		when(mockConnection.isOpen()).thenReturn(true);
+		// Called during physical close
+		when(mockChannel1.isOpen()).thenReturn(true);
+		when(mockChannel2.isOpen()).thenReturn(true);
 
 		CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
 		ccf.setChannelCacheSize(2);

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

@@ -228,7 +228,7 @@ public class RabbitBindingIntegrationTests {
 	}
 
 	private BlockingQueueConsumer createConsumer(Channel channel) {
-		BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, AcknowledgeMode.AUTO, 1, queue.getName());
+		BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, AcknowledgeMode.AUTO, true, 1, queue.getName());
 		consumer.start();
 		return consumer;
 	}

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

@@ -9,6 +9,7 @@ import org.junit.Rule;
 import org.junit.Test;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
 import org.springframework.amqp.rabbit.test.RepeatProcessor;
 import org.springframework.test.annotation.Repeat;
@@ -40,7 +41,7 @@ public class RabbitTemplatePerformanceIntegrationTests {
 		}
 		connectionFactory = new CachingConnectionFactory();
 		connectionFactory.setChannelCacheSize(repeat.getConcurrency());
-		// connectionFactory.setPort(5673);
+		connectionFactory.setPort(BrokerTestUtils.getPort());
 		template.setConnectionFactory(connectionFactory);
 	}
 

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

@@ -0,0 +1,247 @@
+package org.springframework.amqp.rabbit.listener;
+
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.log4j.Level;
+import org.junit.After;
+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.connection.CachingConnectionFactory;
+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.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
+import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
+import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
+
+import com.rabbitmq.client.Channel;
+
+public class MessageListenerCachingConnectionIntegrationTests {
+
+	private static Log logger = LogFactory.getLog(MessageListenerCachingConnectionIntegrationTests.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;
+
+	@Rule
+	public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.DEBUG, RabbitTemplate.class,
+			SimpleMessageListenerContainer.class, BlockingQueueConsumer.class);
+
+	@Rule
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue);
+
+	protected ConnectionFactory createConnectionFactory() {
+		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
+		connectionFactory.setChannelCacheSize(concurrentConsumers);
+		connectionFactory.setPort(BrokerTestUtils.getPort());
+		return 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();
+		}
+	}
+
+	@Test
+	public void testListenerRecoversFromBogusDoubleAck() throws Exception {
+
+		RabbitTemplate template = new RabbitTemplate(createConnectionFactory());
+
+		acknowledgeMode = AcknowledgeMode.MANUAL;
+
+		CountDownLatch latch = new CountDownLatch(messageCount);
+		container = createContainer(new ManualAckListener(latch), createConnectionFactory());
+		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()));
+
+	}
+
+	@Test
+	public void testListenerRecoversFromClosedChannel() throws Exception {
+
+		RabbitTemplate template = new RabbitTemplate(createConnectionFactory());
+
+		CountDownLatch latch = new CountDownLatch(messageCount);
+		container = createContainer(new AbortChannelListener(latch), createConnectionFactory());
+		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()));
+
+	}
+
+	@Test
+	public void testListenerRecoversFromClosedConnection() throws Exception {
+
+		RabbitTemplate template = new RabbitTemplate(createConnectionFactory());
+
+		CountDownLatch latch = new CountDownLatch(messageCount);
+		ConnectionFactory connectionFactory = createConnectionFactory();
+		container = createContainer(new CloseConnectionListener((ConnectionProxy) connectionFactory.createConnection(),
+				latch), connectionFactory);
+		for (int i = 0; i < messageCount; i++) {
+			template.convertAndSend(queue.getName(), i + "foo");
+		}
+
+		int timeout = Math.min(4 + 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()));
+
+	}
+
+	@Test
+	public void testListenerRecoversAndTemplateSharesConnectionFactory() throws Exception {
+
+		ConnectionFactory connectionFactory = createConnectionFactory();
+		RabbitTemplate template = new RabbitTemplate(connectionFactory);
+
+		acknowledgeMode = AcknowledgeMode.MANUAL;
+
+		CountDownLatch latch = new CountDownLatch(messageCount);
+		container = createContainer(new ManualAckListener(latch), connectionFactory);
+		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);
+
+		// TODO: is there a race condition here where the template gets a closed connection and this receive fails?
+		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 ManualAckListener implements ChannelAwareMessageListener {
+
+		private AtomicBoolean failed = new AtomicBoolean(false);
+
+		private final CountDownLatch latch;
+
+		public ManualAckListener(CountDownLatch latch) {
+			this.latch = latch;
+		}
+
+		public void onMessage(Message message, Channel channel) throws Exception {
+			String value = new String(message.getBody());
+			try {
+				logger.debug("Acking: " + value);
+				channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+				if (failed.compareAndSet(false, true)) {
+					// intentional error (causes exception on connection thread):
+					channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+				}
+			} finally {
+				latch.countDown();
+			}
+		}
+	}
+
+	public static class AbortChannelListener implements ChannelAwareMessageListener {
+
+		private AtomicBoolean failed = new AtomicBoolean(false);
+
+		private final CountDownLatch latch;
+
+		public AbortChannelListener(CountDownLatch latch) {
+			this.latch = latch;
+		}
+
+		public void onMessage(Message message, Channel channel) throws Exception {
+			String value = new String(message.getBody());
+			logger.debug("Receiving: " + value);
+			if (failed.compareAndSet(false, true)) {
+				// intentional error (causes exception on connection thread):
+				channel.abort();
+			} else {
+				latch.countDown();
+			}
+		}
+	}
+
+	public static class CloseConnectionListener implements ChannelAwareMessageListener {
+
+		private AtomicBoolean failed = new AtomicBoolean(false);
+
+		private final CountDownLatch latch;
+
+		private final Connection connection;
+
+		public CloseConnectionListener(ConnectionProxy connection, CountDownLatch latch) {
+			this.connection = connection.getTargetConnection();
+			this.latch = latch;
+		}
+
+		public void onMessage(Message message, Channel channel) throws Exception {
+			String value = new String(message.getBody());
+			logger.debug("Receiving: " + value);
+			if (failed.compareAndSet(false, true)) {
+				// intentional error (causes exception on connection thread):
+				connection.close();
+			} else {
+				latch.countDown();
+			}
+		}
+	}
+}

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

@@ -20,6 +20,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
 
 public class MessageListenerContainerLifecycleIntegrationTests {
@@ -86,7 +87,7 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 		// SingleConnectionFactory connectionFactory = new SingleConnectionFactory();
 		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
 		connectionFactory.setChannelCacheSize(concurrentConsumers);
-		// connectionFactory.setPort(5673); // For Tracer
+		connectionFactory.setPort(BrokerTestUtils.getPort());
 		template.setConnectionFactory(connectionFactory);
 		return template;
 	}

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

@@ -21,6 +21,7 @@ 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.BrokerRunning;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
 
 import com.rabbitmq.client.Channel;
@@ -54,7 +55,7 @@ public class MessageListenerManualAckIntegrationTests {
 	public void createConnectionFactory() {
 		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
 		connectionFactory.setChannelCacheSize(concurrentConsumers);
-		// connectionFactory.setPort(5673);
+		connectionFactory.setPort(BrokerTestUtils.getPort());
 		template.setConnectionFactory(connectionFactory);
 	}
 

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

@@ -0,0 +1,15 @@
+package org.springframework.amqp.rabbit.listener;
+
+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 {
+
+	protected ConnectionFactory createConnectionFactory() {
+		SingleConnectionFactory connectionFactory = new SingleConnectionFactory();
+		connectionFactory.setPort(BrokerTestUtils.getPort());
+		return connectionFactory;
+	}
+
+}

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

@@ -26,6 +26,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
 import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.TransactionException;
@@ -60,11 +61,14 @@ public class SimpleMessageListenerContainerIntegrationTests {
 
 	private final boolean externalTransaction;
 
+	private final boolean transactional;
+
 	public SimpleMessageListenerContainerIntegrationTests(int messageCount, int concurrency,
-			AcknowledgeMode acknowledgeMode, int txSize, boolean externalTransaction) {
+			AcknowledgeMode acknowledgeMode, boolean transactional, int txSize, boolean externalTransaction) {
 		this.messageCount = messageCount;
 		this.concurrentConsumers = concurrency;
 		this.acknowledgeMode = acknowledgeMode;
+		this.transactional = transactional;
 		this.txSize = txSize;
 		this.externalTransaction = externalTransaction;
 	}
@@ -76,19 +80,29 @@ public class SimpleMessageListenerContainerIntegrationTests {
 				params(1, 1, 1, AcknowledgeMode.NONE), //
 				params(2, 4, 1, AcknowledgeMode.AUTO), //
 				extern(3, 4, 1, AcknowledgeMode.AUTO), //
-				params(4, 2, 2, AcknowledgeMode.AUTO), //
-				params(5, 2, 2, AcknowledgeMode.NONE), //
-				params(6, 20, 4, AcknowledgeMode.AUTO), //
-				params(7, 20, 4, AcknowledgeMode.NONE), //
-				params(8, 1000, 4, AcknowledgeMode.AUTO), //
-				params(9, 1000, 4, AcknowledgeMode.NONE), //
-				params(10, 1000, 4, AcknowledgeMode.AUTO, 10) //
+				params(4, 4, 1, AcknowledgeMode.AUTO, false), //
+				params(5, 2, 2, AcknowledgeMode.AUTO), //
+				params(6, 2, 2, AcknowledgeMode.NONE), //
+				params(7, 20, 4, AcknowledgeMode.AUTO), //
+				params(8, 20, 4, AcknowledgeMode.NONE), //
+				params(9, 1000, 4, AcknowledgeMode.AUTO), //
+				params(10, 1000, 4, AcknowledgeMode.NONE), //
+				params(11, 1000, 4, AcknowledgeMode.AUTO, 10) //
 				);
 	}
 
-	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, int txSize) {
+	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, boolean transactional, int txSize) {
 		// "i" is just a counter to make it easier to identify the test in the log
-		return new Object[] { messageCount, concurrency, acknowledgeMode, txSize, false };
+		return new Object[] { messageCount, concurrency, acknowledgeMode, transactional, txSize, false };
+	}
+
+	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, int txSize) {
+		// For this test always us a transaction if it makes sense...
+		return params(i, messageCount, concurrency, acknowledgeMode, acknowledgeMode.isTransactionAllowed(), txSize);
+	}
+
+	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, boolean transactional) {
+		return params(i, messageCount, concurrency, acknowledgeMode, transactional, 1);
 	}
 
 	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode) {
@@ -96,14 +110,14 @@ public class SimpleMessageListenerContainerIntegrationTests {
 	}
 
 	private static Object[] extern(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode) {
-		return new Object[] { messageCount, concurrency, acknowledgeMode, 1, true };
+		return new Object[] { messageCount, concurrency, acknowledgeMode, true, 1, true };
 	}
 
 	@Before
 	public void declareQueue() {
 		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
 		connectionFactory.setChannelCacheSize(concurrentConsumers);
-		// connectionFactory.setPort(5673);
+		connectionFactory.setPort(BrokerTestUtils.getPort());
 		template.setConnectionFactory(connectionFactory);
 	}
 
@@ -124,7 +138,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		for (int i = 0; i < messageCount; i++) {
 			template.convertAndSend(queue.getName(), i + "foo");
 		}
-		boolean waited = latch.await(Math.max(1, messageCount / 50), TimeUnit.SECONDS);
+		boolean waited = latch.await(Math.max(2, messageCount / 50), TimeUnit.SECONDS);
 		assertTrue("Timed out waiting for message", waited);
 		assertNull(template.receiveAndConvert(queue.getName()));
 	}
@@ -167,8 +181,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		container.setTxSize(txSize);
 		container.setPrefetchCount(txSize);
 		container.setConcurrentConsumers(concurrentConsumers);
-		// For this test always us a transaction if it makes sense...
-		container.setChannelTransacted(acknowledgeMode.isTransactionAllowed());
+		container.setChannelTransacted(transactional);
 		container.setAcknowledgeMode(acknowledgeMode);
 		if (externalTransaction) {
 			container.setTransactionManager(new TestTransactionManager());

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

@@ -20,6 +20,7 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
@@ -46,7 +47,7 @@ public class UnackedRawIntegrationTests {
 	public void init() throws Exception {
 
 		factory.setHost("localhost");
-		// factory.setPort(5673);
+		factory.setPort(BrokerTestUtils.getPort());
 		conn = factory.newConnection();
 		noTxChannel = conn.createChannel();
 		txChannel = conn.createChannel();

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java

@@ -0,0 +1,31 @@
+/*
+ * 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.test;
+
+/**
+ * Global convenience class for all integration tests, carrying constants and other utilities for broker set up.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class BrokerTestUtils {
+	
+	public static final int DEFAULT_PORT = 5672;
+
+	public static final int TRACER_PORT = 5673;
+
+	public static int getPort() {
+		return DEFAULT_PORT;
+	}
+
+}