diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java index 405c8190..31b42bee 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -198,7 +198,7 @@ public abstract class RabbitUtils { && ((AMQP.Channel.Close) shutdownReason).getMethodId() == 10); // declare } - protected static Object determineShutdownReason(ShutdownSignalException sig) { + public static Object determineShutdownReason(ShutdownSignalException sig) { if (shutDownSignalReasonMethod == null) { return false; } 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 fef1dfe2..968c7dec 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 @@ -113,7 +113,11 @@ public class BlockingQueueConsumer { private final Set missingQueues = Collections.synchronizedSet(new HashSet()); - private final long retryDeclarationInterval = 60000; + private long retryDeclarationInterval = 60000; + + private long failedDeclarationRetryInterval = 5000; + + private int declarationRetries = 3; private long lastRetryDeclaration; @@ -233,6 +237,36 @@ public class BlockingQueueConsumer { public final void setQuiesce(long shutdownTimeout) { } + /** + * Set the number of retries after passive queue declaration fails. + * @param declarationRetries The number of retries, default 3. + * @see #setFailedDeclarationRetryInterval(long) + * @since 1.3.9 + */ + public void setDeclarationRetries(int declarationRetries) { + this.declarationRetries = declarationRetries; + } + + /** + * Set the interval between passive queue declaration attempts in milliseconds. + * @param failedDeclarationRetryInterval the interval, default 5000. + * @see #setDeclarationRetries(int) + * @since 1.3.9 + */ + public void setFailedDeclarationRetryInterval(long failedDeclarationRetryInterval) { + this.failedDeclarationRetryInterval = failedDeclarationRetryInterval; + } + + /** + * When consuming multiple queues, set the interval between declaration attempts when only + * a subset of the queues were available (milliseconds). + * @param retryDeclarationInterval the interval, default 60000. + * @since 1.3.9 + */ + public void setRetryDeclarationInterval(long retryDeclarationInterval) { + this.retryDeclarationInterval = retryDeclarationInterval; + } + protected void basicCancel() { for (String consumerTag : this.consumerTags.keySet()) { try { @@ -396,21 +430,21 @@ public class BlockingQueueConsumer { this.activeObjectCounter.add(this); // mirrored queue might be being moved - int passiveDeclareTries = 3; + int passiveDeclareRetries = this.declarationRetries; do { try { attemptPassiveDeclarations(); - if (passiveDeclareTries < 3 && logger.isInfoEnabled()) { + if (passiveDeclareRetries < this.declarationRetries && logger.isInfoEnabled()) { logger.info("Queue declaration succeeded after retrying"); } - passiveDeclareTries = 0; + passiveDeclareRetries = 0; } catch (DeclarationException e) { - if (passiveDeclareTries > 0 && channel.isOpen()) { + if (passiveDeclareRetries > 0 && channel.isOpen()) { if (logger.isWarnEnabled()) { - logger.warn("Queue declaration failed; retries left=" + (passiveDeclareTries-1), e); + logger.warn("Queue declaration failed; retries left=" + (passiveDeclareRetries), e); try { - Thread.sleep(5000); + Thread.sleep(this.failedDeclarationRetryInterval); } catch (InterruptedException e1) { Thread.currentThread().interrupt(); @@ -432,7 +466,7 @@ public class BlockingQueueConsumer { } } } - while (passiveDeclareTries-- > 0); + while (passiveDeclareRetries-- > 0); if (!acknowledgeMode.isAutoAck()) { // Set basicQos before calling basicConsume (otherwise if we are not acking the broker @@ -483,8 +517,11 @@ public class BlockingQueueConsumer { if (logger.isWarnEnabled()) { logger.warn("Failed to declare queue:" + queueName); } + if (!this.channel.isOpen()) { + throw new AmqpIOException(e); + } if (failures == null) { - failures = new DeclarationException(); + failures = new DeclarationException(e); } failures.addFailedQueue(queueName); } @@ -626,6 +663,10 @@ public class BlockingQueueConsumer { super("Failed to declare queue(s):"); } + public DeclarationException(Throwable t) { + super("Failed to declare queue(s):", t); + } + private final List failedQueues = new ArrayList(); void addFailedQueue(String queue) { 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 94f68c52..20dae7cf 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -170,6 +170,12 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private ContainerDelegate proxy = delegate; + private Integer declarationRetries; + + private Long failedDeclarationRetryInterval; + + private Long retryDeclarationInterval; + /** * Default constructor for convenient dependency injection via setters. */ @@ -546,6 +552,36 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } + /** + * Set the number of retries after passive queue declaration fails. + * @param declarationRetries The number of retries, default 3. + * @see #setFailedDeclarationRetryInterval(long) + * @since 1.3.9 + */ + public void setDeclarationRetries(int declarationRetries) { + this.declarationRetries = declarationRetries; + } + + /** + * Set the interval between passive queue declaration attempts in milliseconds. + * @param failedDeclarationRetryInterval the interval, default 5000. + * @see #setDeclarationRetries(int) + * @since 1.3.9 + */ + public void setFailedDeclarationRetryInterval(long failedDeclarationRetryInterval) { + this.failedDeclarationRetryInterval = failedDeclarationRetryInterval; + } + + /** + * When consuming multiple queues, set the interval between declaration attempts when only + * a subset of the queues were available (milliseconds). + * @param retryDeclarationInterval the interval, default 60000. + * @since 1.3.9 + */ + public void setRetryDeclarationInterval(long retryDeclarationInterval) { + this.retryDeclarationInterval = retryDeclarationInterval; + } + /** * Avoid the possibility of not configuring the CachingConnectionFactory in sync with the number of concurrent * consumers. @@ -852,6 +888,15 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta consumer = new BlockingQueueConsumer(getConnectionFactory(), this.messagePropertiesConverter, cancellationLock, getAcknowledgeMode(), isChannelTransacted(), actualPrefetchCount, this.defaultRequeueRejected, this.consumerArgs, this.exclusive, queues); + if (this.declarationRetries != null) { + consumer.setDeclarationRetries(this.declarationRetries); + } + if (this.failedDeclarationRetryInterval != null) { + consumer.setFailedDeclarationRetryInterval(this.failedDeclarationRetryInterval); + } + if (this.retryDeclarationInterval != null) { + consumer.setRetryDeclarationInterval(this.retryDeclarationInterval); + } return consumer; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java index bce97802..ef2b77b6 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -113,6 +113,7 @@ public class BlockingQueueConsumerTests { when(connectionFactory.createConnection()).thenReturn(connection); when(connection.createChannel(Mockito.anyBoolean())).thenReturn(channel); + when(channel.isOpen()).thenReturn(true); when(channel.queueDeclarePassive(Mockito.anyString())) .then(new Answer() { @@ -134,6 +135,9 @@ public class BlockingQueueConsumerTests { new DefaultMessagePropertiesConverter(), new ActiveObjectCounter(), AcknowledgeMode.AUTO, true, 20, "good", "bad"); + blockingQueueConsumer.setDeclarationRetries(1); + blockingQueueConsumer.setRetryDeclarationInterval(10); + blockingQueueConsumer.setFailedDeclarationRetryInterval(10); blockingQueueConsumer.start(); verify(channel).basicQos(20); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoverySingleConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoverySingleConnectionIntegrationTests.java deleted file mode 100644 index 8487c0b1..00000000 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoverySingleConnectionIntegrationTests.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * Copyright 2002-2014 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 org.springframework.amqp.rabbit.connection.ConnectionFactory; -import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; -import org.springframework.amqp.rabbit.test.BrokerTestUtils; - -public class MessageListenerRecoverySingleConnectionIntegrationTests extends MessageListenerRecoveryCachingConnectionIntegrationTests { - - @Override - protected ConnectionFactory createConnectionFactory() { - SingleConnectionFactory connectionFactory = new SingleConnectionFactory(); - connectionFactory.setHost("localhost"); - connectionFactory.setPort(BrokerTestUtils.getPort()); - return connectionFactory; - } - -} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index 27178f53..a1cdd858 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java @@ -58,6 +58,7 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.support.GenericApplicationContext; +import com.rabbitmq.client.AMQP.Queue.DeclareOk; import com.rabbitmq.client.Channel; /** @@ -318,6 +319,86 @@ public class SimpleMessageListenerContainerIntegration2Tests { assertTrue(latch.await(10, TimeUnit.SECONDS)); assertTrue(networkGlitch.get()); + + container.stop(); + } + + @Test + public void testRestartConsumerOnConnectionLossDuringQueueDeclare() throws Exception { + this.template.convertAndSend(queue.getName(), "foo"); + + ConnectionFactory connectionFactory = new CachingConnectionFactory("localhost", BrokerTestUtils.getPort()); + + final AtomicBoolean networkGlitch = new AtomicBoolean(); + + class MockChannel extends PublisherCallbackChannelImpl { + + public MockChannel(Channel delegate) { + super(delegate); + } + + @Override + public DeclareOk queueDeclarePassive(String queue) throws IOException { + if (networkGlitch.compareAndSet(false, true)) { + getConnection().close(); + throw new IOException("Intentional connection reset"); + } + return super.queueDeclarePassive(queue); + } + + } + + Connection connection = spy(connectionFactory.createConnection()); + when(connection.createChannel(anyBoolean())).then(new Answer() { + + @Override + public Channel answer(InvocationOnMock invocation) throws Throwable { + return new MockChannel((Channel) invocation.callRealMethod()); + } + + }); + + DirectFieldAccessor dfa = new DirectFieldAccessor(connectionFactory); + dfa.setPropertyValue("connection", connection); + + CountDownLatch latch = new CountDownLatch(1); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); + container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch))); + container.setQueueNames(queue.getName()); + container.setRecoveryInterval(500); + container.afterPropertiesSet(); + container.start(); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(networkGlitch.get()); + + container.stop(); + } + + @Test + public void testRestartConsumerMissingQueue() throws Exception { + Queue queue = new AnonymousQueue(); + this.template.convertAndSend(queue.getName(), "foo"); + + ConnectionFactory connectionFactory = new CachingConnectionFactory("localhost", BrokerTestUtils.getPort()); + + CountDownLatch latch = new CountDownLatch(1); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); + container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch))); + container.setQueues(queue); + container.setRecoveryInterval(500); + container.setMissingQueuesFatal(false); + container.setDeclarationRetries(1); + container.setFailedDeclarationRetryInterval(100); + container.afterPropertiesSet(); + container.start(); + + new RabbitAdmin(connectionFactory).declareQueue(queue); + this.template.convertAndSend(queue.getName(), "foo"); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + + container.stop(); } private boolean containerStoppedForAbortWithBadListener() throws InterruptedException {