GH-1021: Fix @SendTo after error is handled
Fixes https://github.com/spring-projects/spring-amqp/issues/1021 Sending the result from a `RabbitListenerErrorHandler` was broken for class-level `@RabbitListener` because the send to expression was lost. **cherry-pick to 2.1.x** * * Also capture the generic return type after the error is handled
This commit is contained in:
committed by
Artem Bilan
parent
cf030af431
commit
3e451e62e1
@@ -272,4 +272,14 @@ public class DelegatingInvocableHandler {
|
||||
return this.defaultHandler != null;
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public InvocationResult getInvocationResultFor(Object result, Object inboundPayload) {
|
||||
InvocableHandlerMethod handler = findHandlerForPayload(inboundPayload.getClass());
|
||||
if (handler != null) {
|
||||
return new InvocationResult(result, this.handlerSendTo.get(handler),
|
||||
handler.getMethod().getGenericReturnType());
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
|
||||
package org.springframework.amqp.rabbit.listener.adapter;
|
||||
|
||||
import org.springframework.lang.Nullable;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
|
||||
|
||||
@@ -94,5 +95,14 @@ public class HandlerAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public InvocationResult getInvocationResultFor(Object result, Object inboundPayload) {
|
||||
if (this.invokerHandlerMethod != null) {
|
||||
return new InvocationResult(result, null, this.invokerHandlerMethod.getMethod().getGenericReturnType());
|
||||
}
|
||||
else {
|
||||
return this.delegatingHandler.getInvocationResultFor(result, inboundPayload);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -149,7 +149,8 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
|
||||
.build();
|
||||
Object errorResult = this.errorHandler.handleError(amqpMessage, message, e);
|
||||
if (errorResult != null) {
|
||||
handleResult(new InvocationResult(errorResult, null, null), amqpMessage, channel, message);
|
||||
handleResult(this.handlerAdapter.getInvocationResultFor(errorResult, message.getPayload()),
|
||||
amqpMessage, channel, message);
|
||||
}
|
||||
else {
|
||||
logger.trace("Error handler returned no result");
|
||||
|
||||
@@ -345,6 +345,11 @@ public class EnableRabbitIntegrationTests {
|
||||
rabbitTemplate.convertAndSend("multi.exch", "multi.rk", bar);
|
||||
rabbitTemplate.setReceiveTimeout(10000);
|
||||
assertThat(this.rabbitTemplate.receiveAndConvert("sendTo.replies")).isEqualTo("BAR: bar");
|
||||
bar.field = "crash";
|
||||
rabbitTemplate.convertAndSend("multi.exch", "multi.rk", bar);
|
||||
assertThat(this.rabbitTemplate.receiveAndConvert("sendTo.replies"))
|
||||
.isEqualTo("CRASHCRASH Test reply from error handler");
|
||||
bar.field = "bar";
|
||||
Baz baz = new Baz();
|
||||
baz.field = "baz";
|
||||
assertThat(rabbitTemplate.convertSendAndReceive("multi.exch", "multi.rk", baz)).isEqualTo("BAZ: baz");
|
||||
@@ -1550,14 +1555,22 @@ public class EnableRabbitIntegrationTests {
|
||||
|
||||
@Bean
|
||||
public RabbitListenerErrorHandler alwaysBARHandler() {
|
||||
return (m, sm, e) -> "BAR";
|
||||
return (msg, springMsg, ex) -> "BAR";
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RabbitListenerErrorHandler upcaseAndRepeatErrorHandler() {
|
||||
return (msg, springMsg, ex) -> {
|
||||
String payload = ((Bar) springMsg.getPayload()).field.toUpperCase();
|
||||
return payload + payload + " " + ex.getCause().getMessage();
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RabbitListenerErrorHandler throwANewException() {
|
||||
return (m, sm, e) -> {
|
||||
this.errorHandlerChannel = sm.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
|
||||
throw new RuntimeException("from error handler", e.getCause());
|
||||
return (msg, springMsg, ex) -> {
|
||||
this.errorHandlerChannel = springMsg.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
|
||||
throw new RuntimeException("from error handler", ex.getCause());
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1637,7 +1650,7 @@ public class EnableRabbitIntegrationTests {
|
||||
|
||||
@Bean
|
||||
public org.springframework.amqp.core.Queue sendToReplies() {
|
||||
return new org.springframework.amqp.core.Queue(sendToRepliesBean(), false, false, true);
|
||||
return new org.springframework.amqp.core.Queue(sendToRepliesBean(), false, false, false);
|
||||
}
|
||||
|
||||
@Bean
|
||||
@@ -1672,12 +1685,15 @@ public class EnableRabbitIntegrationTests {
|
||||
@RabbitListener(bindings = @QueueBinding
|
||||
(value = @Queue,
|
||||
exchange = @Exchange(value = "multi.exch", autoDelete = "true"),
|
||||
key = "multi.rk"))
|
||||
key = "multi.rk"), errorHandler = "upcaseAndRepeatErrorHandler")
|
||||
static class MultiListenerBean {
|
||||
|
||||
@RabbitHandler
|
||||
@SendTo("${foo.bar:#{sendToRepliesBean}}")
|
||||
public String bar(@NonNull Bar bar) {
|
||||
if (bar.field.equals("crash")) {
|
||||
throw new RuntimeException("Test reply from error handler");
|
||||
}
|
||||
return "BAR: " + bar.field;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user