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 58a41120..115a4750 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 @@ -352,14 +352,14 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta try { // Need to recycle the channel in this consumer consumer.stop(); + // Ensure consumer counts are correct (another is not going + // to start because of the exception, but + // we haven't counted down yet) this.cancellationLock.release(consumer); this.consumers.remove(consumer); consumer = createBlockingQueueConsumer(); this.consumers.add(consumer); } catch (RuntimeException e) { - // Ensure consumer counts are correct (another is not going - // to start because of the exception, but - // we haven't counted down yet) logger.warn("Consumer died on restart. " + e.getClass() + ": " + e.getMessage()); // Thrown into the void (probably) in a background thread. // Oh well, here goes... diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryRepeatIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryRepeatIntegrationTests.java new file mode 100644 index 00000000..27afa96d --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryRepeatIntegrationTests.java @@ -0,0 +1,169 @@ +package org.springframework.amqp.rabbit.listener; + +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.commons.logging.Log; +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.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +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; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; +import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; +import org.springframework.amqp.rabbit.test.RepeatProcessor; +import org.springframework.test.annotation.Repeat; + +import com.rabbitmq.client.Channel; + +/** + * Long-running test created to facilitate profiling of SimpleMessageListenerContainer. + * + * @author Dave Syer + * + */ +@Ignore +public class MessageListenerRecoveryRepeatIntegrationTests { + + private static Log logger = LogFactory.getLog(MessageListenerRecoveryRepeatIntegrationTests.class); + + private Queue queue = new Queue("test.queue"); + + private Queue sendQueue = new Queue("test.send"); + + private int concurrentConsumers = 1; + + private int messageCount = 2; + + private int txSize = 1; + + private boolean transactional = false; + + private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO; + + private SimpleMessageListenerContainer container; + + @Rule + public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.INFO, RabbitTemplate.class, + SimpleMessageListenerContainer.class, BlockingQueueConsumer.class); + + @Rule + public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue, sendQueue); + + @Rule + public RepeatProcessor repeatProcessor = new RepeatProcessor(); + + private CloseConnectionListener listener; + + private ConnectionFactory connectionFactory; + + @Before + public void init() { + if (!repeatProcessor.isInitialized()) { + logger.info("Initializing at start of test"); + connectionFactory = createConnectionFactory(); + listener = new CloseConnectionListener(); + container = createContainer(queue.getName(), listener, connectionFactory); + } + } + + @After + public void clear() throws Exception { + if (repeatProcessor.isFinalizing()) { + // Wait for broker communication to finish before trying to stop container + Thread.sleep(300L); + logger.info("Shutting down at end of test"); + if (container != null) { + container.shutdown(); + } + } + } + + @Test + @Repeat(1000) + public void testListenerRecoversFromClosedConnection() throws Exception { + + // logger.info("Testing..."); + + RabbitTemplate template = new RabbitTemplate(connectionFactory); + CountDownLatch latch = new CountDownLatch(messageCount); + listener.setLatch(latch); + + for (int i = 0; i < messageCount; i++) { + template.convertAndSend(queue.getName(), i + "foo"); + } + + int timeout = Math.min(4 + messageCount / (4 * concurrentConsumers), 30); + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + boolean waited = latch.await(timeout, TimeUnit.SECONDS); + assertTrue("Timed out waiting for message", waited); + + assertNull(template.receiveAndConvert(queue.getName())); + + } + + private ConnectionFactory createConnectionFactory() { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + connectionFactory.setChannelCacheSize(concurrentConsumers); + // connectionFactory.setPort(BrokerTestUtils.getTracerPort()); + connectionFactory.setPort(BrokerTestUtils.getPort()); + return connectionFactory; + } + + private SimpleMessageListenerContainer createContainer(String queueName, Object listener, + ConnectionFactory connectionFactory) { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); + container.setMessageListener(new MessageListenerAdapter(listener)); + container.setQueueNames(queueName); + container.setTxSize(txSize); + container.setPrefetchCount(txSize); + container.setConcurrentConsumers(concurrentConsumers); + container.setChannelTransacted(transactional); + container.setAcknowledgeMode(acknowledgeMode); + container.setTaskExecutor(Executors.newFixedThreadPool(concurrentConsumers)); + container.afterPropertiesSet(); + container.start(); + return container; + } + + private static class CloseConnectionListener implements ChannelAwareMessageListener { + + private AtomicBoolean failed = new AtomicBoolean(false); + + private CountDownLatch latch; + + public void setLatch(CountDownLatch latch) { + this.latch = latch; + failed.set(false); + } + + public void onMessage(Message message, Channel channel) throws Exception { + String value = new String(message.getBody()); + logger.info("Receiving: " + value); + if (failed.compareAndSet(false, true)) { + // intentional error (causes exception on connection thread): + // channel.abort(); + // throw new RuntimeException("Planned"); + throw new FatalListenerExecutionException("Planned"); + } else { + 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 0d132a38..b62171f5 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 @@ -12,6 +12,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.log4j.Level; import org.junit.After; import org.junit.Before; import org.junit.Rule; @@ -29,6 +30,7 @@ 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.transaction.TransactionDefinition; import org.springframework.transaction.TransactionException; import org.springframework.transaction.support.AbstractPlatformTransactionManager; @@ -49,9 +51,9 @@ public class SimpleMessageListenerContainerIntegrationTests { private final AcknowledgeMode acknowledgeMode; - // @Rule - // public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.ERROR, RabbitTemplate.class, - // SimpleMessageListenerContainer.class, BlockingQueueConsumer.class); + @Rule + public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.ERROR, RabbitTemplate.class, + SimpleMessageListenerContainer.class, BlockingQueueConsumer.class); @Rule public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/RepeatProcessor.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/RepeatProcessor.java index 28a11f2f..9ac3c73a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/RepeatProcessor.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/RepeatProcessor.java @@ -40,8 +40,9 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.test.annotation.Repeat; /** - * A JUnit method @Rule that looks at Spring repeat annotations on methods and - * executes the test multiple times (without re-initializing the test case). + * A JUnit method @Rule that looks at Spring repeat annotations on methods and executes the test multiple times + * (without re-initializing the test case if necessary). To avoid re-initializing use the {@link #isInitialized()} + * method to protect the @Before and @After methods. * * @author Dave Syer * @since 2.0 @@ -55,6 +56,8 @@ public class RepeatProcessor implements MethodRule { private volatile boolean initialized = false; + private volatile boolean finalizing = false; + public RepeatProcessor() { this(0); } @@ -63,7 +66,7 @@ public class RepeatProcessor implements MethodRule { this.concurrency = concurrency < 0 ? 0 : concurrency; } - public Statement apply(final Statement base, FrameworkMethod method, Object target) { + public Statement apply(final Statement base, FrameworkMethod method, final Object target) { Repeat repeat = AnnotationUtils.findAnnotation(method.getMethod(), Repeat.class); if (repeat == null) { @@ -77,72 +80,73 @@ public class RepeatProcessor implements MethodRule { initializeIfNecessary(target); - try { - if (concurrency <= 0) { - return new Statement() { - @Override - public void evaluate() throws Throwable { - for (int i = 0; i < repeats; i++) { - try { - base.evaluate(); - } - catch (Throwable t) { - throw new IllegalStateException("Failed on iteration: " + i, t); - } - } - } - }; - } + if (concurrency <= 0) { return new Statement() { @Override public void evaluate() throws Throwable { - List> results = new ArrayList>(); - ExecutorService executor = Executors.newFixedThreadPool(concurrency); - CompletionService completionService = new ExecutorCompletionService(executor); try { for (int i = 0; i < repeats; i++) { - final int count = i; - results.add(completionService.submit(new Callable() { - public Boolean call() { - try { - base.evaluate(); - } - catch (Throwable t) { - throw new IllegalStateException("Failed on iteration: " + count, t); - } - return true; - } - })); + try { + base.evaluate(); + } catch (Throwable t) { + throw new IllegalStateException("Failed on iteration: " + i, t); + } } - for (int i = 0; i < repeats; i++) { - Future future = completionService.take(); - assertTrue("Null result from completer", future.get()); - } - } - finally { - executor.shutdownNow(); + } finally { + finalizeIfNecessary(target); } } }; } - finally { - finalizeIfNecessary(target); - } + return new Statement() { + @Override + public void evaluate() throws Throwable { + List> results = new ArrayList>(); + ExecutorService executor = Executors.newFixedThreadPool(concurrency); + CompletionService completionService = new ExecutorCompletionService(executor); + try { + for (int i = 0; i < repeats; i++) { + final int count = i; + results.add(completionService.submit(new Callable() { + public Boolean call() { + try { + base.evaluate(); + } catch (Throwable t) { + throw new IllegalStateException("Failed on iteration: " + count, t); + } + return true; + } + })); + } + for (int i = 0; i < repeats; i++) { + Future future = completionService.take(); + assertTrue("Null result from completer", future.get()); + } + } finally { + executor.shutdownNow(); + finalizeIfNecessary(target); + } + } + }; } private void finalizeIfNecessary(Object target) { + finalizing = true; List afters = new TestClass(target.getClass()).getAnnotatedMethods(After.class); - if (!afters.isEmpty()) { - logger.debug("Running @After methods"); - try { - new RunAfters(new Statement() { - public void evaluate() { - } - }, afters, target).evaluate(); - } - catch (Throwable e) { - Assert.assertThat(e, CoreMatchers.not(CoreMatchers.anything())); + try { + if (!afters.isEmpty()) { + logger.debug("Running @After methods"); + try { + new RunAfters(new Statement() { + public void evaluate() { + } + }, afters, target).evaluate(); + } catch (Throwable e) { + Assert.assertThat(e, CoreMatchers.not(CoreMatchers.anything())); + } } + } finally { + finalizing = false; } } @@ -156,8 +160,7 @@ public class RepeatProcessor implements MethodRule { public void evaluate() { } }, befores, target).evaluate(); - } - catch (Throwable e) { + } catch (Throwable e) { Assert.assertThat(e, CoreMatchers.not(CoreMatchers.anything())); } initialized = true; @@ -171,6 +174,10 @@ public class RepeatProcessor implements MethodRule { return initialized; } + public boolean isFinalizing() { + return finalizing; + } + public int getConcurrency() { return concurrency > 0 ? concurrency : 1; }