From e492b98bf875b885c2db3fc4dbb4154a8a6b4fed Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 25 May 2023 13:12:19 -0400 Subject: [PATCH] GH-2461: RabbitListenerErrorHandler with Async Resolves https://github.com/spring-projects/spring-amqp/issues/2461 --- .../AbstractAdaptableMessageListener.java | 8 ++++---- .../adapter/MessagingMessageListenerAdapter.java | 16 +++++++++++++++- .../rabbit/annotation/EnableRabbitKotlinTests.kt | 4 ++-- src/reference/asciidoc/amqp.adoc | 3 +++ src/reference/asciidoc/whats-new.adoc | 5 ++++- 5 files changed, 28 insertions(+), 8 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java index 6212b040..5f7215c4 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2022 the original author or authors. + * Copyright 2014-2023 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. @@ -385,7 +385,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe basicAck(request, channel); } else { - asyncFailure(request, channel, t); + asyncFailure(request, channel, t, source); } }); } @@ -396,7 +396,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe } MonoHandler.subscribe(resultArg.getReturnValue(), r -> asyncSuccess(resultArg, request, channel, source, r), - t -> asyncFailure(request, channel, t), + t -> asyncFailure(request, channel, t, source), () -> basicAck(request, channel)); } else { @@ -447,7 +447,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe } } - private void asyncFailure(Message request, Channel channel, Throwable t) { + protected void asyncFailure(Message request, Channel channel, Throwable t, Object source) { this.logger.error("Future, Mono, or suspend function was completed with an exception for " + request, t); try { channel.basicNack(request.getMessageProperties().getDeliveryTag(), false, diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java index d38bab75..473d2b5b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -160,6 +160,20 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis } } + @Override + protected void asyncFailure(org.springframework.amqp.core.Message request, Channel channel, Throwable t, + Object source) { + + try { + handleException(request, channel, (Message) source, + new ListenerExecutionFailedException("Async Fail", t, request)); + return; + } + catch (Exception ex) { + } + super.asyncFailure(request, channel, t, source); + } + private void handleException(org.springframework.amqp.core.Message amqpMessage, Channel channel, @Nullable Message message, ListenerExecutionFailedException e) throws Exception { // NOSONAR diff --git a/spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt b/spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt index 981b2848..2fc695fa 100644 --- a/spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt +++ b/spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2018-2023 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. @@ -126,7 +126,7 @@ class EnableRabbitKotlinTests { open class Multi { @RabbitHandler - fun handle(@Suppress("UNUSED_PARAMETER") data: String) { + suspend fun handle(@Suppress("UNUSED_PARAMETER") data: String) { throw RuntimeException("fail") } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 75d0ae1a..3a8727a3 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -3664,6 +3664,9 @@ Starting with version 3.0.5, the `@RabbitListener` (and `@RabbitHandler`) method All the mentioned rules about `AcknowledgeMode.MANUAL` are still apply. The `org.jetbrains.kotlinx:kotlinx-coroutines-reactor` dependency must be present in classpath to allow `suspend` function invocations. +Also starting with version 3.0.5, if a `RabbitListenerErrorHandler` is configured on a listener with an async return type (including Kotlin suspend functions), the error handler is invoked after a failure. +See <> for more information about this error handler and its purpose. + [[threading]] ===== Threading and Asynchronous Consumers diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index cb6d4ae3..d386714b 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -48,9 +48,12 @@ See <> for more information You can now configure a `ReplyPostProcessor` via the container factory rather than via a property on `@RabbitListener`. See <> for more information. -The `@RabbitListener` (and `@RabbitHandler`) methods can now be as a Kotlin `suspend` functions. +The `@RabbitListener` (and `@RabbitHandler`) methods can now be declared as Kotlin `suspend` functions. See <> for more information. +Starting with version 3.0.5, listeners with async return types (including Kotlin suspend functions) invoke the `RabbitListenerErrorHandler` (if configured) after a failure. +Previously, the error handler was only invoked with synchronous invocations. + ==== Connection Factory Changes The default `addressShuffleMode` in `AbstractConnectionFactory` is now `RANDOM`.