From c940e03624db3ceff1ebeb4aabdaadb9ed03e091 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Wed, 23 Mar 2011 13:10:24 +0000 Subject: [PATCH] AMQP-125: add custom lock manager and tests - Add test for transactional send in listener - Add ActiveObjectCounter and use it in SMLC and consumer --- .../rabbit/listener/ActiveObjectCounter.java | 72 +++++++++++ .../listener/BlockingQueueConsumer.java | 25 +++- .../SimpleMessageListenerContainer.java | 36 +++--- .../core/RabbitBindingIntegrationTests.java | 3 +- .../listener/ActiveObjectCounterTests.java | 75 ++++++++++++ ...nerContainerLifecycleIntegrationTests.java | 4 +- ...veryCachingConnectionIntegrationTests.java | 114 ++++++++++++++++-- 7 files changed, 293 insertions(+), 36 deletions(-) create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounterTests.java diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java new file mode 100644 index 00000000..7f27397d --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounter.java @@ -0,0 +1,72 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package org.springframework.amqp.rabbit.listener; + +import java.util.Collection; +import java.util.HashSet; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +/** + * @author Dave Syer + * + */ +public class ActiveObjectCounter { + + private final ConcurrentMap locks = new ConcurrentHashMap(); + + public void add(T object) { + CountDownLatch lock = new CountDownLatch(1); + locks.putIfAbsent(object, lock); + } + + public void release(T object) { + CountDownLatch remove = locks.remove(object); + if (remove != null) { + remove.countDown(); + } + } + + public boolean await(Long timeout, TimeUnit timeUnit) throws InterruptedException { + long t0 = System.currentTimeMillis(); + long t1 = t0 + TimeUnit.MILLISECONDS.convert(timeout, timeUnit); + while (System.currentTimeMillis() <= t1) { + if (locks.isEmpty()) { + return true; + } + Collection objects = new HashSet(locks.keySet()); + for (T object : objects) { + CountDownLatch lock = locks.get(object); + if (lock==null) { + continue; + } + t0 = System.currentTimeMillis(); + if (lock.await(t1 - t0, TimeUnit.MILLISECONDS)) { + locks.remove(object); + } + } + } + return false; + } + + public int getCount() { + return locks.size(); + } + + public void reset() { + locks.clear(); + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 733aa8b7..3f58a656 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -57,13 +57,16 @@ public class BlockingQueueConsumer { private final ConnectionFactory connectionFactory; + private final ActiveObjectCounter stopped; + /** * Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker * until it is started. */ - public BlockingQueueConsumer(ConnectionFactory connectionFactory, AcknowledgeMode acknowledgeMode, - boolean transactional, int prefetchCount, String... queues) { + public BlockingQueueConsumer(ConnectionFactory connectionFactory, ActiveObjectCounter stopped, + AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, String... queues) { this.connectionFactory = connectionFactory; + this.stopped = stopped; this.acknowledgeMode = acknowledgeMode; this.transactional = transactional; this.prefetchCount = prefetchCount; @@ -145,6 +148,7 @@ public class BlockingQueueConsumer { this.channel = ConnectionFactoryUtils.getTransactionalResourceHolder(connectionFactory, transactional) .getChannel(); this.consumer = new InternalConsumer(channel); + this.stopped.add(this); try { // Set basicQos before calling basicConsume (it is ignored if we are not transactional and the broker will // send blocks of 100 messages) @@ -170,10 +174,10 @@ public class BlockingQueueConsumer { public void stop() { cancelled.set(true); - logger.debug("Closing Rabbit Channel: " + channel); if (consumer != null && consumer.getChannel() != null && consumer.getConsumerTag() != null) { RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), transactional); } + logger.debug("Closing Rabbit Channel: " + channel); // This one never throws exceptions... RabbitUtils.closeChannel(channel); } @@ -193,6 +197,15 @@ public class BlockingQueueConsumer { // TODO: interrupt? } + @Override + public void handleCancelOk(String consumerTag) { + if (logger.isDebugEnabled()) { + logger.debug("Received cancellation notice for " + BlockingQueueConsumer.this); + } + // Signal to the container that we have been cancelled + stopped.release(BlockingQueueConsumer.this); + } + @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { @@ -201,8 +214,10 @@ public class BlockingQueueConsumer { return; } } - // TODO: do we want to pass on 'consumerTag'? - logger.debug("Storing delivery for " + BlockingQueueConsumer.this); + if (logger.isDebugEnabled()) { + // TODO: do we want to pass on 'consumerTag'? + logger.debug("Storing delivery for " + BlockingQueueConsumer.this); + } try { // TODO: If transactional we could use a bounded queue and offer() here with a timeout // in which case if it fails we could nack the message and have it requeued. 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 d694dffc..cb793ec8 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 @@ -33,6 +33,8 @@ import org.springframework.aop.Pointcut; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.DefaultPointcutAdvisor; import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.jmx.export.annotation.ManagedMetric; +import org.springframework.jmx.support.MetricType; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; @@ -82,7 +84,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private TransactionAttribute transactionAttribute = new DefaultTransactionAttribute(); - private CountDownLatch cancellationLock; + private ActiveObjectCounter cancellationLock = new ActiveObjectCounter(); public static interface ContainerDelegate { boolean receiveAndExecute(BlockingQueueConsumer consumer) throws Throwable; @@ -276,8 +278,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta initializeProxy(); } + @ManagedMetric(metricType = MetricType.GAUGE) public int getActiveConsumerCount() { - return (int) cancellationLock.getCount(); + return cancellationLock.getCount(); } /** @@ -288,17 +291,15 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta */ protected void doStart() throws Exception { super.doStart(); - initializeConsumers(); synchronized (this.consumersMonitor) { + initializeConsumers(); if (this.consumers == null) { logger.info("Consumers were initialized and then cleared (presumably the container was stopped concurrently)"); return; } - cancellationLock = new CountDownLatch(this.consumers.size()); Set processors = new HashSet(); for (BlockingQueueConsumer consumer : this.consumers) { - AsyncMessageProcessingConsumer processor = new AsyncMessageProcessingConsumer(consumer, - cancellationLock); + AsyncMessageProcessingConsumer processor = new AsyncMessageProcessingConsumer(consumer); processors.add(processor); this.taskExecutor.execute(processor); } @@ -345,6 +346,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta protected void initializeConsumers() { synchronized (this.consumersMonitor) { if (this.consumers == null) { + cancellationLock.reset(); this.consumers = new HashSet(this.concurrentConsumers); for (int i = 0; i < this.concurrentConsumers; i++) { BlockingQueueConsumer consumer = createBlockingQueueConsumer(); @@ -361,8 +363,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta protected BlockingQueueConsumer createBlockingQueueConsumer() { BlockingQueueConsumer consumer; String[] queues = getRequiredQueueNames(); - consumer = new BlockingQueueConsumer(getConnectionFactory(), getAcknowledgeMode(), isChannelTransacted(), - prefetchCount, queues); + consumer = new BlockingQueueConsumer(getConnectionFactory(), cancellationLock, getAcknowledgeMode(), + isChannelTransacted(), prefetchCount, queues); return consumer; } @@ -372,6 +374,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta try { // Need to recycle the channel in this consumer consumer.stop(); + this.cancellationLock.release(consumer); this.consumers.remove(consumer); consumer = createBlockingQueueConsumer(); this.consumers.add(consumer); @@ -380,12 +383,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta // to start because of the exception, but // we haven't counted down yet) logger.warn("Consumer died on restart. " + e.getClass() + ": " + e.getMessage()); - cancellationLock.countDown(); // Thrown into the void (probably) in a background thread. // Oh well, here goes... throw e; } - this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, cancellationLock)); + this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer)); } } } @@ -426,17 +428,14 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private final BlockingQueueConsumer consumer; - private final CountDownLatch latch; - private final CountDownLatch start; private volatile ListenerStartupFatalException startupException; private AtomicBoolean started = new AtomicBoolean(false); - public AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer, CountDownLatch latch) { + public AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) { this.consumer = consumer; - this.latch = latch; this.start = new CountDownLatch(1); } @@ -497,8 +496,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta // Fatal, but no point re-throwing, so just abort. aborted = true; } catch (Throwable t) { - if (logger.isInfoEnabled()) { - logger.info( + if (logger.isDebugEnabled()) { + logger.warn( "Consumer raised exception, processing can restart if the connection factory supports it", t); } else { @@ -511,17 +510,18 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta if (!isActive() || aborted) { logger.debug("Cancelling " + consumer); - latch.countDown(); try { consumer.stop(); } catch (AmqpException e) { - logger.info("Could not stop message consumer on shutdown", e); + logger.info("Could not cancel message consumer", e); + // TODO: should we release the cancellationLock here? } if (aborted) { stop(); } } else { logger.info("Restarting " + consumer); + // cancellationLock is not decremented on restart restart(consumer); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java index 8a87a442..6436ba87 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java @@ -14,6 +14,7 @@ import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.RabbitAccessor; +import org.springframework.amqp.rabbit.listener.ActiveObjectCounter; import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer; import org.springframework.amqp.rabbit.test.BrokerRunning; import org.springframework.amqp.support.converter.SimpleMessageConverter; @@ -232,7 +233,7 @@ public class RabbitBindingIntegrationTests { } private BlockingQueueConsumer createConsumer(RabbitAccessor accessor) { - BlockingQueueConsumer consumer = new BlockingQueueConsumer(accessor.getConnectionFactory(), AcknowledgeMode.AUTO, true, 1, queue.getName()); + BlockingQueueConsumer consumer = new BlockingQueueConsumer(accessor.getConnectionFactory(), new ActiveObjectCounter(), AcknowledgeMode.AUTO, true, 1, queue.getName()); consumer.start(); return consumer; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounterTests.java new file mode 100644 index 00000000..47ceae75 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ActiveObjectCounterTests.java @@ -0,0 +1,75 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.amqp.rabbit.listener; + +import static org.junit.Assert.assertEquals; + +import java.util.concurrent.Callable; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; + + +/** + * @author Dave Syer + * + */ +public class ActiveObjectCounterTests { + + private ActiveObjectCounter counter = new ActiveObjectCounter(); + + @Test + public void testActiveCount() throws Exception { + final Object object1 = new Object(); + final Object object2 = new Object(); + counter.add(object1); + counter.add(object2); + assertEquals(2, counter.getCount()); + counter.release(object2); + assertEquals(1, counter.getCount()); + counter.release(object1); + counter.release(object1); + assertEquals(0, counter.getCount()); + } + + @Test + public void testWaitForLocks() throws Exception { + final Object object1 = new Object(); + final Object object2 = new Object(); + counter.add(object1); + counter.add(object2); + Future future = Executors.newSingleThreadExecutor().submit(new Callable() { + public Boolean call() throws Exception { + counter.release(object1); + counter.release(object2); + counter.release(object2); + return true; + } + }); + assertEquals(true, counter.await(1000L, TimeUnit.MILLISECONDS)); + assertEquals(true, future.get()); + } + + @Test + public void testTimeoutWaitForLocks() throws Exception { + final Object object1 = new Object(); + counter.add(object1); + assertEquals(false, counter.await(200L, TimeUnit.MILLISECONDS)); + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java index d946f874..0a2b4a58 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java @@ -183,12 +183,12 @@ public class MessageListenerContainerLifecycleIntegrationTests { int messagesReceivedBeforeStart = listener.getCount(); container.start(); - assertEquals(concurrentConsumers, container.getActiveConsumerCount()); int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); waited = latch.await(timeout, TimeUnit.SECONDS); logger.info("All messages received after start: " + waited); + assertEquals(concurrentConsumers, container.getActiveConsumerCount()); if (transactional) { assertTrue("Timed out waiting for message", waited); } else { @@ -203,7 +203,7 @@ public class MessageListenerContainerLifecycleIntegrationTests { } finally { // Wait for broker communication to finish before trying to stop // container - Thread.sleep(300L); + Thread.sleep(500L); container.shutdown(); assertEquals(0, container.getActiveConsumerCount()); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java index 0e833a52..2b2f4d55 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java @@ -23,6 +23,7 @@ import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionProxy; import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; +import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.MessageListenerBrokerInterruptionIntegrationTests.VanillaListener; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; @@ -74,6 +75,63 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { } } + @Test + public void testListenerSendsMessageAndThenCommit() throws Exception { + + ConnectionFactory connectionFactory = createConnectionFactory(); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + Queue sendQueue = new Queue("test.send"); + new RabbitAdmin(connectionFactory).declareQueue(sendQueue); + + acknowledgeMode = AcknowledgeMode.AUTO; + transactional = true; + + CountDownLatch latch = new CountDownLatch(1); + container = createContainer(queue.getName(), new ChannelSenderListener(sendQueue.getName(), latch, false), + connectionFactory); + template.convertAndSend(queue.getName(), "foo"); + + int timeout = getTimeout(); + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + boolean waited = latch.await(timeout, TimeUnit.SECONDS); + assertTrue("Timed out waiting for message", waited); + + // All messages committed + assertEquals("bar", new String((byte[])template.receiveAndConvert(sendQueue.getName()))); + assertNull(template.receiveAndConvert(queue.getName())); + + } + + @Test + public void testListenerSendsMessageAndThenRollback() throws Exception { + + ConnectionFactory connectionFactory = createConnectionFactory(); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + Queue sendQueue = new Queue("test.send"); + new RabbitAdmin(connectionFactory).declareQueue(sendQueue); + + acknowledgeMode = AcknowledgeMode.AUTO; + transactional = true; + + CountDownLatch latch = new CountDownLatch(1); + container = createContainer(queue.getName(), new ChannelSenderListener(sendQueue.getName(), latch, true), + connectionFactory); + template.convertAndSend(queue.getName(), "foo"); + + int timeout = getTimeout(); + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + boolean waited = latch.await(timeout, TimeUnit.SECONDS); + assertTrue("Timed out waiting for message", waited); + + container.stop(); + + // Foo message is redelivered + assertEquals("foo", template.receiveAndConvert(queue.getName())); + // Sending of bar message is also rolled back + assertNull(template.receiveAndConvert(sendQueue.getName())); + + } + @Test public void testListenerRecoversFromBogusDoubleAck() throws Exception { @@ -87,7 +145,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { template.convertAndSend(queue.getName(), i + "foo"); } - int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); + int timeout = getTimeout(); logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); boolean waited = latch.await(timeout, TimeUnit.SECONDS); assertTrue("Timed out waiting for message", waited); @@ -107,7 +165,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { template.convertAndSend(queue.getName(), i + "foo"); } - int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); + int timeout = getTimeout(); logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); boolean waited = latch.await(timeout, TimeUnit.SECONDS); assertTrue("Timed out waiting for message", waited); @@ -123,17 +181,18 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { CountDownLatch latch = new CountDownLatch(messageCount); container = createContainer(queue.getName(), new AbortChannelListener(latch), createConnectionFactory()); + Thread.sleep(500L); assertEquals(concurrentConsumers, container.getActiveConsumerCount()); for (int i = 0; i < messageCount; i++) { template.convertAndSend(queue.getName(), i + "foo"); } - int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); + int timeout = getTimeout(); 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())); assertEquals(concurrentConsumers, container.getActiveConsumerCount()); @@ -149,8 +208,9 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { CountDownLatch latch = new CountDownLatch(messageCount); ConnectionFactory connectionFactory = createConnectionFactory(); - container = createContainer(queue.getName(), new CloseConnectionListener((ConnectionProxy) connectionFactory.createConnection(), - latch), connectionFactory); + container = createContainer(queue.getName(), + new CloseConnectionListener((ConnectionProxy) connectionFactory.createConnection(), latch), + connectionFactory); for (int i = 0; i < messageCount; i++) { template.convertAndSend(queue.getName(), i + "foo"); } @@ -178,7 +238,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { template.convertAndSend(queue.getName(), i + "foo"); } - int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); + int timeout = getTimeout(); logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); boolean waited = latch.await(timeout, TimeUnit.SECONDS); assertTrue("Timed out waiting for message", waited); @@ -187,8 +247,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { assertNull(template.receiveAndConvert(queue.getName())); } - - @Test(expected=AmqpIllegalStateException.class) + @Test(expected = AmqpIllegalStateException.class) public void testListenerDoesNotRecoverFromMissingQueue() throws Exception { // TODO: with only 1 this test tends to fail concurrentConsumers = 3; @@ -196,7 +255,12 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { container = createContainer("nonexistent", new VanillaListener(latch), createConnectionFactory()); } - private SimpleMessageListenerContainer createContainer(String queueName, Object listener, ConnectionFactory connectionFactory) { + private int getTimeout() { + return Math.min(1 + messageCount / (4 * concurrentConsumers), 30); + } + + private SimpleMessageListenerContainer createContainer(String queueName, Object listener, + ConnectionFactory connectionFactory) { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); container.setMessageListener(new MessageListenerAdapter(listener)); container.setQueueNames(queueName); @@ -235,6 +299,36 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { } } + public static class ChannelSenderListener implements ChannelAwareMessageListener { + + private final CountDownLatch latch; + + private final boolean fail; + + private final String queueName; + + public ChannelSenderListener(String queueName, CountDownLatch latch, boolean fail) { + this.queueName = queueName; + this.latch = latch; + this.fail = fail; + } + + public void onMessage(Message message, Channel channel) throws Exception { + String value = new String(message.getBody()); + try { + logger.debug("Received: " + value + " Sending: bar"); + channel.basicPublish("", queueName, null, "bar".getBytes()); + if (fail) { + logger.debug("Failing (planned)"); + // intentional error (causes exception on connection thread): + throw new RuntimeException("Planned"); + } + } finally { + latch.countDown(); + } + } + } + public static class AbortChannelListener implements ChannelAwareMessageListener { private AtomicBoolean failed = new AtomicBoolean(false);