From 96070f6720192fddbff5d90cf864f8fd57ae76fc Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 7 Jun 2021 16:05:14 -0400 Subject: [PATCH] GH-1339: Fix RLErrorHandler with Conversion Ex. (#1346) Resolves https://github.com/spring-projects/spring-amqp/issues/1339 Error handler was not called for conversion exceptions, preventing the application from returning some error to the caller for request/reply processing. **cherry-pick to 2.2.x** --- .../MessagingMessageListenerAdapter.java | 91 +++++++++++-------- .../api/RabbitListenerErrorHandler.java | 7 +- .../MessagingMessageListenerAdapterTests.java | 57 +++++++++++- src/reference/asciidoc/amqp.adoc | 6 +- 4 files changed, 117 insertions(+), 44 deletions(-) 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 22c73110..d9d0ed22 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-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -131,8 +131,50 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis @Override public void onMessage(org.springframework.amqp.core.Message amqpMessage, Channel channel) throws Exception { // NOSONAR - Message message = toMessagingMessage(amqpMessage); - invokeHandlerAndProcessResult(amqpMessage, channel, message); + Message message = null; + try { + message = toMessagingMessage(amqpMessage); + invokeHandlerAndProcessResult(amqpMessage, channel, message); + } + catch (ListenerExecutionFailedException ex) { + handleException(amqpMessage, channel, message, ex); + } + catch (ReplyFailureException ex) { + throw ex; + } + catch (Exception ex) { + handleException(amqpMessage, channel, message, new ListenerExecutionFailedException( + "Failed to convert message", ex, amqpMessage)); + } + } + + private void handleException(org.springframework.amqp.core.Message amqpMessage, Channel channel, + @Nullable Message message, ListenerExecutionFailedException e) throws Exception { + + if (this.errorHandler != null) { + try { + Message messageWithChannel = null; + if (message != null) { + messageWithChannel = MessageBuilder.fromMessage(message) + .setHeader(AmqpHeaders.CHANNEL, channel) + .build(); + } + Object errorResult = this.errorHandler.handleError(amqpMessage, messageWithChannel, e); + if (errorResult != null) { + handleResult(this.handlerAdapter.getInvocationResultFor(errorResult, message.getPayload()), + amqpMessage, channel, message); + } + else { + logger.trace("Error handler returned no result"); + } + } + catch (Exception ex) { + returnOrThrow(amqpMessage, channel, message, ex, ex); + } + } + else { + returnOrThrow(amqpMessage, channel, message, e.getCause(), e); + } } protected void invokeHandlerAndProcessResult(@Nullable org.springframework.amqp.core.Message amqpMessage, @@ -142,41 +184,16 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis logger.debug("Processing [" + message + "]"); } InvocationResult result = null; - try { - if (this.messagingMessageConverter.method == null && amqpMessage != null) { - amqpMessage.getMessageProperties() - .setTargetMethod(this.handlerAdapter.getMethodFor(message.getPayload())); - } - result = invokeHandler(amqpMessage, channel, message); - if (result.getReturnValue() != null) { - handleResult(result, amqpMessage, channel, message); - } - else { - logger.trace("No result object given - no result to handle"); - } + if (this.messagingMessageConverter.method == null && amqpMessage != null) { + amqpMessage.getMessageProperties() + .setTargetMethod(this.handlerAdapter.getMethodFor(message.getPayload())); } - catch (ListenerExecutionFailedException e) { - if (this.errorHandler != null) { - try { - Message messageWithChannel = MessageBuilder.fromMessage(message) - .setHeader(AmqpHeaders.CHANNEL, channel) - .build(); - Object errorResult = this.errorHandler.handleError(amqpMessage, messageWithChannel, e); - if (errorResult != null) { - handleResult(this.handlerAdapter.getInvocationResultFor(errorResult, message.getPayload()), - amqpMessage, channel, message); - } - else { - logger.trace("Error handler returned no result"); - } - } - catch (Exception ex) { - returnOrThrow(amqpMessage, channel, message, ex, ex); - } - } - else { - returnOrThrow(amqpMessage, channel, message, e.getCause(), e); - } + result = invokeHandler(amqpMessage, channel, message); + if (result.getReturnValue() != null) { + handleResult(result, amqpMessage, channel, message); + } + else { + logger.trace("No result object given - no result to handle"); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java index 63b425b2..745e351d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2021 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. @@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.listener.api; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException; +import org.springframework.lang.Nullable; /** * An error handler which is called when a {code @RabbitListener} method @@ -35,13 +36,13 @@ public interface RabbitListenerErrorHandler { * Handle the error. If an exception is not thrown, the return value is returned to * the sender using normal {@code replyTo/@SendTo} semantics. * @param amqpMessage the raw message received. - * @param message the converted spring-messaging message. + * @param message the converted spring-messaging message (if available). * @param exception the exception the listener threw, wrapped in a * {@link ListenerExecutionFailedException}. * @return the return value to be sent to the sender. * @throws Exception an exception which may be the original or different. */ - Object handleError(Message amqpMessage, org.springframework.messaging.Message message, + Object handleError(Message amqpMessage, @Nullable org.springframework.messaging.Message message, ListenerExecutionFailedException exception) throws Exception; // NOSONAR } 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 ea1abdd4..cdd4d528 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-2021 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. @@ -29,16 +29,20 @@ import java.util.Arrays; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler; import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException; import org.springframework.amqp.rabbit.test.MessageTestUtils; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.amqp.support.converter.MessageConversionException; +import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.factory.support.StaticListableBeanFactory; @@ -273,9 +277,50 @@ public class MessagingMessageListenerAdapterTests { assertThat(this.sample.batchPayloads.get(0).getClass()).isEqualTo(Foo.class); } + @Test + void errorHandlerAfterConversionEx() 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 null; + } + + }, String.class); + listener.setMessageConverter(new MessageConverter() { + + @Override + public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties) + throws MessageConversionException { + + return null; + } + + @Override + public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException { + throw new MessageConversionException("test"); + } + }); + listener.onMessage(message, channel); + assertThat(ehCalled.get()).isTrue(); + } + protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, Class... parameterTypes) { + return getSimpleInstance(methodName, null, parameterTypes); + } + + protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, RabbitListenerErrorHandler eh, + Class... parameterTypes) { + Method m = ReflectionUtils.findMethod(SampleBean.class, methodName, parameterTypes); - return createInstance(m, false); + return createInstance(m, false, eh); } protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, boolean returnExceptions, @@ -285,7 +330,13 @@ public class MessagingMessageListenerAdapterTests { } protected MessagingMessageListenerAdapter createInstance(Method m, boolean returnExceptions) { - MessagingMessageListenerAdapter adapter = new MessagingMessageListenerAdapter(null, m, returnExceptions, null); + return createInstance(m, returnExceptions, null); + } + + protected MessagingMessageListenerAdapter createInstance(Method m, boolean returnExceptions, + RabbitListenerErrorHandler eh) { + + MessagingMessageListenerAdapter adapter = new MessagingMessageListenerAdapter(null, m, returnExceptions, eh); adapter.setHandlerAdapter(new HandlerAdapter(factory.createInvocableHandlerMethod(sample, m))); return adapter; } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index b4352ae1..17ccab5b 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -3276,7 +3276,7 @@ static class TxServiceImpl implements TxService { @Override @RabbitListener(...) - public String handle(Foo foo, String rk) { + public String handle(Thing thing, String rk) { ... } @@ -3359,6 +3359,10 @@ public Object handleError(Message amqpMessage, org.springframework.messaging.Mes ---- ==== +Starting with version 2.2.18, if a message conversion exception is thrown, the error handler will be called, with `null` in the `message` argument. +This allows the application to send some result to the caller, indicating that a badly-formed message was received. +Previously, such errors were thrown and handled by the container. + ====== Container Management Containers created for annotations are not registered with the application context.