+ * It is parsed in {@link Address} so should be of the form exchange/rk. + *
* It can be a string surrounded by "!{...}" in which case the expression is * evaluated at runtime; see the reference manual for more information. - * @param defaultReplyTo The exchange. + * @param defaultReplyTo The replyTo address. * @since 1.6 */ public void setResponseAddress(String defaultReplyTo) { 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 bd9c9191..094fe969 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 @@ -161,8 +161,11 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis } Object errorResult = this.errorHandler.handleError(amqpMessage, messageWithChannel, e); if (errorResult != null) { - handleResult(this.handlerAdapter.getInvocationResultFor(errorResult, message.getPayload()), - amqpMessage, channel, message); + Object payload = message == null ? null : message.getPayload(); + InvocationResult invResult = payload == null + ? new InvocationResult(errorResult, null, null, null, null) + : this.handlerAdapter.getInvocationResultFor(errorResult, payload); + handleResult(invResult, amqpMessage, channel, message); } else { logger.trace("Error handler returned no result"); @@ -203,15 +206,17 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis if (!this.returnExceptions) { throw exceptionToThrow; } + Object payload = message == null ? null : message.getPayload(); try { + handleResult(new InvocationResult(new RemoteInvocationResult(throwableToReturn), null, - this.handlerAdapter.getReturnTypeFor(message.getPayload()), + payload == null ? Object.class : this.handlerAdapter.getReturnTypeFor(payload), this.handlerAdapter.getBean(), - this.handlerAdapter.getMethodFor(message.getPayload())), + payload == null ? null : this.handlerAdapter.getMethodFor(payload)), amqpMessage, channel, message); } catch (ReplyFailureException rfe) { - if (void.class.equals(this.handlerAdapter.getReturnTypeFor(message.getPayload()))) { + if (payload == null || void.class.equals(this.handlerAdapter.getReturnTypeFor(payload))) { throw exceptionToThrow; } else { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java index cdd4d528..a268fb29 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java @@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.listener.adapter; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; @@ -293,7 +294,7 @@ public class MessagingMessageListenerAdapterTests { return null; } - }, String.class); + }, false, String.class); listener.setMessageConverter(new MessageConverter() { @Override @@ -312,15 +313,71 @@ public class MessagingMessageListenerAdapterTests { assertThat(ehCalled.get()).isTrue(); } + @Test + void errorHandlerAfterConversionExWithResult() throws Exception { + org.springframework.amqp.core.Message message = MessageTestUtils.createTextMessage("foo"); + Channel channel = mock(Channel.class); + AtomicBoolean ehCalled = new AtomicBoolean(); + MessagingMessageListenerAdapter listener = getSimpleInstance("fail", + new RabbitListenerErrorHandler() { + + @Override + public Object handleError(org.springframework.amqp.core.Message amqpMessage, Message message, + ListenerExecutionFailedException exception) throws Exception { + + ehCalled.set(true); + return "foo"; + } + + }, false, String.class); + listener.setMessageConverter(new MessageConverter() { + + @Override + public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties) + throws MessageConversionException { + + return new org.springframework.amqp.core.Message(((String) object).getBytes(), messageProperties); + } + + @Override + public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException { + throw new MessageConversionException("test"); + } + }); + listener.setResponseAddress("foo/bar"); + listener.onMessage(message, channel); + verify(channel).basicPublish(any(), any(), anyBoolean(), any(), any()); + assertThat(ehCalled.get()).isTrue(); + } + + @Test + void errorHandlerAfterConversionExWithReturnEx() throws Exception { + org.springframework.amqp.core.Message message = MessageTestUtils.createTextMessage("foo"); + Channel channel = mock(Channel.class); + MessagingMessageListenerAdapter listener = getSimpleInstance("fail", null, true, String.class); + class MC extends SimpleMessageConverter { + + @Override + public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException { + throw new MessageConversionException("test"); + } + + } + listener.setMessageConverter(new MC()); + listener.setResponseAddress("foo/bar"); + listener.onMessage(message, channel); + verify(channel).basicPublish(any(), any(), anyBoolean(), any(), any()); + } + protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, Class... parameterTypes) { - return getSimpleInstance(methodName, null, parameterTypes); + return getSimpleInstance(methodName, null, false, parameterTypes); } protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, RabbitListenerErrorHandler eh, - Class... parameterTypes) { + boolean returnEx, Class... parameterTypes) { Method m = ReflectionUtils.findMethod(SampleBean.class, methodName, parameterTypes); - return createInstance(m, false, eh); + return createInstance(m, returnEx, eh); } protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, boolean returnExceptions,