From 3e451e62e13125cf2fcae94a3ebc17e66e3441bb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 14 Jun 2019 11:09:23 -0400 Subject: [PATCH] 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 --- .../adapter/DelegatingInvocableHandler.java | 10 +++++++ .../listener/adapter/HandlerAdapter.java | 10 +++++++ .../MessagingMessageListenerAdapter.java | 3 +- .../EnableRabbitIntegrationTests.java | 28 +++++++++++++++---- 4 files changed, 44 insertions(+), 7 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java index 045c1dc5..5edc740a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java @@ -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; + } + } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java index 2aec72ae..ae2cd65e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java @@ -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); + } + } } 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 aeb982b6..3a9a1827 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 @@ -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"); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index 63ffd507..e1a67214 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -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; }