AMQP-776: More Consumer Events

JIRA: https://jira.spring.io/browse/AMQP-776
JIRA: https://jira.spring.io/browse/AMQP-777
JIRA: https://jira.spring.io/browse/AMQP-782

Publish an event when a consumer successfully consumes from a queue.
Publish an event when an SMLC listener throws an `Error`.
Doc polishing.

Update minimum client version in docs; remove reference to broker version
since that's no longer linked to the client.

__cherry-pick to 1.7.x (minus DMLC change)__

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java
#	src/reference/asciidoc/quick-tour.adoc

* Rework `SimpleMessageListenerContainerIntegration2Tests` do not use
lambda for the `ApplicationEventPublisher` since it is there since
Spring 5 only

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

@@ -54,6 +54,7 @@ import org.springframework.amqp.rabbit.support.Delivery;
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.amqp.support.ConsumerTagStrategy;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.backoff.BackOffExecution;
@@ -143,6 +144,8 @@ public class BlockingQueueConsumer {
 
 	private boolean locallyTransacted;
 
+	private ApplicationEventPublisher applicationEventPublisher;
+
 	private volatile long abortStarted;
 
 	private volatile boolean normalCancel;
@@ -354,6 +357,10 @@ public class BlockingQueueConsumer {
 		this.locallyTransacted = locallyTransacted;
 	}
 
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
 	/**
 	 * Clear the delivery tags when rolling back with an external transaction
 	 * manager.
@@ -641,6 +648,9 @@ public class BlockingQueueConsumer {
 		else {
 			logger.error("Null consumer tag received for queue " + queue);
 		}
+		if (this.applicationEventPublisher != null) {
+			this.applicationEventPublisher.publishEvent(new ConsumeOkEvent(this, queue, consumerTag));
+		}
 	}
 
 	private void attemptPassiveDeclarations() {

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2017 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.event.AmqpEvent;
+
+/**
+ * @author Gary Russell
+ * @since 1.7.5
+ *
+ */
+@SuppressWarnings("serial")
+public class ConsumeOkEvent extends AmqpEvent {
+
+	private final String queue;
+
+	private final String consumerTag;
+
+	public ConsumeOkEvent(Object source, String queue, String consumerTag) {
+		super(source);
+		this.queue = queue;
+		this.consumerTag = consumerTag;
+	}
+
+	@Override
+	public String toString() {
+		return "ConsumeOkEvent [queue=" + this.queue + ", consumerTag=" + this.consumerTag
+				+ ", consumer=" + getSource() + "]";
+	}
+
+}

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

@@ -1120,6 +1120,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		}
 		consumer.setBackOffExecution(this.recoveryBackOff.start());
 		consumer.setShutdownTimeout(this.shutdownTimeout);
+		consumer.setApplicationEventPublisher(this.applicationEventPublisher);
 		return consumer;
 	}
 
@@ -1608,6 +1609,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 			catch (Error e) { //NOSONAR
 				// ok to catch Error - we're aborting so will stop
 				logger.error("Consumer thread error, thread abort.", e);
+				logConsumerException(e);
 				aborted = true;
 			}
 			catch (Throwable t) { //NOSONAR

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

@@ -280,10 +280,15 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 		assertNull(template.receiveAndConvert(queue.getName()));
 		container.stop();
 		assertTrue(eventLatch.await(10, TimeUnit.SECONDS));
+		assertThat(events.size(), equalTo(8));
 		assertThat(events.get(0), instanceOf(AsyncConsumerStartedEvent.class));
-		assertSame(events.get(1), eventRef.get());
-		assertThat(events.get(2), instanceOf(AsyncConsumerRestartedEvent.class));
-		assertThat(events.get(3), instanceOf(AsyncConsumerStoppedEvent.class));
+		assertThat(events.get(1), instanceOf(ConsumeOkEvent.class));
+		assertThat(events.get(2), instanceOf(ConsumeOkEvent.class));
+		assertSame(events.get(3), eventRef.get());
+		assertThat(events.get(4), instanceOf(AsyncConsumerRestartedEvent.class));
+		assertThat(events.get(5), instanceOf(ConsumeOkEvent.class));
+		assertThat(events.get(6), instanceOf(ConsumeOkEvent.class));
+		assertThat(events.get(7), instanceOf(AsyncConsumerStoppedEvent.class));
 	}
 
 	@Test
@@ -332,13 +337,25 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 		context.refresh();
 		container1.setApplicationContext(context);
 		container1.setExclusive(true);
+		final CountDownLatch consumeLatch1 = new CountDownLatch(1);
+		container1.setApplicationEventPublisher(new ApplicationEventPublisher() {
+
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				if (event instanceof ConsumeOkEvent) {
+					consumeLatch1.countDown();
+				}
+			}
+
+			@Override
+			public void publishEvent(Object event) {
+
+			}
+
+		});
 		container1.afterPropertiesSet();
 		container1.start();
-		int n = 0;
-		while (n++ < 100 && container1.getActiveConsumerCount() < 1) {
-			Thread.sleep(100);
-		}
-		assertTrue(n < 100);
+		assertTrue(consumeLatch1.await(10, TimeUnit.SECONDS));
 		CountDownLatch latch2 = new CountDownLatch(1000);
 		SimpleMessageListenerContainer container2 = new SimpleMessageListenerContainer(template.getConnectionFactory());
 		container2.setMessageListener(new MessageListenerAdapter(new PojoListener(latch2)));
@@ -347,18 +364,22 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 		container2.setRecoveryInterval(1000);
 		container2.setExclusive(true); // not really necessary, but likely people will make all consumers exclusive.
 		final AtomicReference<ListenerContainerConsumerFailedEvent> eventRef = new AtomicReference<>();
+		final CountDownLatch consumeLatch2 = new CountDownLatch(1);
 		container2.setApplicationEventPublisher(new ApplicationEventPublisher() {
 
-			@Override
-			public void publishEvent(Object event) {
-				//NOSONAR
-			}
-
 			@Override
 			public void publishEvent(ApplicationEvent event) {
 				if (event instanceof ListenerContainerConsumerFailedEvent) {
 					eventRef.set((ListenerContainerConsumerFailedEvent) event);
 				}
+				else if (event instanceof ConsumeOkEvent) {
+					consumeLatch2.countDown();
+				}
+			}
+
+			@Override
+			public void publishEvent(Object event) {
+
 			}
 
 		});
@@ -374,6 +395,7 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 		assertEquals(1000, latch2.getCount());
 		container1.stop();
 		// container 2 should recover and process the next batch of messages
+		assertTrue(consumeLatch2.await(10, TimeUnit.SECONDS));
 		for (int i = 0; i < 1000; i++) {
 			template.convertAndSend(queue.getName(), i + "foo");
 		}
@@ -596,6 +618,38 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 				+ "executor have enough threads to support the container concurrency?"));
 	}
 
+	@Test
+	public void testErrorStopsContainer() throws Exception {
+		this.container = createContainer((MessageListener) (m) -> {
+			throw new Error("testError");
+		}, false, this.queue.getName());
+		final CountDownLatch latch = new CountDownLatch(1);
+		this.container.setApplicationEventPublisher(new ApplicationEventPublisher() {
+
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				if (event instanceof ListenerContainerConsumerFailedEvent) {
+					latch.countDown();
+				}
+			}
+
+			@Override
+			public void publishEvent(Object event) {
+
+			}
+
+		});
+		this.container.setDefaultRequeueRejected(false);
+		this.container.start();
+		this.template.convertAndSend(this.queue.getName(), "foo");
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		int n = 0;
+		while (n++ < 100 && this.container.isRunning()) {
+			Thread.sleep(100);
+		}
+		assertFalse(this.container.isRunning());
+	}
+
 	private boolean containerStoppedForAbortWithBadListener() throws InterruptedException {
 		Log logger = spy(TestUtils.getPropertyValue(container, "logger", Log.class));
 		new DirectFieldAccessor(container).setPropertyValue("logger", logger);

src/reference/asciidoc/amqp.adoc

@@ -1382,7 +1382,7 @@ Batched messages are automatically de-batched by listener containers (using the
 See <<template-batching>> for more information about batching.
 
 [[consumer-events]]
-===== Consumer Failure Events
+===== Consumer Events
 
 Starting with _version 1.5_, the `SimpleMessageListenerContainer` publishes application events whenever a listener
 (consumer) experiences a failure of some kind.
@@ -1406,6 +1406,15 @@ See also <<channel-close-logging>>.
 
 Fatal errors are always logged at `ERROR` level; this it not modifiable.
 
+Several other events are published at various stages of the container lifecycle:
+
+- `AsyncConsumerStartedEvent` (when the consumer is started)
+- `AsyncConsumerRestartedEvent` (when the consumer is restarted after a failure - `SimpleMessageListenerContainer` only)
+- `AsyncConsumerTerminatedEvent` (when a consumer is stopped normally)
+- `AsyncConsumerStoppedEvent` (when the consumer is stopped - `SimpleMessageListenerContainer` only)
+- `ConsumeOkEvent` (when a `consumeOk` is received from the broker, contains the queue name and `consumerTag`)
+- `ListenerContainerIdleEvent` (see <<idle-containers>>)
+
 [[consumerTags]]
 ===== Consumer Tags
 

src/reference/asciidoc/quick-tour.adoc

@@ -34,8 +34,6 @@ Annotation-based listeners and the `RabbitMessagingTemplate` require Spring Fram
 
 The minimum `amqp-client` java client library version is 4.0.0.
 
-Note the this refers to the java client library; generally, it will work with older broker versions.
-
 ===== Very, Very Quick
 
 Using plain, imperative Java to send and receive a message: