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

AMQP-103: added integration test for declarative retry

Browse Source
This commit is contained in:
Dave Syer
2011-03-25 11:58:57 +00:00
parent 02215be0d9
commit 0bc94c1676
3 changed files with 164 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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());
}
}

15
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

@@ -86,14 +86,14 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
private ActiveObjectCounter<BlockingQueueConsumer> cancellationLock = new ActiveObjectCounter<BlockingQueueConsumer>();
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

150
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerRetryIntegrationTests.java Normal file
View File

@@ -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<Object>() {
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();
}
}
}