diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java index 7a151500..1c63fa63 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistry.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-2016 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. @@ -34,8 +34,10 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.SmartLifecycle; +import org.springframework.context.event.ContextRefreshedEvent; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -61,7 +63,8 @@ import org.springframework.util.StringUtils; * @see MessageListenerContainer * @see RabbitListenerContainerFactory */ -public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLifecycle, ApplicationContextAware { +public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLifecycle, ApplicationContextAware, + ApplicationListener { protected final Log logger = LogFactory.getLog(getClass()); @@ -72,6 +75,8 @@ public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLife private ConfigurableApplicationContext applicationContext; + private boolean contextRefreshed; + @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { @@ -227,9 +232,7 @@ public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLife @Override public void start() { for (MessageListenerContainer listenerContainer : getListenerContainers()) { - if (listenerContainer.isAutoStartup()) { - startIfNecessary(listenerContainer); - } + startIfNecessary(listenerContainer); } } @@ -261,16 +264,24 @@ public class RabbitListenerEndpointRegistry implements DisposableBean, SmartLife /** * Start the specified {@link MessageListenerContainer} if it should be started - * on startup. + * on startup or when start is called explicitly after startup. * @see MessageListenerContainer#isAutoStartup() */ - private static void startIfNecessary(MessageListenerContainer listenerContainer) { - if (listenerContainer.isAutoStartup()) { + private void startIfNecessary(MessageListenerContainer listenerContainer) { + if (this.contextRefreshed || listenerContainer.isAutoStartup()) { listenerContainer.start(); } } + @Override + public void onApplicationEvent(ContextRefreshedEvent event) { + if (event.getApplicationContext().equals(this.applicationContext)) { + this.contextRefreshed = true; + } + } + + private static class AggregatingCallback implements Runnable { private final AtomicInteger count; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index 48f9a6d7..98b3dc42 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -21,6 +21,7 @@ import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; @@ -61,7 +62,9 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar; +import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException; import org.springframework.amqp.rabbit.test.BrokerRunning; @@ -169,6 +172,9 @@ public class EnableRabbitIntegrationTests { @Autowired private ListenerInterceptor interceptor; + @Autowired + private RabbitListenerEndpointRegistry registry; + /** * Defer queue deletion until after the context has been stopped by the * {@link DirtiesContext}. @@ -212,6 +218,16 @@ public class EnableRabbitIntegrationTests { assertEquals("FOO", rabbitTemplate.convertSendAndReceive("auto.exch", "auto.anon.rk", "foo")); } + @Test + public void autoStart() { + MessageListenerContainer listenerContainer = this.registry.getListenerContainer("notStarted"); + assertNotNull(listenerContainer); + assertFalse(listenerContainer.isRunning()); + this.registry.start(); + assertTrue(listenerContainer.isRunning()); + listenerContainer.stop(); + } + @Test public void autoDeclareAnonWitAtts() { String received = (String) rabbitTemplate.convertSendAndReceive("auto.exch", "auto.anon.atts.rk", "foo"); @@ -647,6 +663,15 @@ public class EnableRabbitIntegrationTests { latch.countDown(); } + @RabbitListener(id="notStarted", containerFactory = "rabbitAutoStartFalseListenerContainerFactory", + bindings = @QueueBinding( + value = @Queue(autoDelete = "true", exclusive="true", durable="true"), + exchange = @Exchange(value = "auto.start", autoDelete = "true"), + key = "auto.start") + ) + public void handleWithAutoStartFalse(String foo) { + } + } public static class Foo1 { @@ -795,6 +820,15 @@ public class EnableRabbitIntegrationTests { return factory; } + @Bean + public SimpleRabbitListenerContainerFactory rabbitAutoStartFalseListenerContainerFactory() { + SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); + factory.setConnectionFactory(rabbitConnectionFactory()); + factory.setReceiveTimeout(10L); + factory.setAutoStartup(false); + return factory; + } + @Bean public SimpleRabbitListenerContainerFactory jsonListenerContainerFactory() { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();