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

AMQP-438: Fix MessagingMessageListenerAdapter

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-438

Since `DefaultExceptionStrategy` treats only `org.springframework.amqp.support.converter.MessageConversionException` as `fatal`
and assuming backward compatibility for SF < 4.1, add `catch (org.springframework.messaging.converter.MessageConversionException ex) {`
to the `MessagingMessageListenerAdapter` to wrap that exception to the `org.springframework.amqp.support.converter.MessageConversionException`.
Having that the `DefaultExceptionStrategy` works with `@RabbitListener` as it is with generic `<rabbit:listener>`.
This commit is contained in:
Artem Bilan
2014-10-31 12:26:30 +02:00
parent 0d92ab7447
commit d58a7b2599
5 changed files with 92 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java
View File

@@ -40,7 +40,7 @@ import org.springframework.util.ErrorHandler;
* @since 1.3.2
*
*/
public final class ConditionalRejectingErrorHandler implements ErrorHandler {
public class ConditionalRejectingErrorHandler implements ErrorHandler {
protected static final Log logger = LogFactory.getLog(ConditionalRejectingErrorHandler.class);

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

@@ -44,6 +44,7 @@ import com.rabbitmq.client.Channel;
*
* @author Stephane Nicoll
* @author Gary Russell
* @author Artem Bilan
* @since 1.4
*/
public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageListener {
@@ -111,6 +112,11 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
try {
return this.handlerMethod.invoke(message, amqpMessage, channel);
}
catch (org.springframework.messaging.converter.MessageConversionException ex) {
throw new ListenerExecutionFailedException(createMessagingErrorMessage("Listener method could not " +
"be invoked with the incoming message"),
new MessageConversionException("Cannot handle message", ex));
}
catch (MessagingException ex) {
throw new ListenerExecutionFailedException(createMessagingErrorMessage("Listener method could not " +
"be invoked with the incoming message"), ex);

77
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
View File

@@ -16,20 +16,34 @@
package org.springframework.amqp.rabbit.annotation;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.amqp.rabbit.test.BrokerRunning;
import org.springframework.amqp.rabbit.test.MessageTestUtils;
import org.springframework.beans.factory.annotation.Autowired;
@@ -42,6 +56,7 @@ import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ErrorHandler;
/**
*
@@ -56,11 +71,18 @@ public class EnableRabbitIntegrationTests {
@ClassRule
public static final BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues(
"test.simple", "test.header", "test.message", "test.reply", "test.sendTo", "test.sendTo.reply");
"test.simple", "test.header", "test.message", "test.reply", "test.sendTo", "test.sendTo.reply",
"test.invalidPojo");
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private CountDownLatch errorHandlerLatch;
@Autowired
private AtomicReference<Throwable> errorHandlerError;
@Test
public void simpleEndpoint() {
assertEquals("FOO", rabbitTemplate.convertSendAndReceive("test.simple", "foo"));
@@ -108,6 +130,23 @@ public class EnableRabbitIntegrationTests {
assertEquals("BAR", result);
}
@Test
public void testInvalidPojoConversion() throws InterruptedException {
this.rabbitTemplate.convertAndSend("test.invalidPojo", "bar");
assertTrue(this.errorHandlerLatch.await(10, TimeUnit.SECONDS));
Throwable throwable = this.errorHandlerError.get();
assertNotNull(throwable);
assertThat(throwable, instanceOf(AmqpRejectAndDontRequeueException.class));
assertThat(throwable.getCause(), instanceOf(ListenerExecutionFailedException.class));
assertThat(throwable.getCause().getCause(),
instanceOf(org.springframework.amqp.support.converter.MessageConversionException.class));
assertThat(throwable.getCause().getCause().getCause(),
instanceOf(org.springframework.messaging.converter.MessageConversionException.class));
assertThat(throwable.getCause().getCause().getCause().getMessage(),
containsString("Failed to convert message payload 'bar' to 'java.util.Date'"));
}
public static class MyService {
@RabbitListener(queues = "test.simple")
@@ -136,6 +175,12 @@ public class EnableRabbitIntegrationTests {
public String capitalizeAndSendTo(String foo) {
return foo.toUpperCase();
}
@RabbitListener(queues = "test.invalidPojo")
public void handleIt(Date body) {
}
}
@Configuration
@@ -146,9 +191,39 @@ public class EnableRabbitIntegrationTests {
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(rabbitConnectionFactory());
factory.setErrorHandler(errorHandler());
return factory;
}
@Bean
public CountDownLatch errorHandlerLatch() {
return new CountDownLatch(1);
}
@Bean
public AtomicReference<Throwable> errorHandlerError() {
return new AtomicReference<Throwable>();
}
@Bean
public ErrorHandler errorHandler() {
ErrorHandler handler = Mockito.spy(new ConditionalRejectingErrorHandler());
Mockito.doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
try {
return invocation.callRealMethod();
}
catch (Throwable e) {
errorHandlerError().set(e);
errorHandlerLatch().countDown();
throw e;
}
}
}).when(handler).handleError(Mockito.any(Throwable.class));
return handler;
}
@Bean
public MyService myService() {
return new MyService();

8
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpointTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2014 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.
@@ -42,9 +42,6 @@ import org.mockito.ArgumentCaptor;
import org.springframework.amqp.core.Address;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.adapter.ReplyFailureException;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
@@ -73,6 +70,7 @@ import com.rabbitmq.client.Channel;
/**
* @author Stephane Nicoll
* @author Artem Bilan
*/
public class MethodRabbitListenerEndpointTests {
@@ -378,7 +376,7 @@ public class MethodRabbitListenerEndpointTests {
Channel channel = mock(Channel.class);
thrown.expect(ListenerExecutionFailedException.class);
thrown.expectCause(Matchers.isA(org.springframework.messaging.converter.MessageConversionException.class));
thrown.expectCause(Matchers.isA(MessageConversionException.class));
thrown.expectMessage(getDefaultListenerMethod(Integer.class).toGenericString()); // ref to method
listener.onMessage(createTextMessage("test"), channel); // test is not a valid integer
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2014 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.
@@ -31,7 +31,6 @@ import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFaile
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.support.StaticListableBeanFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.ReflectionUtils;
@@ -40,6 +39,7 @@ import com.rabbitmq.client.Channel;
/**
* @author Stephane Nicoll
* @author Artem Bilan
*/
public class MessagingMessageListenerAdapterTests {
@@ -102,7 +102,10 @@ public class MessagingMessageListenerAdapterTests {
fail("Should have thrown an exception");
}
catch (ListenerExecutionFailedException ex) {
assertEquals(MessageConversionException.class, ex.getCause().getClass());
assertEquals(org.springframework.amqp.support.converter.MessageConversionException.class,
ex.getCause().getClass());
assertEquals(org.springframework.messaging.converter.MessageConversionException.class,
ex.getCause().getCause().getClass());
}
catch (Exception ex) {
fail("Should not have thrown another exception");