Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2461: RabbitListenerErrorHandler with Async

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/2461
This commit is contained in:
Gary Russell
2023-05-25 13:12:19 -04:00
committed by GitHub
parent e452af157e
commit e492b98bf8
5 changed files with 28 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
View File

@@ -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,

16
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java
View File

@@ -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

4
spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt
View File

@@ -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")
}

3
src/reference/asciidoc/amqp.adoc
View File

@@ -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 <<annotation-error-handling>> for more information about this error handler and its purpose.
[[threading]]
===== Threading and Asynchronous Consumers

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -48,9 +48,12 @@ See <<Jackson2JsonMessageConverter-from-message>> for more information
You can now configure a `ReplyPostProcessor` via the container factory rather than via a property on `@RabbitListener`.
See <<async-annotation-driven-reply>> 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 <<async-returns>> 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`.