diff --git a/.travis.yml b/.travis.yml index 9cb2eed8..fdc0e1b9 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,3 +1,4 @@ +dist: trusty language: java jdk: oraclejdk8 sudo: false diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java index 2563e44d..705c1e31 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java @@ -396,7 +396,8 @@ public abstract class AbstractRabbitListenerContainerFactory args = new HashMap<>(); + args.put("x-dead-letter-exchange", ""); + args.put("x-dead-letter-routing-key", queue5DLQ().getName()); + return new AnonymousQueue(args); + } + + @Bean + public Queue queue5DLQ() { + return new AnonymousQueue(); + } + + @Bean + public Queue queue6() { + Map args = new HashMap<>(); + args.put("x-dead-letter-exchange", ""); + args.put("x-dead-letter-routing-key", queue6DLQ().getName()); + return new AnonymousQueue(args); + } + + @Bean + public Queue queue6DLQ() { + return new AnonymousQueue(); + } + + @Bean + public Queue queue7() { + return new AnonymousQueue(); + } + @Bean public Listener listener() { return new Listener(); @@ -196,6 +264,12 @@ public class AsyncListenerTests { private final CountDownLatch latch4 = new CountDownLatch(1); + private final CountDownLatch latch5 = new CountDownLatch(1); + + private final CountDownLatch latch6 = new CountDownLatch(1); + + private final AtomicBoolean first7 = new AtomicBoolean(true); + @RabbitListener(id = "foo", queues = "#{queue1.name}") public ListenableFuture listen1(String foo) { SettableListenableFuture future = new SettableListenableFuture<>(); @@ -231,6 +305,43 @@ public class AsyncListenerTests { return future; } + @RabbitListener(id = "fiz", queues = "#{queue5.name}") + public ListenableFuture listen5(@SuppressWarnings("unused") String foo) { + SettableListenableFuture future = new SettableListenableFuture<>(); + future.setException(new AmqpRejectAndDontRequeueException("asyncToDLQ")); + return future; + } + + @RabbitListener(id = "buz", queues = "#{queue5DLQ.name}") + public void listen5DLQ(@SuppressWarnings("unused") String foo) { + this.latch5.countDown(); + } + + @RabbitListener(id = "fix", queues = "#{queue6.name}", containerFactory = "dontRequeueFactory") + public ListenableFuture listen6(@SuppressWarnings("unused") String foo) { + SettableListenableFuture future = new SettableListenableFuture<>(); + future.setException(new IllegalStateException("asyncDefaultToDLQ")); + return future; + } + + @RabbitListener(id = "fox", queues = "#{queue6DLQ.name}") + public void listen6DLQ(@SuppressWarnings("unused") String foo) { + this.latch6.countDown(); + } + + @RabbitListener(id = "overrideFactoryRequeue", queues = "#{queue7.name}", + containerFactory = "dontRequeueFactory") + public ListenableFuture listen7(@SuppressWarnings("unused") String foo) { + SettableListenableFuture future = new SettableListenableFuture<>(); + if (this.first7.compareAndSet(true, false)) { + future.setException(new ImmediateRequeueAmqpException("asyncOverrideDefaultToDLQ")); + } + else { + future.set("listen7"); + } + return future; + } + } } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index ea88d918..dfcaba81 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -2834,8 +2834,9 @@ Starting with version 2.1, `@RabbitListener` (and `@RabbitHandler`) methods can IMPORTANT: The listener container factory must be configured with `AcknowledgeMode.MANUAL` so that the consumer thread will not ack the message; instead, the asynchronous completion will ack or nack the message when the async operation completes. When the async result is completed with an error, whether the message is requeued or not depends on the exception type thrown, the container configuration, and the container error handler. -By default, the message will be requeued, unless the container's `defaultRequeueRejected` property is set to `false`. +By default, the message will be requeued, unless the container's `defaultRequeueRejected` property is set to `false` (it is `true` by default). If the async result is completed with an `AmqpRejectAndDontRequeueException`, the message will not be requeued. +If the container's `defaultRequeueRejected` property is `false`, you can override that by setting the future's exception to a `ImmediateRequeueException` and the message will be requeued. If some exception occurs within the listener method that prevents creation of the async result object, you MUST catch that exception and return an appropriate return object that will cause the message to be acknowledged or requeued. [[threading]] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index d7988576..6ac9b0ff 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -86,7 +86,7 @@ See <> for more information. ===== Async `@RabbitListener` Return `@RabbitListener` methods can now return `ListenableFuture` or `Mono`. -See <> for more information. +See <> for more information. ===== Connection Factory Bean Changes