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

GH-1339: Fix RLErrorHandler with Conversion Ex. (#1346)

Browse Source 
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**
This commit is contained in:
Gary Russell
2021-06-07 16:05:14 -04:00
committed by GitHub
parent 3b8fc2a7c2
commit 96070f6720
4 changed files with 117 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java
View File

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

57
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java
View File

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

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

@@ -3276,7 +3276,7 @@ static class TxServiceImpl implements TxService<Foo> {
@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.