From 0bc94c1676324da068f27d33ea9333f01745e394 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Fri, 25 Mar 2011 11:58:57 +0000 Subject: [PATCH] AMQP-103: added integration test for declarative retry --- .../AbstractMessageListenerContainer.java | 6 +- .../SimpleMessageListenerContainer.java | 15 +- ...istenerContainerRetryIntegrationTests.java | 150 ++++++++++++++++++ 3 files changed, 164 insertions(+), 7 deletions(-) create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java 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 b83e21df..9b04faac 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 @@ -423,8 +423,10 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im protected void invokeErrorHandler(Throwable ex) { if (this.errorHandler != null) { this.errorHandler.handleError(ex); - } else if (logger.isWarnEnabled()) { - logger.warn("Execution of Rabbit message listener failed, and no ErrorHandler has been set.", ex); + } else if (logger.isDebugEnabled()) { + logger.debug("Execution of Rabbit message listener failed, and no ErrorHandler has been set.", ex); + } else if (logger.isInfoEnabled()) { + logger.info("Execution of Rabbit message listener failed, and no ErrorHandler has been set: "+ex.getClass()+": "+ex.getMessage()); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 461f5734..f9f583fe 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java @@ -86,14 +86,14 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private ActiveObjectCounter cancellationLock = new ActiveObjectCounter(); public static interface ContainerDelegate { - boolean receiveAndExecute(BlockingQueueConsumer consumer) throws Throwable; + void invokeListener(Channel channel, Message message) throws Exception; } private Advice[] advices = new Advice[0]; private ContainerDelegate delegate = new ContainerDelegate() { - public boolean receiveAndExecute(BlockingQueueConsumer consumer) throws Throwable { - return SimpleMessageListenerContainer.this.receiveAndExecute(consumer); + public void invokeListener(Channel channel, Message message) throws Exception { + SimpleMessageListenerContainer.super.invokeListener(channel, message); } }; @@ -324,7 +324,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } try { - logger.debug("Waiting for workers to finish."); + logger.info("Waiting for workers to finish."); boolean finished = cancellationLock.await(shutdownTimeout, TimeUnit.MILLISECONDS); if (finished) { logger.info("Successfully waited for workers to finish."); @@ -473,7 +473,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta while (isActive() || continuable) { try { // Will come back false when the queue is drained - continuable = proxy.receiveAndExecute(consumer) && !isChannelTransacted(); + continuable = receiveAndExecute(consumer) && !isChannelTransacted(); } catch (ListenerExecutionFailedException ex) { // Continue to process, otherwise re-throw } @@ -521,6 +521,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } + + @Override + protected void invokeListener(Channel channel, Message message) throws Exception { + proxy.invokeListener(channel, message); + } /** * Wait for a period determined by the {@link #setRecoveryInterval(long) recoveryInterval} to give the container a diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java new file mode 100644 index 00000000..6e25061c --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java @@ -0,0 +1,150 @@ +package org.springframework.amqp.rabbit.listener; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.aopalliance.aop.Advice; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.log4j.Level; +import org.junit.Rule; +import org.junit.Test; +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessagePostProcessor; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +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.retry.interceptor.MethodArgumentsKeyGenerator; +import org.springframework.retry.interceptor.MethodInvocationRecoverer; +import org.springframework.retry.interceptor.NewMethodArgumentsIdentifier; +import org.springframework.retry.interceptor.StatefulRetryOperationsInterceptor; +import org.springframework.retry.support.RetryTemplate; + +public class MessageListenerContainerRetryIntegrationTests { + + private static Log logger = LogFactory.getLog(MessageListenerContainerRetryIntegrationTests.class); + + private static Queue queue = new Queue("test.queue"); + + @Rule + public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue); + + @Rule + public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.INFO, RabbitTemplate.class, + SimpleMessageListenerContainer.class, BlockingQueueConsumer.class); + + private RabbitTemplate createTemplate(int concurrentConsumers) { + RabbitTemplate template = new RabbitTemplate(); + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + connectionFactory.setChannelCacheSize(concurrentConsumers); + connectionFactory.setPort(BrokerTestUtils.getPort()); + template.setConnectionFactory(connectionFactory); + return template; + } + + @Test + public void testStatefulRetryPerMessage() throws Exception { + + int messageCount = 10; + int concurrentConsumers = 3; + + RabbitTemplate template = createTemplate(concurrentConsumers); + + for (int i = 0; i < messageCount; i++) { + template.convertAndSend(queue.getName(), (Object) (i + "foo"), new MessagePostProcessor() { + // There is no message id by default + public Message postProcessMessage(Message message) throws AmqpException { + message.getMessageProperties().setMessageId(UUID.randomUUID().toString()); + return message; + } + }); + } + + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(template.getConnectionFactory()); + final CountDownLatch latch = new CountDownLatch(messageCount); + PojoListener listener = new PojoListener(); + container.setMessageListener(new MessageListenerAdapter(listener)); + container.setAcknowledgeMode(AcknowledgeMode.AUTO); + container.setChannelTransacted(true); + container.setConcurrentConsumers(concurrentConsumers); + + StatefulRetryOperationsInterceptor retryInterceptor = new StatefulRetryOperationsInterceptor(); + retryInterceptor.setRetryOperations(new RetryTemplate()); + retryInterceptor.setNewItemIdentifier(new NewMethodArgumentsIdentifier() { + public boolean isNew(Object[] args) { + Message message = (Message) args[1]; + return !message.getMessageProperties().isRedelivered(); + } + }); + retryInterceptor.setRecoverer(new MethodInvocationRecoverer() { + public Object recover(Object[] args, Throwable cause) { + logger.info("Recovered: " + Arrays.asList(args)); + latch.countDown(); + return null; + } + }); + retryInterceptor.setKeyGenerator(new MethodArgumentsKeyGenerator() { + public Object getKey(Object[] args) { + Message message = (Message) args[1]; + logger.info("Key: " + new String(message.getBody())); + return message.getMessageProperties().getMessageId(); + } + }); + container.setAdviceChain(new Advice[] { retryInterceptor }); + + container.setQueueNames(queue.getName()); + container.afterPropertiesSet(); + container.start(); + + try { + + int timeout = Math.min(1 + messageCount / concurrentConsumers, 30); + + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + boolean waited = latch.await(timeout, TimeUnit.SECONDS); + logger.info("All messages received after start: " + waited); + assertEquals(concurrentConsumers, container.getActiveConsumerCount()); + assertTrue("Timed out waiting for message", waited); + + assertEquals(concurrentConsumers, container.getActiveConsumerCount()); + + // Retried each one 3 times... + assertEquals(3*messageCount, listener.count.get()); + + } finally { + container.shutdown(); + assertEquals(0, container.getActiveConsumerCount()); + } + + // All failed messages recovered + assertNull(template.receiveAndConvert(queue.getName())); + + } + + public static class PojoListener { + private AtomicInteger count = new AtomicInteger(); + + public void handleMessage(String value) throws Exception { + logger.debug(value + count.getAndIncrement()); + throw new RuntimeException("Planned"); + } + + public int getCount() { + return count.get(); + } + } + +}