AMQP-595: Fix Auto-Startup for @RabbitListener

JIRA: https://jira.spring.io/browse/AMQP-595

Ignore container's auto-startup once the context is refreshed.

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<ContextRefreshedEvent> {
 
 	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;

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();