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 b8b55686..a38ee502 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 @@ -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() { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConsumeOkEvent.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConsumeOkEvent.java new file mode 100644 index 00000000..3d277d13 --- /dev/null +++ b/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() + "]"; + } + +} 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 e852550a..7905b90d 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 @@ -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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index 74579d76..0e737b86 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/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 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); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 7825aaf8..969bb34f 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -1382,7 +1382,7 @@ Batched messages are automatically de-batched by listener containers (using the See <> 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 <>. 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 <>) + [[consumerTags]] ===== Consumer Tags diff --git a/src/reference/asciidoc/quick-tour.adoc b/src/reference/asciidoc/quick-tour.adoc index 8f113dba..e4b5c3eb 100644 --- a/src/reference/asciidoc/quick-tour.adoc +++ b/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: