AMQP-785: SMLC Lifecycle fixes

JIRA: https://jira.spring.io/browse/AMQP-785

Fixes: #689

- Only stop the container on one thread
- Ignore concurrent stops
- Interrupt consumer threads that are attempting to declare queues
- In `restart()` don't start a new consumer if the container is stopping
- Defer publishing consumer failure events until container is stopped
- Add a RecoveryListener if needed to ensure channels are never recovered
- Fix event publishing for `Error` - it is fatal

__backport to 1.7.x will require work__

Clear the declaring flag when exiting `start()` with exception.

Release the `activeObjectCounter` when interrupted while declaring.

Polishing stopped container lifecycle

Since restarted consumer is not be aware about stopped container,
it can restart properly when RabbitMQ comes back on-line independently
of the container state

* Add `active` flag to the `ActiveObjectCounter` and `deactivate()`
and `isActive()` hooks
* Use `ActiveObjectCounter.deactivate()` in the container shutdown
* Use `ActiveObjectCounter.isActive()` in the
`BlockingQueueConsumer.cancelled()`
* Use `BlockingQueueConsumer.cancelled()` in its `start()` toi check
container activity before performing network job
* Check `isActive()` state in the `AbstractMessageListenerContainer.shutdown()`
* Remove `SimpleMessageListenerContainer.containerStopping` in favor of
`isActive()` hook

Conflicts:
	build.gradle
	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java
	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
Resolved.

Author: garyrussell

build.gradle

@@ -103,9 +103,9 @@ subprojects { subproject ->
 		rabbitmqHttpClientVersion = '1.1.1.RELEASE'
 		slf4jVersion = "1.7.25"
 
-		springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.11.RELEASE'
+		springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.13.RELEASE'
 
-		springRetryVersion = '1.2.0.RELEASE'
+		springRetryVersion = '1.2.1.RELEASE'
 	}
 
 	eclipse {

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

@@ -547,12 +547,17 @@ public void initialize() {
 	 * Stop the shared Connection, call {@link #doShutdown()}, and close this container.
 	 */
 	public void shutdown() {
-		logger.debug("Shutting down Rabbit listener container");
 		synchronized (this.lifecycleMonitor) {
+			if (!isActive()) {
+				logger.info("Shutdown ignored - container is not active already");
+				return;
+			}
 			this.active = false;
 			this.lifecycleMonitor.notifyAll();
 		}
 
+		logger.debug("Shutting down Rabbit listener container");
+
 		// Shut down the invokers.
 		try {
 			doShutdown();
@@ -602,6 +607,9 @@ public final boolean isActive() {
 	 */
 	@Override
 	public void start() {
+		if (isRunning()) {
+			return;
+		}
 		if (!this.initialized) {
 			synchronized (this.lifecycleMonitor) {
 				if (!this.initialized) {

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

@@ -24,13 +24,19 @@
 import java.util.concurrent.TimeUnit;
 
 /**
+ * A mechanism to keep track of active objects.
+ * @param <T> the object type.
+ *
  * @author Dave Syer
+ * @author Artem Bilan
  *
  */
 public class ActiveObjectCounter<T> {
 
 	private final ConcurrentMap<T, CountDownLatch> locks = new ConcurrentHashMap<T, CountDownLatch>();
 
+	private volatile boolean active = true;
+
 	public void add(T object) {
 		CountDownLatch lock = new CountDownLatch(1);
 		this.locks.putIfAbsent(object, lock);
@@ -71,6 +77,15 @@ public int getCount() {
 
 	public void reset() {
 		this.locks.clear();
+		this.active = true;
+	}
+
+	public void deactivate() {
+		this.active = false;
+	}
+
+	public boolean isActive() {
+		return this.active;
 	}
 
 }

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

@@ -64,7 +64,10 @@
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
+import com.rabbitmq.client.Recoverable;
+import com.rabbitmq.client.RecoveryListener;
 import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.impl.recovery.AutorecoveringChannel;
 import com.rabbitmq.utility.Utility;
 
 /**
@@ -79,7 +82,7 @@
  * @author Alex Panchenko
  * @author Johno Crawford
  */
-public class BlockingQueueConsumer {
+public class BlockingQueueConsumer implements RecoveryListener {
 
 	private static Log logger = LogFactory.getLog(BlockingQueueConsumer.class);
 
@@ -150,6 +153,10 @@ public class BlockingQueueConsumer {
 
 	private volatile boolean normalCancel;
 
+	volatile Thread thread;
+
+	volatile boolean declaring;
+
 	/**
 	 * Create a consumer. The consumer must not attempt to use
 	 * the connection factory or communicate with the broker
@@ -424,7 +431,8 @@ protected boolean hasDelivery() {
 
 	protected boolean cancelled() {
 		return this.cancelled.get() || (this.abortStarted > 0 &&
-				this.abortStarted + this.shutdownTimeout > System.currentTimeMillis());
+				this.abortStarted + this.shutdownTimeout > System.currentTimeMillis())
+				|| !this.activeObjectCounter.isActive();
 	}
 
 	/**
@@ -559,10 +567,14 @@ public void start() throws AmqpException {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Starting consumer " + this);
 		}
+
+		this.thread = Thread.currentThread();
+
 		try {
 			this.resourceHolder = ConnectionFactoryUtils.getTransactionalResourceHolder(this.connectionFactory,
 					this.transactional);
 			this.channel = this.resourceHolder.getChannel();
+			addRecoveryListener();
 		}
 		catch (AmqpAuthenticationException e) {
 			throw new FatalListenerStartupException("Authentication failure", e);
@@ -573,7 +585,11 @@ public void start() throws AmqpException {
 
 		// mirrored queue might be being moved
 		int passiveDeclareRetries = this.declarationRetries;
+		this.declaring = true;
 		do {
+			if (cancelled()) {
+				break;
+			}
 			try {
 				attemptPassiveDeclarations();
 				if (passiveDeclareRetries < this.declarationRetries && logger.isInfoEnabled()) {
@@ -589,7 +605,10 @@ public void start() throws AmqpException {
 							Thread.sleep(this.failedDeclarationRetryInterval);
 						}
 						catch (InterruptedException e1) {
+							this.declaring = false;
 							Thread.currentThread().interrupt();
+							this.activeObjectCounter.release(this);
+							throw RabbitExceptionTranslator.convertRabbitAccessException(e1);
 						}
 					}
 				}
@@ -602,15 +621,17 @@ else if (e.getFailedQueues().size() < this.queues.length) {
 					this.lastRetryDeclaration = System.currentTimeMillis();
 				}
 				else {
+					this.declaring = false;
 					this.activeObjectCounter.release(this);
 					throw new QueuesNotAvailableException("Cannot prepare queue for listener. "
 							+ "Either the queue doesn't exist or the broker will not allow us to use it.", e);
 				}
 			}
 		}
-		while (passiveDeclareRetries-- > 0);
+		while (passiveDeclareRetries-- > 0 && !cancelled());
+		this.declaring = false;
 
-		if (!this.acknowledgeMode.isAutoAck()) {
+		if (!this.acknowledgeMode.isAutoAck() && !cancelled()) {
 			// Set basicQos before calling basicConsume (otherwise if we are not acking the broker
 			// will send blocks of 100 messages)
 			try {
@@ -624,9 +645,11 @@ else if (e.getFailedQueues().size() < this.queues.length) {
 
 
 		try {
-			for (String queueName : this.queues) {
-				if (!this.missingQueues.contains(queueName)) {
-					consumeFromQueue(queueName);
+			if (!cancelled()) {
+				for (String queueName : this.queues) {
+					if (!this.missingQueues.contains(queueName)) {
+						consumeFromQueue(queueName);
+					}
 				}
 			}
 		}
@@ -635,6 +658,19 @@ else if (e.getFailedQueues().size() < this.queues.length) {
 		}
 	}
 
+	/**
+	 * Add a listener if necessary so we can immediately close an autorecovered
+	 * channel if necessary since the async consumer will no longer exist.
+	 */
+	private void addRecoveryListener() {
+		if (this.channel instanceof ChannelProxy) {
+			if (((ChannelProxy) this.channel).getTargetChannel() instanceof AutorecoveringChannel) {
+				((AutorecoveringChannel) ((ChannelProxy) this.channel).getTargetChannel())
+						.addRecoveryListener(this);
+			}
+		}
+	}
+
 	private void consumeFromQueue(String queue) throws IOException {
 		String consumerTag = this.channel.basicConsume(queue, this.acknowledgeMode.isAutoAck(),
 				(this.tagStrategy != null ? this.tagStrategy.createConsumerTag(queue) : ""), this.noLocal, this.exclusive,
@@ -810,6 +846,28 @@ public boolean commitIfNecessary(boolean locallyTransacted) throws IOException {
 
 	}
 
+	@Override
+	public void handleRecovery(Recoverable recoverable) {
+		// should never get here
+		handleRecoveryStarted(recoverable);
+	}
+
+	@Override
+	public void handleRecoveryStarted(Recoverable recoverable) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Closing an autorecovered channel: " + recoverable);
+		}
+		try {
+			((Channel) recoverable).close();
+		}
+		catch (IOException e) {
+			logger.debug("Error closing an autorecovered channel");
+		}
+		catch (TimeoutException e) {
+			logger.debug("Error closing an autorecovered channel");
+		}
+	}
+
 	@Override
 	public String toString() {
 		return "Consumer@" + ObjectUtils.getIdentityHexString(this) + ": "