+ * 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). + *
+ *+ * 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. + *
+ *+ * 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). + *
+ * + * @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 benull)
@@ -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. 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.
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. + *
+ * 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. + *
+ * 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. - *
By default, only a Spring {@link MessageListener} object or a Spring + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * Subclasses need to implement this method for their specific invoker management process. + *
+ * A shared Rabbit Connection + * @throws Exception + * @see #getSharedConnection() + */ + protected abstract void doInitialize() throws Exception; + + /** + * Close the registered invokers. + *
+ * Subclasses need to implement this method for their specific invoker management process. + *
+ * A shared Rabbit Connection, if any, will automatically be closed afterwards. + * @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. + *
+ * This implementation always returns true; the default 'running' state is purely determined by
+ * {@link #start()} / {@link #stop()}.
+ *
+ * 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 false
+ * 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.
+ *
+ * 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. + *
+ * 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. + *
+ * 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. + *
+ * 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 null)
+ * @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.
Default implementation performs a plain
- * invocation of the onMessage method.
+ * Invoke the specified listener as Spring Rabbit MessageListener.
+ *
+ * Default implementation performs a plain invocation of the onMessage 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.
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. + *
+ * 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.
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. + *
+ * 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); + } + } + } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListeningContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListeningContainer.java deleted file mode 100644 index 9efaeb40..00000000 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListeningContainer.java +++ /dev/null @@ -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. - *
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. - *
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. - *
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. - *
This implementation always returns true; the default 'running'
- * state is purely determined by {@link #start()} / {@link #stop()}.
- *
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
- * false 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.
- *
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. - *
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. - *
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. - *
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 null)
- * @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.
- *
Subclasses need to implement this method for their specific - * invoker management process. - *
A shared Rabbit Connection - * @throws Exception - * @see #getSharedConnection() - */ - protected abstract void doInitialize() throws Exception; - - /** - * Close the registered invokers. - *
Subclasses need to implement this method for their specific - * invoker management process. - *
A shared Rabbit Connection, if any, will automatically be closed - * afterwards. - * @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); - } - } - -} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 7d9b89e8..f2339628 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/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(); } } \ No newline at end of file diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 94f7da74..28c21b14 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/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.
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. + *
+ * 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++) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java index da7310bd..9dcf299e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java +++ b/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. *
Delegates the message to the target listener method, with appropriate
* conversion of the message argument. If the target method returns a
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitAccessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitAccessor.java
index d4ba0be4..72b49335 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitAccessor.java
+++ b/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.
*
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java
index 0eb9ddbc..fdb511aa 100755
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java
+++ b/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
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java
index 9860b5eb..fae46ce5 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java
+++ b/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