AMQP-100: Add AcknowledgeMode to complement channelTransacted (boolean)

spring-amqp-core/src/main/java/org/springframework/amqp/core/AcknowledgeMode.java

@@ -0,0 +1,35 @@
+/*
+ * 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.core;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public enum AcknowledgeMode {
+
+	NONE, MANUAL, AUTO;
+
+	public boolean isTransactionAllowed() {
+		return this == AUTO || this == MANUAL;
+	}
+
+	public boolean isAutoAck() {
+		return this == NONE;
+	}
+
+	public boolean isManual() {
+		return this == MANUAL;
+	}
+
+}

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

@@ -16,24 +16,47 @@ package org.springframework.amqp.rabbit.listener;
 import java.io.IOException;
 
 import org.springframework.amqp.AmqpIOException;
+import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.Connection;
 import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
 import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
 import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
+import org.springframework.amqp.rabbit.support.RabbitAccessor;
 import org.springframework.amqp.rabbit.support.RabbitUtils;
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
 
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
 
 /**
  * @author Mark Pollack
  * @author Mark Fisher
+ * @author Dave Syer
  */
-public abstract class AbstractMessageListenerContainer extends AbstractRabbitListeningContainer {
+public abstract class AbstractMessageListenerContainer extends RabbitAccessor implements BeanNameAware, DisposableBean,
+		SmartLifecycle {
+
+	// TODO See if can replace methods with general throws Exception signature to use a more specific exception.
+
+	private volatile String beanName;
+
+	private volatile Connection sharedConnection;
+
+	private volatile boolean autoStartup = true;
+
+	private int phase = Integer.MAX_VALUE;
+
+	private volatile boolean active = false;
+
+	private volatile boolean running = false;
+
+	private final Object lifecycleMonitor = new Object();
 
 	private volatile String queueName;
 
@@ -43,6 +66,42 @@ public abstract class AbstractMessageListenerContainer extends AbstractRabbitLis
 
 	private volatile Object messageListener;
 
+	private volatile AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
+
+	/**
+	 * <p>
+	 * Flag controlling the behaviour of the container with respect to message acknowledgement. The most common usage is
+	 * to let the container handle the acknowledgements (so the listener doesn't need to know about the channel or the
+	 * message).
+	 * </p>
+	 * <p>
+	 * Set to {@link AcknowledgeMode#MANUAL} if the listener will send the acknowledgements itself using
+	 * {@link Channel#basicAck(long, boolean)}. Manual acks are consistent with either a transactional or
+	 * non-transactional channel, but if you are doing no other work on the channel at the same other than receiving a
+	 * single message then the transaction is probably unnecessary.
+	 * </p>
+	 * <p>
+	 * Set to {@link AcknowledgeMode#NONE} to tell the broker not to expect any acknowledgements, and it will assume all
+	 * messages are acknowledged as soon as they are sent (this is "autoack" in native Rabbit broker terms). If
+	 * {@link AcknowledgeMode#NONE} then the channel cannot be transactional (so the container will fail on start up if
+	 * that flag is accidentally set).
+	 * </p>
+	 * 
+	 * @param acknowledgeMode the acknowledge mode to set. Defaults to {@link AcknowledgeMode#AUTO}
+	 * 
+	 * @see AcknowledgeMode
+	 */
+	public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
+		this.acknowledgeMode = acknowledgeMode;
+	}
+
+	/**
+	 * @return the acknowledgeMode
+	 */
+	public AcknowledgeMode getAcknowledgeMode() {
+		return acknowledgeMode;
+	}
+
 	/**
 	 * Set the name of the queue to receive messages from.
 	 * @param queueName the desired queue (can not be <code>null</code>)
@@ -87,11 +146,14 @@ public abstract class AbstractMessageListenerContainer extends AbstractRabbitLis
 
 	/**
 	 * Set whether to expose the listener Rabbit Channel to a registered {@link ChannelAwareMessageListener} as well as
-	 * to {@link org.springframework.amqp.rabbit.core.RabbitTemplate} calls. <p> Default is "true", reusing the
-	 * listener's {@link Channel}. Turn this off to expose a fresh Rabbit Channel fetched from the same underlying
-	 * Rabbit {@link Connection} instead. <p> Note that Channels managed by an external transaction manager will always
-	 * get exposed to {@link org.springframework.amqp.rabbit.core.RabbitTemplate} calls. So in terms of RabbitTemplate
-	 * exposure, this setting only affects locally transacted Channels.
+	 * to {@link org.springframework.amqp.rabbit.core.RabbitTemplate} calls.
+	 * <p>
+	 * Default is "true", reusing the listener's {@link Channel}. Turn this off to expose a fresh Rabbit Channel fetched
+	 * from the same underlying Rabbit {@link Connection} instead.
+	 * <p>
+	 * Note that Channels managed by an external transaction manager will always get exposed to
+	 * {@link org.springframework.amqp.rabbit.core.RabbitTemplate} calls. So in terms of RabbitTemplate exposure, this
+	 * setting only affects locally transacted Channels.
 	 * @see ChannelAwareMessageListener
 	 */
 	public void setExposeListenerChannel(boolean exposeListenerChannel) {
@@ -113,7 +175,8 @@ public abstract class AbstractMessageListenerContainer extends AbstractRabbitLis
 
 	/**
 	 * Check the given message listener, throwing an exception if it does not correspond to a supported listener type.
-	 * <p> By default, only a Spring {@link MessageListener} object or a Spring
+	 * <p>
+	 * By default, only a Spring {@link MessageListener} object or a Spring
 	 * {@link org.springframework.jms.listener.SessionAwareMessageListener} object will be accepted.
 	 * @param messageListener the message listener object to check
 	 * @throws IllegalArgumentException if the supplied listener is not a MessageListener or SessionAwareMessageListener
@@ -142,6 +205,347 @@ public abstract class AbstractMessageListenerContainer extends AbstractRabbitLis
 		this.errorHandler = errorHandler;
 	}
 
+	/**
+	 * Set whether to automatically start the container after initialization.
+	 * <p>
+	 * Default is "true"; set this to "false" to allow for manual startup through the {@link #start()} method.
+	 */
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	/**
+	 * Specify the phase in which this container should be started and stopped. The startup order proceeds from lowest
+	 * to highest, and the shutdown order is the reverse of that. By default this value is Integer.MAX_VALUE meaning
+	 * that this container starts as late as possible and stops as soon as possible.
+	 */
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	/**
+	 * Return the phase in which this container will be started and stopped.
+	 */
+	public int getPhase() {
+		return this.phase;
+	}
+
+	public void setBeanName(String beanName) {
+		this.beanName = beanName;
+	}
+
+	/**
+	 * Return the bean name that this listener container has been assigned in its containing bean factory, if any.
+	 */
+	protected final String getBeanName() {
+		return this.beanName;
+	}
+
+	/**
+	 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
+	 */
+	public final void afterPropertiesSet() {
+		super.afterPropertiesSet();
+		Assert.state(
+				exposeListenerChannel || !getAcknowledgeMode().isManual(),
+				"You cannot acknowledge messages manually if the channel is not exposed to the listener "
+						+ "(please check your configuration and set exposeListenerChannel=true or acknowledgeMode!=MANUAL)");
+		Assert.state(
+				!(getAcknowledgeMode().isAutoAck() && isChannelTransacted()),
+				"The acknowledgeMode is NONE (autoack in Rabbit terms) which is not consistent with having a "
+						+ "transactional channel. Either use a different AcknowledgeMode or make sure channelTransacted=false");
+		validateConfiguration();
+		initialize();
+	}
+
+	/**
+	 * Validate the configuration of this container.
+	 * <p>
+	 * The default implementation is empty. To be overridden in subclasses.
+	 */
+	protected void validateConfiguration() {
+	}
+
+	/**
+	 * Calls {@link #shutdown()} when the BeanFactory destroys the container instance.
+	 * @see #shutdown()
+	 */
+	public void destroy() {
+		shutdown();
+	}
+
+	// -------------------------------------------------------------------------
+	// Lifecycle methods for starting and stopping the container
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Initialize this container.
+	 * <p>
+	 * Creates a Rabbit Connection and calls {@link #doInitialize()}.
+	 */
+	public void initialize() {
+		try {
+			synchronized (this.lifecycleMonitor) {
+				this.lifecycleMonitor.notifyAll();
+			}
+			doInitialize();
+		} catch (Exception ex) {
+			ConnectionFactoryUtils.releaseConnection(this.sharedConnection);
+			this.sharedConnection = null;
+			throw convertRabbitAccessException(ex);
+		}
+	}
+
+	/**
+	 * Stop the shared Connection, call {@link #doShutdown()}, and close this container.
+	 */
+	public void shutdown() {
+		logger.debug("Shutting down Rabbit listener container");
+		synchronized (this.lifecycleMonitor) {
+			this.active = false;
+			this.lifecycleMonitor.notifyAll();
+		}
+
+		// Shut down the invokers.
+		try {
+			doShutdown();
+		} catch (Exception ex) {
+			throw convertRabbitAccessException(ex);
+		} finally {
+			synchronized (this.lifecycleMonitor) {
+				this.running = false;
+				this.lifecycleMonitor.notifyAll();
+			}
+			if (sharedConnectionEnabled()) {
+				ConnectionFactoryUtils.releaseConnection(this.sharedConnection);
+				this.sharedConnection = null;
+			}
+		}
+	}
+
+	/**
+	 * Register any invokers within this container.
+	 * <p>
+	 * Subclasses need to implement this method for their specific invoker management process.
+	 * <p>
+	 * A shared Rabbit Connection
+	 * @throws Exception
+	 * @see #getSharedConnection()
+	 */
+	protected abstract void doInitialize() throws Exception;
+
+	/**
+	 * Close the registered invokers.
+	 * <p>
+	 * Subclasses need to implement this method for their specific invoker management process.
+	 * <p>
+	 * A shared Rabbit Connection, if any, will automatically be closed <i>afterwards</i>.
+	 * @see #shutdown()
+	 */
+	protected abstract void doShutdown();
+
+	/**
+	 * Return whether this container is currently active, that is, whether it has been set up but not shut down yet.
+	 */
+	public final boolean isActive() {
+		synchronized (this.lifecycleMonitor) {
+			return this.active;
+		}
+	}
+
+	/**
+	 * Start this container.
+	 * @see #doStart
+	 */
+	public void start() {
+		try {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Starting Rabbit listener container.");
+			}
+			doStart();
+		} catch (Exception ex) {
+			throw convertRabbitAccessException(ex);
+		}
+	}
+
+	/**
+	 * Start the shared Connection, if any, and notify all invoker tasks.
+	 * @throws Exception if thrown by Rabbit API methods
+	 * @see #establishSharedConnection
+	 */
+	protected void doStart() throws Exception {
+		// Lazily establish a shared Connection, if necessary.
+		if (sharedConnectionEnabled()) {
+			establishSharedConnection();
+		}
+
+		// Reschedule paused tasks, if any.
+		synchronized (this.lifecycleMonitor) {
+			this.active = true;
+			this.running = true;
+			this.lifecycleMonitor.notifyAll();
+		}
+
+	}
+
+	/**
+	 * Stop this container.
+	 * @see #doStop
+	 */
+	public void stop() {
+		try {
+			doStop();
+		} catch (Exception ex) {
+			throw convertRabbitAccessException(ex);
+		} finally {
+			synchronized (this.lifecycleMonitor) {
+				this.running = false;
+				this.lifecycleMonitor.notifyAll();
+			}
+		}
+	}
+
+	public void stop(Runnable callback) {
+		this.stop();
+		callback.run();
+	}
+
+	/**
+	 * Notify all invoker tasks and stop the shared Connection, if any.
+	 * @see #stopSharedConnection
+	 */
+	protected void doStop() {
+		if (sharedConnectionEnabled()) {
+			stopSharedConnection();
+		}
+	}
+
+	/**
+	 * Determine whether this container is currently running, that is, whether it has been started and not stopped yet.
+	 * @see #start()
+	 * @see #stop()
+	 * @see #runningAllowed()
+	 */
+	public final boolean isRunning() {
+		synchronized (this.lifecycleMonitor) {
+			return (this.running && runningAllowed());
+		}
+	}
+
+	/**
+	 * Check whether this container's listeners are generally allowed to run.
+	 * <p>
+	 * This implementation always returns <code>true</code>; the default 'running' state is purely determined by
+	 * {@link #start()} / {@link #stop()}.
+	 * <p>
+	 * Subclasses may override this method to check against temporary conditions that prevent listeners from actually
+	 * running. In other words, they may apply further restrictions to the 'running' state, returning <code>false</code>
+	 * if such a restriction prevents listeners from running.
+	 */
+	protected boolean runningAllowed() {
+		return true;
+	}
+
+	// -------------------------------------------------------------------------
+	// Management of a shared Rabbit Connection
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Establish a shared Connection for this container.
+	 * <p>
+	 * The default implementation delegates to {@link #createSharedConnection()}, which does one immediate attempt and
+	 * throws an exception if it fails. Can be overridden to have a recovery process in place, retrying until a
+	 * Connection can be successfully established.
+	 * @throws Exception if thrown by Rabbit API methods
+	 */
+	protected void establishSharedConnection() throws Exception {
+		if (this.sharedConnection == null) {
+			this.sharedConnection = createSharedConnection();
+			logger.debug("Established shared Rabbit Connection");
+		}
+	}
+
+	/**
+	 * Refresh the shared Connection that this container holds.
+	 * <p>
+	 * Called on startup and also after an infrastructure exception that occurred during invoker setup and/or execution.
+	 * @throws Exception if thrown by Rabbit API methods
+	 */
+	protected final void refreshSharedConnection() throws Exception {
+		ConnectionFactoryUtils.releaseConnection(this.sharedConnection);
+		this.sharedConnection = null;
+		this.sharedConnection = createSharedConnection();
+	}
+
+	/**
+	 * Create a shared Connection for this container.
+	 * <p>
+	 * The default implementation creates a standard Connection and prepares it through {@link #prepareSharedConnection}.
+	 * @return the prepared Connection
+	 * @throws Exception if the creation failed
+	 */
+	protected Connection createSharedConnection() throws Exception {
+		Connection con = createConnection();
+		try {
+			prepareSharedConnection(con);
+			return con;
+		} catch (Exception ex) {
+			RabbitUtils.closeConnection(con);
+			throw ex;
+		}
+	}
+
+	/**
+	 * Prepare the given Connection, which is about to be registered as shared Connection for this container.
+	 * <p>
+	 * The default implementation sets the specified client id, if any. Subclasses can override this to apply further
+	 * settings.
+	 * @param connection the Connection to prepare
+	 */
+	protected void prepareSharedConnection(Connection connection) {
+	}
+
+	/**
+	 * Stop the shared Connection, logging any exception thrown by Rabbit API methods.
+	 */
+	protected void stopSharedConnection() {
+		if (this.sharedConnection != null) {
+			try {
+				this.sharedConnection.close();
+			} catch (Exception ex) {
+				logger.debug("Ignoring Connection close exception - assuming already closed: " + ex);
+			}
+		}
+	}
+
+	/**
+	 * Return the shared Rabbit Connection maintained by this container. Available after initialization.
+	 * @return the shared Connection (never <code>null</code>)
+	 * @throws IllegalStateException if this container does not maintain a shared Connection, or if the Connection
+	 * hasn't been initialized yet
+	 * @see #sharedConnectionEnabled()
+	 */
+	protected final Connection getSharedConnection() {
+		if (!sharedConnectionEnabled()) {
+			throw new IllegalStateException("This listener container does not maintain a shared Connection");
+		}
+		if (this.sharedConnection == null) {
+			throw new SharedConnectionNotInitializedException(
+					"This listener container's shared Connection has not been initialized yet");
+		}
+		return this.sharedConnection;
+	}
+
+	/**
+	 * Return whether a shared Rabbit Connection should be maintained by this container base class.
+	 * @see #getSharedConnection()
+	 */
+	protected abstract boolean sharedConnectionEnabled();
+
 	/**
 	 * Invoke the registered ErrorHandler, if any. Log at error level otherwise.
 	 * @param ex the uncaught error that arose during Rabbit processing.
@@ -257,8 +661,9 @@ public abstract class AbstractMessageListenerContainer extends AbstractRabbitLis
 	}
 
 	/**
-	 * Invoke the specified listener as Spring Rabbit MessageListener. <p> Default implementation performs a plain
-	 * invocation of the <code>onMessage</code> method.
+	 * Invoke the specified listener as Spring Rabbit MessageListener.
+	 * <p>
+	 * Default implementation performs a plain invocation of the <code>onMessage</code> method.
 	 * @param listener the Rabbit MessageListener to invoke
 	 * @param message the received Rabbit Message
 	 * @see org.springframework.amqp.core.MessageListener#onMessage
@@ -276,10 +681,13 @@ public abstract class AbstractMessageListenerContainer extends AbstractRabbitLis
 	protected void commitIfNecessary(Channel channel, Message message) throws IOException {
 
 		long deliveryTag = message.getMessageProperties().getDeliveryTag();
+		boolean ackRequired = !getAcknowledgeMode().isAutoAck() && !getAcknowledgeMode().isManual();
 		if (isChannelLocallyTransacted(channel)) {
-			channel.basicAck(deliveryTag, false);
+			if (ackRequired) {
+				channel.basicAck(deliveryTag, false);
+			}
 			RabbitUtils.commitIfNecessary(channel);
-		} else if (isChannelTransacted()) {
+		} else if (isChannelTransacted() && ackRequired) {
 			// Not locally transacted but it is transacted so it
 			// could be synchronized with an external transaction
 			ConnectionFactoryUtils.registerDeliveryTag(getConnectionFactory(), channel, deliveryTag);
@@ -337,8 +745,9 @@ public abstract class AbstractMessageListenerContainer extends AbstractRabbitLis
 
 	/**
 	 * Check whether the given Channel is locally transacted, that is, whether its transaction is managed by this
-	 * listener container's Channel handling and not by an external transaction coordinator. <p> Note:This method is
-	 * about finding out whether the Channel's transaction is local or externally coordinated.
+	 * listener container's Channel handling and not by an external transaction coordinator.
+	 * <p>
+	 * Note:This method is about finding out whether the Channel's transaction is local or externally coordinated.
 	 * @param channel the Channel to check
 	 * @return whether the given Channel is locally transacted
 	 * @see #isChannelTransacted()
@@ -348,9 +757,11 @@ public abstract class AbstractMessageListenerContainer extends AbstractRabbitLis
 	}
 
 	/**
-	 * Handle the given exception that arose during listener execution. <p> The default implementation logs the
-	 * exception at error level, not propagating it to the Rabbit provider - assuming that all handling of
-	 * acknowledgment and/or transactions is done by this listener container. This can be overridden in subclasses.
+	 * Handle the given exception that arose during listener execution.
+	 * <p>
+	 * The default implementation logs the exception at error level, not propagating it to the Rabbit provider -
+	 * assuming that all handling of acknowledgment and/or transactions is done by this listener container. This can be
+	 * overridden in subclasses.
 	 * @param ex the exception to handle
 	 */
 	protected void handleListenerException(Throwable ex) {
@@ -381,4 +792,21 @@ public abstract class AbstractMessageListenerContainer extends AbstractRabbitLis
 	private static class MessageRejectedWhileStoppingException extends RuntimeException {
 
 	}
+
+	/**
+	 * Exception that indicates that the initial setup of this container's shared Rabbit Connection failed. This is
+	 * indicating to invokers that they need to establish the shared Connection themselves on first access.
+	 */
+	@SuppressWarnings("serial")
+	public static class SharedConnectionNotInitializedException extends RuntimeException {
+
+		/**
+		 * Create a new SharedConnectionNotInitializedException.
+		 * @param msg the detail message
+		 */
+		protected SharedConnectionNotInitializedException(String msg) {
+			super(msg);
+		}
+	}
+
 }

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

@@ -1,422 +0,0 @@
-/*
- * Copyright 2002-2010 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.amqp.rabbit.listener;
-
-
-import org.springframework.amqp.rabbit.connection.Connection;
-import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
-import org.springframework.amqp.rabbit.support.RabbitAccessor;
-import org.springframework.amqp.rabbit.support.RabbitUtils;
-import org.springframework.beans.factory.BeanNameAware;
-import org.springframework.beans.factory.DisposableBean;
-import org.springframework.context.SmartLifecycle;
-
-/**
- * @author Mark Pollack
- */
-public abstract class AbstractRabbitListeningContainer extends RabbitAccessor implements BeanNameAware, DisposableBean, SmartLifecycle {
-
-	//TODO See if can replace methods with general throws Exception signature to use a more specific exception.
-	
-	private volatile String beanName;
-	
-	private volatile Connection sharedConnection;
-	
-	private volatile boolean autoStartup = true;
-	
-	private int phase = Integer.MAX_VALUE;
-	
-	private volatile boolean active = false;
-	
-	private volatile boolean running = false;
-	
-	private final Object lifecycleMonitor = new Object();
-
-
-	/**
-	 * Set whether to automatically start the container after initialization.
-	 * <p>Default is "true"; set this to "false" to allow for manual startup
-	 * through the {@link #start()} method.
-	 */
-	public void setAutoStartup(boolean autoStartup) {
-		this.autoStartup = autoStartup;
-	}
-
-	public boolean isAutoStartup() {
-		return this.autoStartup;
-	}
-
-	/**
-	 * Specify the phase in which this container should be started and
-	 * stopped. The startup order proceeds from lowest to highest, and
-	 * the shutdown order is the reverse of that. By default this value
-	 * is Integer.MAX_VALUE meaning that this container starts as late
-	 * as possible and stops as soon as possible.
-	 */
-	public void setPhase(int phase) {
-		this.phase = phase;
-	}
-
-	/**
-	 * Return the phase in which this container will be started and stopped.
-	 */
-	public int getPhase() {
-		return this.phase;
-	}
-
-	public void setBeanName(String beanName) {
-		this.beanName = beanName;
-	}
-
-	/**
-	 * Return the bean name that this listener container has been assigned
-	 * in its containing bean factory, if any.
-	 */
-	protected final String getBeanName() {
-		return this.beanName;
-	}
-	
-	/**
-	 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
-	 */
-	public void afterPropertiesSet() {
-		super.afterPropertiesSet();
-		validateConfiguration();
-		initialize();
-	}
-	
-	/**
-	 * Validate the configuration of this container.
-	 * <p>The default implementation is empty. To be overridden in subclasses.
-	 */
-	protected void validateConfiguration() {
-	}
-	
-	/**
-	 * Calls {@link #shutdown()} when the BeanFactory destroys the container instance.
-	 * @see #shutdown()
-	 */
-	public void destroy() {
-		shutdown();
-	}
-
-
-	//-------------------------------------------------------------------------
-	// Lifecycle methods for starting and stopping the container
-	//-------------------------------------------------------------------------
-
-	/**
-	 * Initialize this container.
-	 * <p>Creates a Rabbit Connection and calls {@link #doInitialize()}.
-	 */
-	public void initialize()  {
-		try {
-			synchronized (this.lifecycleMonitor) {
-				this.lifecycleMonitor.notifyAll();
-			}
-			doInitialize();
-		}
-		catch (Exception ex) {
-			ConnectionFactoryUtils.releaseConnection(this.sharedConnection);
-			this.sharedConnection = null;
-			throw convertRabbitAccessException(ex);
-		}
-	}
-	
-	/**
-	 * Stop the shared Connection, call {@link #doShutdown()},
-	 * and close this container.
-	 */
-	public void shutdown() {
-		logger.debug("Shutting down Rabbit listener container");
-		synchronized (this.lifecycleMonitor) {
-			this.active = false;
-			this.lifecycleMonitor.notifyAll();
-		}
-
-		// Shut down the invokers.
-		try {
-			doShutdown();
-		}
-		catch (Exception ex) {
-			throw convertRabbitAccessException(ex);
-		}
-		finally {
-			synchronized (this.lifecycleMonitor) {
-				this.running = false;
-				this.lifecycleMonitor.notifyAll();
-			}
-			if (sharedConnectionEnabled()) {				
-				ConnectionFactoryUtils.releaseConnection(this.sharedConnection);
-				this.sharedConnection = null;				
-			}
-		}
-	}
-	
-	/**
-	 * Return whether this container is currently active,
-	 * that is, whether it has been set up but not shut down yet.
-	 */
-	public final boolean isActive() {
-		synchronized (this.lifecycleMonitor) {
-			return this.active;
-		}
-	}
-	
-	/**
-	 * Start this container.
-	 * @see #doStart
-	 */
-	public void start()  {
-		try {
-			if (logger.isDebugEnabled()) {
-				logger.debug("Starting Rabbit listener container.");
-			}
-			doStart();
-		}
-		catch (Exception ex) {
-			throw convertRabbitAccessException(ex);
-		}
-	}
-	
-	/**
-	 * Start the shared Connection, if any, and notify all invoker tasks.
-	 * @throws Exception  if thrown by Rabbit API methods
-	 * @see #establishSharedConnection
-	 */
-	protected void doStart() throws Exception {
-		// Lazily establish a shared Connection, if necessary.
-		if (sharedConnectionEnabled()) {
-			establishSharedConnection();
-		}
-
-		// Reschedule paused tasks, if any.
-		synchronized (this.lifecycleMonitor) {
-			this.active = true;
-			this.running = true;
-			this.lifecycleMonitor.notifyAll();
-		}
-
-	}
-	
-	/**
-	 * Stop this container.
-	 * @see #doStop
-	 */
-	public void stop() {
-		try {
-			doStop();
-		}
-		catch (Exception ex) {
-			throw convertRabbitAccessException(ex);
-		} finally {
-			synchronized (this.lifecycleMonitor) {
-				this.running = false;
-				this.lifecycleMonitor.notifyAll();
-			}			
-		}
-	}
-
-	public void stop(Runnable callback) {
-		this.stop();
-		callback.run();
-	}
-	
-	/**
-	 * Notify all invoker tasks and stop the shared Connection, if any.
-	 * @see #stopSharedConnection
-	 */
-	protected void doStop() {
-		if (sharedConnectionEnabled()) {
-			stopSharedConnection();
-		}
-	}
-	
-	/**
-	 * Determine whether this container is currently running,
-	 * that is, whether it has been started and not stopped yet.
-	 * @see #start()
-	 * @see #stop()
-	 * @see #runningAllowed()
-	 */
-	public final boolean isRunning() {
-		synchronized (this.lifecycleMonitor) {
-			return (this.running && runningAllowed());
-		}
-	}
-	
-	/**
-	 * Check whether this container's listeners are generally allowed to run.
-	 * <p>This implementation always returns <code>true</code>; the default 'running'
-	 * state is purely determined by {@link #start()} / {@link #stop()}.
-	 * <p>Subclasses may override this method to check against temporary
-	 * conditions that prevent listeners from actually running. In other words,
-	 * they may apply further restrictions to the 'running' state, returning
-	 * <code>false</code> if such a restriction prevents listeners from running.
-	 */
-	protected boolean runningAllowed() {
-		return true;
-	}
-
-
-	//-------------------------------------------------------------------------
-	// Management of a shared Rabbit Connection
-	//-------------------------------------------------------------------------
-	
-	/**
-	 * Establish a shared Connection for this container.
-	 * <p>The default implementation delegates to {@link #createSharedConnection()},
-	 * which does one immediate attempt and throws an exception if it fails.
-	 * Can be overridden to have a recovery process in place, retrying
-	 * until a Connection can be successfully established.
-	 * @throws Exception if thrown by Rabbit API methods
-	 */
-	protected void establishSharedConnection() throws Exception {
-		if (this.sharedConnection == null) {
-			this.sharedConnection = createSharedConnection();
-			logger.debug("Established shared Rabbit Connection");
-		}
-	}
-
-	/**
-	 * Refresh the shared Connection that this container holds.
-	 * <p>Called on startup and also after an infrastructure exception
-	 * that occurred during invoker setup and/or execution.
-	 * @throws Exception if thrown by Rabbit API methods
-	 */
-	protected final void refreshSharedConnection() throws Exception {
-		ConnectionFactoryUtils.releaseConnection(
-				this.sharedConnection);
-		this.sharedConnection = null;
-		this.sharedConnection = createSharedConnection();					
-	}
-
-	/**
-	 * Create a shared Connection for this container.
-	 * <p>The default implementation creates a standard Connection
-	 * and prepares it through {@link #prepareSharedConnection}.
-	 * @return the prepared Connection
-	 * @throws Exception if the creation failed
-	 */
-	protected Connection createSharedConnection() throws Exception {
-		Connection con = createConnection();
-		try {
-			prepareSharedConnection(con);
-			return con;
-		}
-		catch (Exception ex) {
-			RabbitUtils.closeConnection(con);
-			throw ex;
-		}
-	}
-	
-	/**
-	 * Prepare the given Connection, which is about to be registered
-	 * as shared Connection for this container.
-	 * <p>The default implementation sets the specified client id, if any.
-	 * Subclasses can override this to apply further settings.
-	 * @param connection the Connection to prepare
-	 */
-	protected void prepareSharedConnection(Connection connection) {
-	}
-
-	/**
-	 * Stop the shared Connection, logging any exception thrown by
-	 * Rabbit API methods.
-	 */
-	protected void stopSharedConnection() {		
-		if (this.sharedConnection != null) {
-			try {
-				this.sharedConnection.close();
-			}
-			catch (Exception ex) {
-				logger.debug("Ignoring Connection close exception - assuming already closed: " + ex);
-			}
-		}		
-	}
-
-	/**
-	 * Return the shared Rabbit Connection maintained by this container.
-	 * Available after initialization.
-	 * @return the shared Connection (never <code>null</code>)
-	 * @throws IllegalStateException if this container does not maintain a
-	 * shared Connection, or if the Connection hasn't been initialized yet
-	 * @see #sharedConnectionEnabled()
-	 */
-	protected final Connection getSharedConnection() {
-		if (!sharedConnectionEnabled()) {
-			throw new IllegalStateException(
-					"This listener container does not maintain a shared Connection");
-		}
-		if (this.sharedConnection == null) {
-			throw new SharedConnectionNotInitializedException(
-					"This listener container's shared Connection has not been initialized yet");
-		}
-		return this.sharedConnection;
-	}
-
-
-	//-------------------------------------------------------------------------
-	// Template methods to be implemented by subclasses
-	//-------------------------------------------------------------------------
-
-	/**
-	 * Return whether a shared Rabbit Connection should be maintained
-	 * by this container base class.
-	 * @see #getSharedConnection()
-	 */
-	protected abstract boolean sharedConnectionEnabled();
-
-	/**
-	 * Register any invokers within this container.
-	 * <p>Subclasses need to implement this method for their specific
-	 * invoker management process.
-	 * <p>A shared Rabbit Connection
-	 * @throws Exception 
-	 * @see #getSharedConnection()
-	 */
-	protected abstract void doInitialize() throws Exception;
-
-	/**
-	 * Close the registered invokers.
-	 * <p>Subclasses need to implement this method for their specific
-	 * invoker management process.
-	 * <p>A shared Rabbit Connection, if any, will automatically be closed
-	 * <i>afterwards</i>.
-	 * @see #shutdown()
-	 */
-	protected abstract void doShutdown();
-	
-
-	/**
-	 * Exception that indicates that the initial setup of this container's
-	 * shared Rabbit Connection failed. This is indicating to invokers that they need
-	 * to establish the shared Connection themselves on first access.
-	 */
-	@SuppressWarnings("serial")
-	public static class SharedConnectionNotInitializedException extends RuntimeException {
-
-		/**
-		 * Create a new SharedConnectionNotInitializedException.
-		 * @param msg the detail message
-		 */
-		protected SharedConnectionNotInitializedException(String msg) {
-			super(msg);
-		}
-	}
-
-}

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

@@ -9,6 +9,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.support.RabbitUtils;
@@ -39,8 +40,6 @@ public class BlockingQueueConsumer {
 	// When this is non-null the connection has been closed (should never happen in normal operation).
 	private volatile ShutdownSignalException shutdown;
 
-	private final boolean transactional;
-
 	private final String[] queues;
 
 	private final int prefetchCount;
@@ -51,11 +50,13 @@ public class BlockingQueueConsumer {
 
 	private final InternalConsumer consumer;
 
-	public BlockingQueueConsumer(Channel channel, boolean transactional, int prefetchCount, String... queues) {
+	private final AcknowledgeMode acknowledgeMode;
+
+	public BlockingQueueConsumer(Channel channel, AcknowledgeMode acknowledgeMode, int prefetchCount, String... queues) {
 		this.channel = channel;
+		this.acknowledgeMode = acknowledgeMode;
 		this.prefetchCount = prefetchCount;
 		this.queues = queues;
-		this.transactional = transactional;
 		this.consumer = new InternalConsumer(channel);
 	}
 
@@ -83,7 +84,7 @@ public class BlockingQueueConsumer {
 	 */
 	private Message handle(Delivery delivery) throws InterruptedException {
 		if ((delivery == null && shutdown != null)) {
-			throw Utility.fixStackTrace(shutdown);
+			throw shutdown;
 		}
 		if (delivery == null) {
 			return null;
@@ -136,7 +137,7 @@ public class BlockingQueueConsumer {
 			channel.basicQos(prefetchCount);
 			for (int i = 0; i < queues.length; i++) {
 				channel.queueDeclarePassive(queues[i]);
-				channel.basicConsume(queues[i], !transactional, consumer);
+				channel.basicConsume(queues[i], acknowledgeMode.isAutoAck(), consumer);
 				if (logger.isDebugEnabled()) {
 					logger.debug("Started " + this);
 				}
@@ -149,7 +150,7 @@ public class BlockingQueueConsumer {
 	public void stop() {
 		cancelled.set(true);
 		logger.debug("Closing Rabbit Channel: " + channel);
-		RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), transactional);
+		RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), acknowledgeMode.isTransactionAllowed());
 		RabbitUtils.closeChannel(channel);
 	}
 
@@ -170,7 +171,7 @@ public class BlockingQueueConsumer {
 		public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
 				throws IOException {
 			if (cancelled.get()) {
-				if (transactional) {
+				if (acknowledgeMode.isTransactionAllowed()) {
 					return;
 				}
 			}
@@ -218,8 +219,8 @@ public class BlockingQueueConsumer {
 
 	@Override
 	public String toString() {
-		return "Consumer: tag=[" + consumer.getConsumerTag() + "], channel=" + channel + ", transactional="
-				+ transactional + " local queue size=" + queue.size();
+		return "Consumer: tag=[" + consumer.getConsumerTag() + "], channel=" + channel + ", acknowledgeMode="
+				+ acknowledgeMode + " local queue size=" + queue.size();
 	}
 
 }

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

@@ -84,10 +84,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	}
 
 	/**
-	 * Specify the number of concurrent consumers to create. Default is 1. <p> Raising the number of concurrent
-	 * consumers is recommended in order to scale the consumption of messages coming in from a queue. However, note that
-	 * any ordering guarantees are lost once multiple consumers are registered. In general, stick with 1 consumer for
-	 * low-volume queues.
+	 * Specify the number of concurrent consumers to create. Default is 1.
+	 * <p>
+	 * Raising the number of concurrent consumers is recommended in order to scale the consumption of messages coming in
+	 * from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In
+	 * general, stick with 1 consumer for low-volume queues.
 	 */
 	public void setConcurrentConsumers(int concurrentConsumers) {
 		Assert.isTrue(concurrentConsumers > 0, "'concurrentConsumers' value must be at least 1 (one)");
@@ -143,8 +144,16 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 	 * Avoid the possibility of not configuring the CachingConnectionFactory in sync with the number of concurrent
 	 * consumers.
 	 */
-	public void afterPropertiesSet() {
-		super.afterPropertiesSet();
+	@Override
+	protected void validateConfiguration() {
+
+		super.validateConfiguration();
+
+		Assert.state(
+				!(getAcknowledgeMode().isAutoAck() && transactionManager != null),
+				"The acknowledgeMode is NONE (autoack in Rabbit terms) which is not consistent with having an "
+						+ "external transaction manager. Either use a different AcknowledgeMode or make sure the transactionManager is null.");
+
 		if (this.getConnectionFactory() instanceof CachingConnectionFactory) {
 			CachingConnectionFactory cf = (CachingConnectionFactory) getConnectionFactory();
 			if (cf.getChannelCacheSize() < this.concurrentConsumers) {
@@ -162,6 +171,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 				}
 			}
 		}
+
 	}
 
 	// -------------------------------------------------------------------------
@@ -255,7 +265,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		BlockingQueueConsumer consumer;
 		String queueNames = getRequiredQueueName();
 		String[] queues = StringUtils.commaDelimitedListToStringArray(queueNames);
-		consumer = new BlockingQueueConsumer(channel, isChannelTransacted(), prefetchCount, queues);
+		consumer = new BlockingQueueConsumer(channel, getAcknowledgeMode(), prefetchCount, queues);
 		return consumer;
 	}
 
@@ -308,7 +318,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 				logger.debug("Consumer thread interrupted, processing stopped.");
 				Thread.currentThread().interrupt();
 			} catch (ShutdownSignalException e) {
-				logger.debug("Consumer received ShutdownSignal, processing stopped.");
+				logger.debug("Consumer received ShutdownSignal, processing stopped.", e);
 			} catch (Throwable t) {
 				logger.debug("Consumer received fatal exception, processing stopped.", t);
 			} finally {
@@ -355,8 +365,10 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 			int totalMsgCount = 0;
 
 			ConnectionFactory connectionFactory = getConnectionFactory();
-			ConnectionFactoryUtils
-					.bindResourceToTransaction(new RabbitResourceHolder(channel), connectionFactory, true);
+			if (getAcknowledgeMode().isTransactionAllowed()) {
+				ConnectionFactoryUtils.bindResourceToTransaction(new RabbitResourceHolder(channel), connectionFactory,
+						true);
+			}
 
 			for (int i = 0; i < txSize; i++) {
 

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

@@ -119,6 +119,8 @@ import com.rabbitmq.client.Channel;
  * @author Juergen Hoeller
  * @author Mark Pollack
  * @author Mark Fisher
+ * @author Dave Syer
+ * 
  * @see #setDelegate
  * @see #setDefaultListenerMethod
  * @see #setResponseRoutingKey(String)
@@ -298,7 +300,7 @@ public class MessageListenerAdapter	implements MessageListener, ChannelAwareMess
 	}
 
 	/**
-	 * Spring {@link org.springframework.jms.listener.SessionAwareMessageListener}
+	 * Spring {@link ChannelAwareMessageListener}
 	 * entry point.
 	 * <p>Delegates the message to the target listener method, with appropriate
 	 * conversion of the message argument. If the target method returns a

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitAccessor.java

@@ -32,6 +32,7 @@ import com.rabbitmq.client.Channel;
 
 /**
  * @author Mark Fisher
+ * @author Dave Syer
  */
 public abstract class RabbitAccessor implements InitializingBean {
 
@@ -40,7 +41,20 @@ public abstract class RabbitAccessor implements InitializingBean {
 
 	private volatile ConnectionFactory connectionFactory;
 
-	private volatile boolean channelTransacted;
+	private volatile boolean transactional;
+
+	public boolean isChannelTransacted() {
+		return this.transactional;
+	}
+
+	/**
+	 * Flag to indicate that channels created by this component will be transactional.
+	 * 
+	 * @param transactional the flag value to set
+	 */
+	public void setChannelTransacted(boolean transactional) {
+		this.transactional = transactional;
+	}
 
 	/**
 	 * Set the ConnectionFactory to use for obtaining RabbitMQ {@link Connection Connections}.
@@ -56,24 +70,7 @@ public abstract class RabbitAccessor implements InitializingBean {
 	public ConnectionFactory getConnectionFactory() {
 		return this.connectionFactory;
 	}
-
-	/**
-	 * Set the transaction mode that is used for a RabbitMQ {@link Channel},
-	 * Default is "false".
-	 */
-	public void setChannelTransacted(boolean channelTransacted) {
-		this.channelTransacted = channelTransacted;
-	}
-
-	/**
-	 * Return whether the RabbitMQ {@link Channel channels} used by this
-	 * accessor are supposed to be transacted.
-	 * @see #setChannelTransacted(boolean)
-	 */
-	public boolean isChannelTransacted() {
-		return this.channelTransacted;
-	}
-
+	
 	public void afterPropertiesSet() {
 		Assert.notNull(this.connectionFactory, "ConnectionFactory is required");
 	}
@@ -89,18 +86,6 @@ public abstract class RabbitAccessor implements InitializingBean {
 		return this.connectionFactory.createConnection();
 	}
 	
-	/**
-	 * Create a RabbitMQ Channel for the given Connection.
-	 * @param con the RabbitMQ Connection to create a Channel for
-	 * @return the new RabbitMQ Channel
-	 * @throws IOException if thrown by RabbitMQ API methods
-	 */
-	protected Channel createChannel(Connection con) throws IOException {
-		Assert.notNull(con, "connection must not be null");
-		Channel channel = con.createChannel(false);
-		return channel;
-	}
-	
 	/**
 	 * Fetch an appropriate Connection from the given RabbitResourceHolder.
 	 * 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java

@@ -22,7 +22,6 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.log4j.Level;
 import org.junit.AfterClass;
-import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Rule;
 import org.junit.Test;
@@ -48,26 +47,24 @@ public class RabbitBrokerAdminIntegrationTests {
 	public Log4jLevelAdjuster logLevel = new Log4jLevelAdjuster(Level.INFO, RabbitBrokerAdmin.class);
 
 	/*
-	 * Ensure broker dies if a test fails (otherwise the erl process has to be killed manually)
+	 * Ensure broker dies if a test fails (otherwise the erl process might have to be killed manually)
 	 */
 	@Rule
-	public BrokerPanic panic = new BrokerPanic();
+	public static BrokerPanic panic = new BrokerPanic();
 
 	private static RabbitBrokerAdmin brokerAdmin;
 
     private static final String NODE_NAME = "spring@localhost";
 
-	@Before
-	public void init() throws Exception {
-		panic.setBrokerAdmin(brokerAdmin);
-	}
-
 	@BeforeClass
 	public static void start() throws Exception {
-		brokerAdmin = new RabbitBrokerAdmin(NODE_NAME, 15672);
-        brokerAdmin.setRabbitLogBaseDirectory("target/rabbitmq/log");
-		brokerAdmin.setRabbitMnesiaBaseDirectory("target/rabbitmq/mnesia"); 
+		// Set up broker admin for non-root user
+		brokerAdmin = new RabbitBrokerAdmin("spring@localhost", 15672);
+		brokerAdmin.setRabbitLogBaseDirectory("target/rabbitmq/log");
+		brokerAdmin.setRabbitMnesiaBaseDirectory("target/rabbitmq/mnesia");
+		brokerAdmin.setStartupTimeout(10000L);
 		brokerAdmin.startNode();
+		panic.setBrokerAdmin(brokerAdmin);
 	}
 
 	@AfterClass

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

@@ -5,6 +5,7 @@ import static org.junit.Assert.assertNotNull;
 
 import org.junit.Rule;
 import org.junit.Test;
+import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.FanoutExchange;
 import org.springframework.amqp.core.Message;
@@ -39,8 +40,7 @@ public class RabbitBindingIntegrationTests {
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, true, 1, queue.getName());
-				consumer.start();
+				BlockingQueueConsumer consumer = createConsumer(channel);
 				String tag = consumer.getConsumerTag();
 				assertNotNull(tag);
 
@@ -62,6 +62,7 @@ public class RabbitBindingIntegrationTests {
 				return null;
 
 			}
+
 		});
 
 	}
@@ -78,8 +79,7 @@ public class RabbitBindingIntegrationTests {
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, true, 1, queue.getName());
-				consumer.start();
+				BlockingQueueConsumer consumer = createConsumer(channel);
 				String tag = consumer.getConsumerTag();
 				assertNotNull(tag);
 
@@ -122,8 +122,7 @@ public class RabbitBindingIntegrationTests {
 		BlockingQueueConsumer consumer = template.execute(new ChannelCallback<BlockingQueueConsumer>() {
 			public BlockingQueueConsumer doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, true, 1, queue.getName());
-				consumer.start();
+				BlockingQueueConsumer consumer = createConsumer(channel);
 				String tag = consumer.getConsumerTag();
 				assertNotNull(tag);
 
@@ -157,8 +156,7 @@ public class RabbitBindingIntegrationTests {
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, true, 1, queue.getName());
-				consumer.start();
+				BlockingQueueConsumer consumer = createConsumer(channel);
 				String tag = consumer.getConsumerTag();
 				assertNotNull(tag);
 
@@ -178,8 +176,7 @@ public class RabbitBindingIntegrationTests {
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, true, 1, queue.getName());
-				consumer.start();
+				BlockingQueueConsumer consumer = createConsumer(channel);
 				String tag = consumer.getConsumerTag();
 				assertNotNull(tag);
 
@@ -211,8 +208,7 @@ public class RabbitBindingIntegrationTests {
 		template.execute(new ChannelCallback<Void>() {
 			public Void doInRabbit(Channel channel) throws Exception {
 
-				BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, true, 1, queue.getName());
-				consumer.start();
+				BlockingQueueConsumer consumer = createConsumer(channel);
 				String tag = consumer.getConsumerTag();
 				assertNotNull(tag);
 
@@ -231,6 +227,12 @@ public class RabbitBindingIntegrationTests {
 
 	}
 
+	private BlockingQueueConsumer createConsumer(Channel channel) {
+		BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, AcknowledgeMode.AUTO, 1, queue.getName());
+		consumer.start();
+		return consumer;
+	}
+
 	private String getResult(final BlockingQueueConsumer consumer) throws InterruptedException {
 		Message response = consumer.nextMessage(200L);
 		if (response == null) {

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

@@ -93,8 +93,7 @@ public class RabbitTemplateIntegrationTests {
 				}
 			});
 			fail("Expected PlannedException");
-		}
-		catch (UncategorizedAmqpException e) {
+		} catch (UncategorizedAmqpException e) {
 			// TODO: allow client exception to propagate if no AMQP related
 			assertTrue(e.getCause() instanceof PlannedException);
 		}
@@ -143,7 +142,7 @@ public class RabbitTemplateIntegrationTests {
 	public void testReceiveInExternalTransactionAutoAck() throws Exception {
 		template.convertAndSend(ROUTE, "message");
 		// Should just result in auto-ack (not synched with external tx)
-		template.setChannelTransacted(false);
+		template.setChannelTransacted(true);
 		String result = new TransactionTemplate(new TestTransactionManager())
 				.execute(new TransactionCallback<String>() {
 					public String doInTransaction(TransactionStatus status) {
@@ -158,7 +157,7 @@ public class RabbitTemplateIntegrationTests {
 	@Test
 	public void testReceiveInExternalTransactionWithRollback() throws Exception {
 		// Makes receive (and send in principle) transactional
-		template.setChannelTransacted(true); 
+		template.setChannelTransacted(true);
 		template.convertAndSend(ROUTE, "message");
 		try {
 			new TransactionTemplate(new TestTransactionManager()).execute(new TransactionCallback<String>() {
@@ -168,8 +167,7 @@ public class RabbitTemplateIntegrationTests {
 				}
 			});
 			fail("Expected PlannedException");
-		}
-		catch (PlannedException e) {
+		} catch (PlannedException e) {
 			// Expected
 		}
 		String result = (String) template.receiveAndConvert(ROUTE);
@@ -181,7 +179,7 @@ public class RabbitTemplateIntegrationTests {
 	@Test
 	public void testReceiveInExternalTransactionWithNoRollback() throws Exception {
 		// Makes receive non-transactional
-		template.setChannelTransacted(false); 
+		template.setChannelTransacted(false);
 		template.convertAndSend(ROUTE, "message");
 		try {
 			new TransactionTemplate(new TestTransactionManager()).execute(new TransactionCallback<String>() {
@@ -191,8 +189,7 @@ public class RabbitTemplateIntegrationTests {
 				}
 			});
 			fail("Expected PlannedException");
-		}
-		catch (PlannedException e) {
+		} catch (PlannedException e) {
 			// Expected
 		}
 		// No rollback
@@ -226,8 +223,7 @@ public class RabbitTemplateIntegrationTests {
 				}
 			});
 			fail("Expected PlannedException");
-		}
-		catch (PlannedException e) {
+		} catch (PlannedException e) {
 			// Expected
 		}
 		String result = (String) template.receiveAndConvert(ROUTE);

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

@@ -14,6 +14,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.log4j.Level;
 import org.junit.Rule;
 import org.junit.Test;
+import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -32,11 +33,17 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 		public boolean isTransactional() {
 			return this != OFF;
 		}
-		public int getPrefetch() {
-			return this==PREFETCH ? 10 : -1;
+
+		public AcknowledgeMode getAcknowledgeMode() {
+			return this == OFF ? AcknowledgeMode.NONE : AcknowledgeMode.AUTO;
 		}
+
+		public int getPrefetch() {
+			return this == PREFETCH ? 10 : -1;
+		}
+
 		public int getTxSize() {
-			return this==PREFETCH ? 5 : -1;			
+			return this == PREFETCH ? 5 : -1;
 		}
 	}
 
@@ -71,7 +78,8 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 
 	@Rule
 	public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.INFO, RabbitTemplate.class,
-			SimpleMessageListenerContainer.class, BlockingQueueConsumer.class, MessageListenerContainerLifecycleIntegrationTests.class);
+			SimpleMessageListenerContainer.class, BlockingQueueConsumer.class,
+			MessageListenerContainerLifecycleIntegrationTests.class);
 
 	private RabbitTemplate createTemplate(int concurrentConsumers) {
 		RabbitTemplate template = new RabbitTemplate();
@@ -129,10 +137,11 @@ public class MessageListenerContainerLifecycleIntegrationTests {
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(template.getConnectionFactory());
 		PojoListener listener = new PojoListener(latch);
 		container.setMessageListener(new MessageListenerAdapter(listener));
-		container.setChannelTransacted(transactional);
+		container.setAcknowledgeMode(transactionMode.getAcknowledgeMode());
+		container.setChannelTransacted(transactionMode.isTransactional());
 		container.setConcurrentConsumers(concurrentConsumers);
 
-		if (transactionMode.getPrefetch()>0) {			
+		if (transactionMode.getPrefetch() > 0) {
 			container.setPrefetchCount(transactionMode.getPrefetch());
 			container.setTxSize(transactionMode.getTxSize());
 		}

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

@@ -0,0 +1,136 @@
+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 org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.log4j.Level;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+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.Log4jLevelAdjuster;
+
+import com.rabbitmq.client.Channel;
+
+public class MessageListenerManualAckIntegrationTests {
+
+	private static Log logger = LogFactory.getLog(MessageListenerManualAckIntegrationTests.class);
+
+	private Queue queue = new Queue("test.queue");
+
+	private RabbitTemplate template = new RabbitTemplate();
+
+	private int concurrentConsumers = 1;
+
+	private int messageCount = 50;
+
+	private int txSize = 1;
+	
+	private boolean transactional = false;
+	
+	private SimpleMessageListenerContainer container;
+
+	@Rule
+	public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.DEBUG, RabbitTemplate.class,
+			SimpleMessageListenerContainer.class, BlockingQueueConsumer.class);
+
+	@Rule
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue);
+
+	@Before
+	public void createConnectionFactory() {
+		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
+		connectionFactory.setChannelCacheSize(concurrentConsumers);
+		// connectionFactory.setPort(5673);
+		template.setConnectionFactory(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 testListenerWithManualAckNonTransactional() throws Exception {
+		CountDownLatch latch = new CountDownLatch(messageCount);
+		container = createContainer(new TestListener(latch));
+		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 testListenerWithManualAckTransactional() throws Exception {
+		transactional = true;
+		CountDownLatch latch = new CountDownLatch(messageCount);
+		container = createContainer(new TestListener(latch));
+		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()));
+	}
+
+	private SimpleMessageListenerContainer createContainer(Object listener) {
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(template.getConnectionFactory());
+		container.setMessageListener(new MessageListenerAdapter(listener));
+		container.setQueueName(queue.getName());
+		container.setTxSize(txSize);
+		container.setPrefetchCount(txSize);
+		container.setConcurrentConsumers(concurrentConsumers);
+		container.setChannelTransacted(transactional);
+		container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
+		container.afterPropertiesSet();
+		container.start();
+		return container;
+	}
+
+	public static class TestListener implements ChannelAwareMessageListener {
+
+		private final CountDownLatch latch;
+
+		public TestListener(CountDownLatch latch) {
+			this.latch = latch;
+		}
+
+		public void handleMessage(String value) {
+		}
+
+		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);
+			} finally {
+				latch.countDown();
+			}
+		}
+	}
+
+}

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

@@ -20,6 +20,7 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
+import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -36,20 +37,13 @@ public class SimpleMessageListenerContainerIntegrationTests {
 
 	private static Log logger = LogFactory.getLog(SimpleMessageListenerContainerIntegrationTests.class);
 
-	private enum TransactionType {
-		NONE, NATIVE, EXTERNAL;
-		public boolean isTransactional() {
-			return this != NONE;
-		}
-	}
-
 	private Queue queue = new Queue("test.queue");
 
 	private RabbitTemplate template = new RabbitTemplate();
 
 	private final int concurrentConsumers;
 
-	private final TransactionType transactional;
+	private final AcknowledgeMode acknowledgeMode;
 
 	@Rule
 	public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.ERROR, RabbitTemplate.class,
@@ -64,30 +58,45 @@ public class SimpleMessageListenerContainerIntegrationTests {
 
 	private final int txSize;
 
+	private final boolean externalTransaction;
+
 	public SimpleMessageListenerContainerIntegrationTests(int messageCount, int concurrency,
-			TransactionType transacted, int txSize) {
+			AcknowledgeMode acknowledgeMode, int txSize, boolean externalTransaction) {
 		this.messageCount = messageCount;
 		this.concurrentConsumers = concurrency;
-		this.transactional = transacted;
+		this.acknowledgeMode = acknowledgeMode;
 		this.txSize = txSize;
+		this.externalTransaction = externalTransaction;
 	}
 
 	@Parameters
 	public static List<Object[]> getParameters() {
-		return Arrays.asList(params(0, 1, 1, TransactionType.NATIVE), params(1, 1, 1, TransactionType.NONE),
-				params(2, 4, 1, TransactionType.NATIVE), params(3, 4, 1, TransactionType.EXTERNAL),
-				params(4, 2, 2, TransactionType.NATIVE), params(5, 2, 2, TransactionType.NONE),
-				params(6, 20, 4, TransactionType.NATIVE), params(7, 20, 4, TransactionType.NONE),
-				params(8, 1000, 4, TransactionType.NATIVE), params(9, 1000, 4, TransactionType.NONE),
-				params(10, 1000, 4, TransactionType.NATIVE, 10));
+		return Arrays.asList( //
+				params(0, 1, 1, AcknowledgeMode.AUTO), //
+				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) //
+				);
 	}
 
-	private static Object[] params(int i, int messageCount, int concurrency, TransactionType transacted, int txSize) {
-		return new Object[] { messageCount, concurrency, transacted, txSize };
+	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, 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 };
 	}
 
-	private static Object[] params(int i, int messageCount, int concurrency, TransactionType transacted) {
-		return params(i, messageCount, concurrency, transacted, 1);
+	private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode) {
+		return params(i, messageCount, concurrency, acknowledgeMode, 1);
+	}
+
+	private static Object[] extern(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode) {
+		return new Object[] { messageCount, concurrency, acknowledgeMode, 1, true };
 	}
 
 	@Before
@@ -115,7 +124,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 / 100), TimeUnit.SECONDS);
+		boolean waited = latch.await(Math.max(1, messageCount / 50), TimeUnit.SECONDS);
 		assertTrue("Timed out waiting for message", waited);
 		assertNull(template.receiveAndConvert(queue.getName()));
 	}
@@ -124,16 +133,18 @@ public class SimpleMessageListenerContainerIntegrationTests {
 	public void testListenerWithException() throws Exception {
 		CountDownLatch latch = new CountDownLatch(messageCount);
 		container = createContainer(new PojoListener(latch, true));
-		if (transactional.isTransactional()) {
+		if (acknowledgeMode.isTransactionAllowed()) {
 			// Should only need one message if it is going to fail
-			template.convertAndSend(queue.getName(), "foo");
+			for (int i = 0; i < concurrentConsumers; i++) {
+				template.convertAndSend(queue.getName(), i + "foo");
+			}
 		} else {
 			for (int i = 0; i < messageCount; i++) {
 				template.convertAndSend(queue.getName(), i + "foo");
 			}
 		}
 		try {
-			boolean waited = latch.await(Math.max(1, messageCount / 100), TimeUnit.SECONDS);
+			boolean waited = latch.await(5 + Math.max(1, messageCount / 20), TimeUnit.SECONDS);
 			assertTrue("Timed out waiting for message", waited);
 		} finally {
 			// Wait for broker communication to finish before trying to stop
@@ -142,7 +153,7 @@ public class SimpleMessageListenerContainerIntegrationTests {
 			container.shutdown();
 			Thread.sleep(300L);
 		}
-		if (transactional.isTransactional()) {
+		if (acknowledgeMode.isTransactionAllowed()) {
 			assertNotNull(template.receiveAndConvert(queue.getName()));
 		} else {
 			assertNull(template.receiveAndConvert(queue.getName()));
@@ -156,8 +167,10 @@ public class SimpleMessageListenerContainerIntegrationTests {
 		container.setTxSize(txSize);
 		container.setPrefetchCount(txSize);
 		container.setConcurrentConsumers(concurrentConsumers);
-		container.setChannelTransacted(transactional.isTransactional());
-		if (transactional == TransactionType.EXTERNAL) {
+		// For this test always us a transaction if it makes sense...
+		container.setChannelTransacted(acknowledgeMode.isTransactionAllowed());
+		container.setAcknowledgeMode(acknowledgeMode);
+		if (externalTransaction) {
 			container.setTransactionManager(new TestTransactionManager());
 		}
 		container.afterPropertiesSet();

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

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package org.springframework.amqp.rabbit.listener;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
+import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionException;
+import org.springframework.transaction.support.AbstractPlatformTransactionManager;
+import org.springframework.transaction.support.DefaultTransactionStatus;
+
+/**
+ * @author David Syer
+ * 
+ */
+public class SimpleMessageListenerContainerTests {
+
+	@Rule
+	public ExpectedException expectedException = ExpectedException.none();
+
+	@Test
+	public void testInconsistentTransactionConfiguration() throws Exception {
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(new SingleConnectionFactory());
+		container.setMessageListener(new MessageListenerAdapter(this));
+		container.setQueueName("foo");
+		container.setChannelTransacted(false);
+		container.setAcknowledgeMode(AcknowledgeMode.NONE);
+		container.setTransactionManager(new TestTransactionManager());
+		expectedException.expect(IllegalStateException.class);
+		container.afterPropertiesSet();
+	}
+
+	@Test
+	public void testInconsistentAcknowledgeConfiguration() throws Exception {
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(new SingleConnectionFactory());
+		container.setMessageListener(new MessageListenerAdapter(this));
+		container.setQueueName("foo");
+		container.setChannelTransacted(true);
+		container.setAcknowledgeMode(AcknowledgeMode.NONE);
+		expectedException.expect(IllegalStateException.class);
+		container.afterPropertiesSet();
+	}
+
+	@SuppressWarnings("serial")
+	private class TestTransactionManager extends AbstractPlatformTransactionManager {
+
+		@Override
+		protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
+		}
+
+		@Override
+		protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
+		}
+
+		@Override
+		protected Object doGetTransaction() throws TransactionException {
+			return new Object();
+		}
+
+		@Override
+		protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
+		}
+
+	}
+}

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

@@ -46,7 +46,7 @@ public class UnackedRawIntegrationTests {
 	public void init() throws Exception {
 
 		factory.setHost("localhost");
-		factory.setPort(5673);
+		// factory.setPort(5673);
 		conn = factory.newConnection();
 		noTxChannel = conn.createChannel();
 		txChannel = conn.createChannel();

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManagerIntegrationTests.java

@@ -61,8 +61,8 @@ public class RabbitTransactionManagerIntegrationTests {
 
 	@Test
 	public void testReceiveInTransactionWithRollback() throws Exception {
-		template.setChannelTransacted(true); // Makes receive (and send in
-												// principle) transactional
+		// Makes receive (and send in principle) transactional
+		template.setChannelTransacted(true); 
 		template.convertAndSend(ROUTE, "message");
 		try {
 			transactionTemplate.execute(new TransactionCallback<String>() {