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

AMQP-130: always wait for listeners to attempt to start before container decides it has started

Browse Source
This commit is contained in:
Dave Syer
2011-03-25 09:59:01 +00:00
parent 5e784495ed
commit 02215be0d9
3 changed files with 12 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

@@ -57,16 +57,16 @@ public class BlockingQueueConsumer {
private final ConnectionFactory connectionFactory;
private final ActiveObjectCounter<BlockingQueueConsumer> stopped;
private final ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter;
/**
* 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, ActiveObjectCounter<BlockingQueueConsumer> stopped,
public BlockingQueueConsumer(ConnectionFactory connectionFactory, ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter,
AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, String... queues) {
this.connectionFactory = connectionFactory;
this.stopped = stopped;
this.activeObjectCounter = activeObjectCounter;
this.acknowledgeMode = acknowledgeMode;
this.transactional = transactional;
this.prefetchCount = prefetchCount;
@@ -148,7 +148,7 @@ public class BlockingQueueConsumer {
this.channel = ConnectionFactoryUtils.getTransactionalResourceHolder(connectionFactory, transactional)
.getChannel();
this.consumer = new InternalConsumer(channel);
this.stopped.add(this);
this.activeObjectCounter.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)
@@ -157,6 +157,7 @@ public class BlockingQueueConsumer {
channel.queueDeclarePassive(queues[i]);
}
} catch (IOException e) {
this.activeObjectCounter.release(this);
throw new ListenerStartupFatalException("Cannot prepare queue for listener. "
+ "Either the queue doesn't exist or the broker will not allow us to use it.", e);
}
@@ -202,7 +203,7 @@ public class BlockingQueueConsumer {
logger.debug("Received cancellation notice for " + BlockingQueueConsumer.this);
}
// Signal to the container that we have been cancelled
stopped.release(BlockingQueueConsumer.this);
activeObjectCounter.release(BlockingQueueConsumer.this);
}
@Override

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

@@ -19,7 +19,6 @@ 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;
@@ -432,8 +431,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
private volatile ListenerStartupFatalException startupException;
private AtomicBoolean started = new AtomicBoolean(false);
public AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) {
this.consumer = consumer;
this.start = new CountDownLatch(1);
@@ -448,9 +445,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
* @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");
}
@@ -460,7 +454,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
public void run() {
boolean aborted = false;
this.started.set(true);
try {
@@ -506,6 +499,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
}
// In all cases count down to allow container to progress beyond startup
start.countDown();
if (!isActive() || aborted) {
@@ -516,6 +510,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
logger.info("Could not cancel message consumer", e);
}
if (aborted) {
logger.info("Stopping container from aborted consumer");
stop();
}
} else {

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java
View File

@@ -12,7 +12,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.log4j.Level;
import org.junit.After;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.amqp.AmqpIllegalStateException;
@@ -257,8 +256,11 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
}
@Test(expected = AmqpIllegalStateException.class)
@Ignore
public void testSingleListenerDoesNotRecoverFromMissingQueue() throws Exception {
/*
* A single listener sometimes doesn't have time to attempt to start before we ask it if it has failed, so this
* is a good test of that potential bug.
*/
concurrentConsumers = 1;
CountDownLatch latch = new CountDownLatch(messageCount);
container = createContainer("nonexistent", new VanillaListener(latch), createConnectionFactory());