diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index ca270f97..b6eaa10b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -24,6 +24,7 @@ import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils; import org.springframework.amqp.rabbit.connection.RabbitResourceHolder; import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; +import org.springframework.amqp.rabbit.listener.adapter.ListenerExecutionFailedException; import org.springframework.amqp.rabbit.support.RabbitAccessor; import org.springframework.amqp.rabbit.support.RabbitUtils; import org.springframework.beans.factory.BeanNameAware; @@ -623,7 +624,9 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im * @param listener the Spring ChannelAwareMessageListener to invoke * @param channel the Rabbit Channel to operate on * @param message the received Rabbit Message - * @throws Exception if thrown by Rabbit API methods + * @throws Exception if thrown by Rabbit API methods or listener itself. + *

+ * Exception thrown from listener will be wrapped to {@link ListenerExecutionFailedException}. * @see ChannelAwareMessageListener * @see #setExposeListenerChannel(boolean) */ @@ -639,8 +642,11 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im channelToUse = resourceHolder.getChannel(); } // Actually invoke the message listener... - listener.onMessage(message, channelToUse); - + try { + listener.onMessage(message, channelToUse); + } catch (Exception e) { + throw (Exception) wrapToListenerExecutionFailedExceptionIfNeeded(e); + } } finally { ConnectionFactoryUtils.releaseResources(resourceHolder); } @@ -650,12 +656,18 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im * Invoke the specified listener as Spring Rabbit MessageListener. *

* Default implementation performs a plain invocation of the onMessage method. + *

+ * Exception thrown from listener will be wrapped to {@link ListenerExecutionFailedException}. * @param listener the Rabbit MessageListener to invoke * @param message the received Rabbit Message * @see org.springframework.amqp.core.MessageListener#onMessage */ protected void doInvokeListener(MessageListener listener, Message message) { - listener.onMessage(message); + try { + listener.onMessage(message); + } catch (RuntimeException e) { + throw (RuntimeException) wrapToListenerExecutionFailedExceptionIfNeeded(e); + } } /** @@ -808,4 +820,16 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im } } + /** + * @param e + * @return If 'e' is of type {@link ListenerExecutionFailedException} - return 'e' as it is, otherwise wrap it to + * {@link ListenerExecutionFailedException} and return. + */ + protected Exception wrapToListenerExecutionFailedExceptionIfNeeded(Exception e) { + if (!(e instanceof ListenerExecutionFailedException)) { + // Wrap exception to ListenerExecutionFailedException. + return new ListenerExecutionFailedException("Listener threw exception", e); + } + return e; + } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java new file mode 100644 index 00000000..da79b9ed --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java @@ -0,0 +1,254 @@ +package org.springframework.amqp.rabbit.listener; + +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.log4j.Level; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.adapter.ListenerExecutionFailedException; +import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; +import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; +import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; +import org.springframework.util.ErrorHandler; + +import com.rabbitmq.client.Channel; + +public class MessageListenerContainerErrorHandlerIntegrationTests { + + private static Log logger = LogFactory.getLog(MessageListenerContainerErrorHandlerIntegrationTests.class); + + private static Queue queue = new Queue("test.queue"); + + // Mock error handler + private ErrorHandler errorHandler = mock(ErrorHandler.class); + + @Rule + public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue); + + @Rule + public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.INFO, RabbitTemplate.class, + SimpleMessageListenerContainer.class, BlockingQueueConsumer.class, + MessageListenerContainerErrorHandlerIntegrationTests.class); + + @Before + public void setUp() { + reset(errorHandler); + } + + @Test + public void testErrorHandlerInvokeExceptionFromPojo() throws Exception { + int messageCount = 3; + CountDownLatch latch = new CountDownLatch(messageCount); + doTest(messageCount, errorHandler, latch, new MessageListenerAdapter(new PojoThrowingExceptionListener(latch, + new Exception("Pojo exception")))); + + // Verify that error handler was invoked + verify(errorHandler, times(messageCount)).handleError(any(Throwable.class)); + } + + @Test + public void testErrorHandlerInvokeRuntimeExceptionFromPojo() throws Exception { + int messageCount = 3; + CountDownLatch latch = new CountDownLatch(messageCount); + doTest(messageCount, errorHandler, latch, new MessageListenerAdapter(new PojoThrowingExceptionListener(latch, + new RuntimeException("Pojo runtime exception")))); + + // Verify that error handler was invoked + verify(errorHandler, times(messageCount)).handleError(any(Throwable.class)); + } + + @Test + public void testErrorHandlerInvokeSpecRuntimeExceptionFromListener() throws Exception { + int messageCount = 3; + CountDownLatch latch = new CountDownLatch(messageCount); + doTest(messageCount, errorHandler, latch, new ThrowingExceptionListener(latch, + new ListenerExecutionFailedException("Listener throws specific runtime exception", null))); + + // Verify that error handler was invoked + verify(errorHandler, times(messageCount)).handleError(any(Throwable.class)); + } + + /** + * TODO: {@link #testErrorHandlerInvokeSpecRuntimeExceptionFromListener()} is very similar test, but throws specific + * type of {@link RuntimeException} - {@link ListenerExecutionFailedException}. That exception is handled in + * different way and listener is invoked for all available messages.
+ * In this test listener throws {@link RuntimeException} and it fails if you are expecting multiple messages to be + * processed.
+ * Investigation required! + * @throws Exception + */ + @Test + public void testErrorHandlerInvokeRuntimeExceptionFromListener() throws Exception { + // TODO If messageCount is more than 1 and RuntimeException is thrown from listener - test will fail. + // But if listener throws ListenerExecutionFailedException test will pass (see + // testErrorHandlerInvokeSpecRuntimeExceptionFromListener()) + // even with multiple messages. Investigation required. + int messageCount = 3; + CountDownLatch latch = new CountDownLatch(messageCount); + doTest(messageCount, errorHandler, latch, new ThrowingExceptionListener(latch, new RuntimeException( + "Listener runtime exception"))); + + // Verify that error handler was invoked + verify(errorHandler, times(messageCount)).handleError(any(Throwable.class)); + } + + @Test + public void testErrorHandlerInvokeExceptionFromChannelAwareListener() throws Exception { + int messageCount = 3; + CountDownLatch latch = new CountDownLatch(messageCount); + doTest(messageCount, errorHandler, latch, new ThrowingExceptionChannelAwareListener(latch, new Exception( + "Channel aware listener exception"))); + + // Verify that error handler was invoked + verify(errorHandler, times(messageCount)).handleError(any(Throwable.class)); + } + + @Test + public void testErrorHandlerInvokeRuntimeExceptionFromChannelAwareListener() throws Exception { + int messageCount = 3; + CountDownLatch latch = new CountDownLatch(messageCount); + doTest(messageCount, errorHandler, latch, new ThrowingExceptionChannelAwareListener(latch, + new RuntimeException("Channel aware listener runtime exception"))); + + // Verify that error handler was invoked + verify(errorHandler, times(messageCount)).handleError(any(Throwable.class)); + } + + public void doTest(int messageCount, ErrorHandler errorHandler, CountDownLatch latch, Object listener) + throws Exception { + int concurrentConsumers = 1; + RabbitTemplate template = createTemplate(concurrentConsumers); + + // Send messages to the queue + for (int i = 0; i < messageCount; i++) { + template.convertAndSend(queue.getName(), i + "foo"); + } + + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(template.getConnectionFactory()); + container.setMessageListener(listener); + container.setAcknowledgeMode(AcknowledgeMode.NONE); + container.setChannelTransacted(false); + container.setConcurrentConsumers(concurrentConsumers); + + container.setPrefetchCount(messageCount); + container.setTxSize(messageCount); + container.setQueueName(queue.getName()); + container.setErrorHandler(errorHandler); + container.afterPropertiesSet(); + container.start(); + + boolean waited = latch.await(500, TimeUnit.MILLISECONDS); + if (messageCount > 1) { + assertTrue("Expected to receive all messages before stop", waited); + } + + try { + assertNull(template.receiveAndConvert(queue.getName())); + } finally { + container.shutdown(); + } + } + + private RabbitTemplate createTemplate(int concurrentConsumers) { + RabbitTemplate template = new RabbitTemplate(); + // SingleConnectionFactory connectionFactory = new SingleConnectionFactory(); + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + connectionFactory.setChannelCacheSize(concurrentConsumers); + connectionFactory.setPort(BrokerTestUtils.getPort()); + template.setConnectionFactory(connectionFactory); + return template; + } + + // /////////////// + // Helper classes + // /////////////// + public static class PojoThrowingExceptionListener { + private CountDownLatch latch; + private Throwable exception; + + public PojoThrowingExceptionListener(CountDownLatch latch, Throwable exception) { + this.latch = latch; + this.exception = exception; + } + + public void handleMessage(String value) throws Throwable { + try { + logger.debug("Message in pojo: " + value); + Thread.sleep(100L); + throw exception; + } finally { + latch.countDown(); + } + } + } + + public static class ThrowingExceptionListener implements MessageListener { + private CountDownLatch latch; + private RuntimeException exception; + + public ThrowingExceptionListener(CountDownLatch latch, RuntimeException exception) { + this.latch = latch; + this.exception = exception; + } + + public void onMessage(Message message) { + try { + String value = new String(message.getBody()); + logger.debug("Message in listener: " + value); + try { + Thread.sleep(100L); + } catch (InterruptedException e) { + // Ignore this exception + } + throw exception; + } finally { + latch.countDown(); + } + } + } + + public static class ThrowingExceptionChannelAwareListener implements ChannelAwareMessageListener { + private CountDownLatch latch; + private Exception exception; + + public ThrowingExceptionChannelAwareListener(CountDownLatch latch, Exception exception) { + this.latch = latch; + this.exception = exception; + } + + public void onMessage(Message message, Channel channel) throws Exception { + try { + String value = new String(message.getBody()); + logger.debug("Message in channel aware listener: " + value); + try { + Thread.sleep(100L); + } catch (InterruptedException e) { + // Ignore this exception + } + throw exception; + } finally { + latch.countDown(); + } + } + } +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java index aff18f97..5de4fd5e 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java @@ -15,14 +15,18 @@ import org.apache.commons.logging.LogFactory; import org.apache.log4j.Level; import org.junit.After; import org.junit.Before; +import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import org.junit.runners.Parameterized.Parameters; import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.amqp.rabbit.test.BrokerRunning; @@ -33,6 +37,8 @@ import org.springframework.transaction.TransactionException; import org.springframework.transaction.support.AbstractPlatformTransactionManager; import org.springframework.transaction.support.DefaultTransactionStatus; +import com.rabbitmq.client.Channel; + @RunWith(Parameterized.class) public class SimpleMessageListenerContainerIntegrationTests { @@ -131,10 +137,46 @@ public class SimpleMessageListenerContainerIntegrationTests { } } + @Test + public void testPojoListenerSunnyDay() throws Exception { + CountDownLatch latch = new CountDownLatch(messageCount); + doSunnyDayTest(latch, new MessageListenerAdapter(new PojoListener(latch))); + } + @Test public void testListenerSunnyDay() throws Exception { CountDownLatch latch = new CountDownLatch(messageCount); - container = createContainer(new PojoListener(latch)); + doSunnyDayTest(latch, new Listener(latch)); + } + + @Test + @Ignore ("RabbitMQ connection can not be obtained if running in Parameterized mode") + public void testChannelAwareListenerSunnyDay() throws Exception { + CountDownLatch latch = new CountDownLatch(messageCount); + doSunnyDayTest(latch, new ChannelAwareListener(latch)); + } + + @Test + public void testPojoListenerWithException() throws Exception { + CountDownLatch latch = new CountDownLatch(messageCount); + doListenerWithExceptionTest(latch, new MessageListenerAdapter(new PojoListener(latch, true))); + } + + @Test + public void testListenerWithException() throws Exception { + CountDownLatch latch = new CountDownLatch(messageCount); + doListenerWithExceptionTest(latch, new Listener(latch, true)); + } + + @Test + @Ignore ("RabbitMQ connection can not be obtained if running in Parameterized mode") + public void testChannelAwareListenerWithException() throws Exception { + CountDownLatch latch = new CountDownLatch(messageCount); + doListenerWithExceptionTest(latch, new ChannelAwareListener(latch, true)); + } + + private void doSunnyDayTest(CountDownLatch latch, Object listener) throws Exception { + container = createContainer(listener); for (int i = 0; i < messageCount; i++) { template.convertAndSend(queue.getName(), i + "foo"); } @@ -142,11 +184,9 @@ public class SimpleMessageListenerContainerIntegrationTests { assertTrue("Timed out waiting for message", waited); assertNull(template.receiveAndConvert(queue.getName())); } - - @Test - public void testListenerWithException() throws Exception { - CountDownLatch latch = new CountDownLatch(messageCount); - container = createContainer(new PojoListener(latch, true)); + + private void doListenerWithExceptionTest(CountDownLatch latch, Object listener) throws Exception { + container = createContainer(listener); if (acknowledgeMode.isTransactionAllowed()) { // Should only need one message if it is going to fail for (int i = 0; i < concurrentConsumers; i++) { @@ -176,7 +216,7 @@ public class SimpleMessageListenerContainerIntegrationTests { private SimpleMessageListenerContainer createContainer(Object listener) { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(template.getConnectionFactory()); - container.setMessageListener(new MessageListenerAdapter(listener)); + container.setMessageListener(listener); container.setQueueName(queue.getName()); container.setTxSize(txSize); container.setPrefetchCount(txSize); @@ -221,6 +261,71 @@ public class SimpleMessageListenerContainerIntegrationTests { } } } + + public static class Listener implements MessageListener { + private AtomicInteger count = new AtomicInteger(); + + private final CountDownLatch latch; + + private final boolean fail; + + public Listener(CountDownLatch latch) { + this(latch, false); + } + + public Listener(CountDownLatch latch, boolean fail) { + this.latch = latch; + this.fail = fail; + } + + public void onMessage(Message message) { + String value = new String(message.getBody()); + try { + int counter = count.getAndIncrement(); + if (logger.isDebugEnabled() && counter % 500 == 0) { + logger.debug(value + counter); + } + if (fail) { + throw new RuntimeException("Planned failure"); + } + } finally { + latch.countDown(); + } + } + } + + public static class ChannelAwareListener implements ChannelAwareMessageListener { + private AtomicInteger count = new AtomicInteger(); + + private final CountDownLatch latch; + + private final boolean fail; + + public ChannelAwareListener(CountDownLatch latch) { + this(latch, false); + } + + public ChannelAwareListener(CountDownLatch latch, boolean fail) { + this.latch = latch; + this.fail = fail; + } + + public void onMessage(Message message, Channel channel) throws Exception { + String value = new String(message.getBody()); + try { + int counter = count.getAndIncrement(); + if (logger.isDebugEnabled() && counter % 500 == 0) { + logger.debug(value + counter); + } + if (fail) { + throw new RuntimeException("Planned failure"); + } + } finally { + latch.countDown(); + } + } + + } @SuppressWarnings("serial") private class TestTransactionManager extends AbstractPlatformTransactionManager { @@ -243,5 +348,4 @@ public class SimpleMessageListenerContainerIntegrationTests { } } - }