AMQP-101 Added tests-cases to test ErrorHandler invokations from listener container.

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. 
+	 * <p/>
+	 * 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.
 	 * <p>
 	 * Default implementation performs a plain invocation of the <code>onMessage</code> method.
+	 * <p/>
+	 * 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;
+	}
 }

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. <br/>
+	 * In this test listener throws {@link RuntimeException} and it fails if you are expecting multiple messages to be
+	 * processed.<br/>
+	 * 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();
+			}
+		}
+	}
+}

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