From f1d783cf5c09b16c06263b2a1475ef12658aa3ef Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Mon, 14 Mar 2011 16:48:47 +0000 Subject: [PATCH] AMQP-114: tweak concurrency protection in consumer startup detection --- .../listener/SimpleMessageListenerContainer.java | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) 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 3785d395..079aad77 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 @@ -19,6 +19,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import org.aopalliance.aop.Advice; import org.springframework.amqp.AmqpException; @@ -431,7 +432,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private final CountDownLatch start; - private ListenerStartupFatalException startupException; + private volatile ListenerStartupFatalException startupException; + + private AtomicBoolean started = new AtomicBoolean(false); public AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer, CountDownLatch latch) { this.consumer = consumer; @@ -441,13 +444,16 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta /** * Retrieve the fatal startup exception if this processor completely failed to locate the broker resources it - * needed. Blocks up to 60 seconds waiting (but should always return promptly in normal circumstances). + * needed. Blocks up to 60 seconds waiting (but should always return promptly in normal circumstances). * * @return a startup exception if there was one * @throws TimeoutException if the consumer hasn't started * @throws InterruptedException if the consumer startup is interrupted */ public ListenerStartupFatalException getStartupException() throws TimeoutException, InterruptedException { + if (!started.get()) { + return null; + } if (!start.await(60000L, TimeUnit.MILLISECONDS)) { throw new TimeoutException("Timed out waiting for startup"); } @@ -457,6 +463,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta public void run() { boolean aborted = false; + this.started.set(true); try {