From d58a7b2599b59a43391a8f0d6960b5dc213f17fb Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 31 Oct 2014 12:26:30 +0200 Subject: [PATCH] AMQP-438: Fix `MessagingMessageListenerAdapter` 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 ``. --- .../ConditionalRejectingErrorHandler.java | 2 +- .../MessagingMessageListenerAdapter.java | 6 ++ .../EnableRabbitIntegrationTests.java | 77 ++++++++++++++++++- .../MethodRabbitListenerEndpointTests.java | 8 +- .../MessagingMessageListenerAdapterTests.java | 9 ++- 5 files changed, 92 insertions(+), 10 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java index d1a4f04a..f5fa6201 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java @@ -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); 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 652d4bd2..b29af291 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 @@ -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); 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 c37bd0c1..69b2795f 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 @@ -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 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 errorHandlerError() { + return new AtomicReference(); + } + + @Bean + public ErrorHandler errorHandler() { + ErrorHandler handler = Mockito.spy(new ConditionalRejectingErrorHandler()); + Mockito.doAnswer(new Answer() { + @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(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpointTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpointTests.java index ee92cb10..59ef65d6 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpointTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpointTests.java @@ -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 } 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 fed5b973..74c63ef9 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 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");