GH-3320: Refine lifecycle control in StdIntFlow (#3322)

* GH-3320: Refine lifecycle control in StdIntFlow

Fixes https://github.com/spring-projects/spring-integration/issues/3320

Turns out that lifecycle control for the whole bunch of components
in one `IntegrationFlow` is useful in fields.

* Change the logic in the `StandardIntegrationFlow` to let to call
`start()` and `stop()` independently how the flow was registered in
the application context.
This way it can be autowired as a `Lifecycle` to let end-user to
avoid the search for proper component in the flow to stop or start
manually - all the components registered with the flow are going
to be stopped or started respectively

* * Add `this.` to class property usage to satisfy Checkstyle

Co-authored-by: Artem Bilan <abilan@vmware.com>

spring-integration-core/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -63,13 +63,12 @@ import org.springframework.messaging.MessageChannel;
  * @see IntegrationFlows
  * @see org.springframework.integration.dsl.context.IntegrationFlowBeanPostProcessor
  * @see org.springframework.integration.dsl.context.IntegrationFlowContext
+ * @see SmartLifecycle
  */
 public class StandardIntegrationFlow implements IntegrationFlow, SmartLifecycle {
 
 	private final Map<Object, String> integrationComponents;
 
-	private final List<SmartLifecycle> lifecycles = new LinkedList<>();
-
 	private MessageChannel inputChannel;
 
 	private boolean running;
@@ -113,11 +112,9 @@ public class StandardIntegrationFlow implements IntegrationFlow, SmartLifecycle
 		if (!this.running) {
 			List<Object> components = new LinkedList<>(this.integrationComponents.keySet());
 			ListIterator<Object> iterator = components.listIterator(this.integrationComponents.size());
-			this.lifecycles.clear();
 			while (iterator.hasPrevious()) {
 				Object component = iterator.previous();
 				if (component instanceof SmartLifecycle) {
-					this.lifecycles.add((SmartLifecycle) component);
 					((SmartLifecycle) component).start();
 				}
 			}
@@ -127,30 +124,29 @@ public class StandardIntegrationFlow implements IntegrationFlow, SmartLifecycle
 
 	@Override
 	public void stop(Runnable callback) {
-		if (this.lifecycles.size() > 0) {
-			AggregatingCallback aggregatingCallback = new AggregatingCallback(this.lifecycles.size(), callback);
-			ListIterator<SmartLifecycle> iterator = this.lifecycles.listIterator(this.lifecycles.size());
-			while (iterator.hasPrevious()) {
-				SmartLifecycle lifecycle = iterator.previous();
+		AggregatingCallback aggregatingCallback = new AggregatingCallback(this.integrationComponents.size(), callback);
+		for (Object component : this.integrationComponents.keySet()) {
+			if (component instanceof SmartLifecycle) {
+				SmartLifecycle lifecycle = (SmartLifecycle) component;
 				if (lifecycle.isRunning()) {
 					lifecycle.stop(aggregatingCallback);
-				}
-				else {
-					aggregatingCallback.run();
+					continue;
 				}
 			}
-		}
-		else {
-			callback.run();
+			aggregatingCallback.run();
 		}
 		this.running = false;
 	}
 
 	@Override
 	public void stop() {
-		ListIterator<SmartLifecycle> iterator = this.lifecycles.listIterator(this.lifecycles.size());
-		while (iterator.hasPrevious()) {
-			iterator.previous().stop();
+		for (Object component : this.integrationComponents.keySet()) {
+			if (component instanceof SmartLifecycle) {
+				SmartLifecycle lifecycle = (SmartLifecycle) component;
+				if (lifecycle.isRunning()) {
+					lifecycle.stop();
+				}
+			}
 		}
 		this.running = false;
 	}

spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java

@@ -42,6 +42,7 @@ import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.Lifecycle;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
@@ -64,6 +65,7 @@ import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.dsl.Pollers;
 import org.springframework.integration.dsl.Transformers;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.integration.gateway.GatewayProxyFactoryBean;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.LoggingHandler;
 import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
@@ -511,6 +513,27 @@ public class IntegrationFlowTests {
 		);
 	}
 
+	@Autowired
+	@Qualifier("controlBusFlow")
+	Lifecycle controlBusFlow;
+
+	@Test
+	public void testStandardIntegrationFlowLifecycle() {
+		this.controlBusFlow.stop();
+
+		GatewayProxyFactoryBean controlBusGateway =
+				this.beanFactory.getBean("&controlBusGateway", GatewayProxyFactoryBean.class);
+		assertThat(controlBusGateway.isRunning()).isFalse();
+		Lifecycle controlBus = this.beanFactory.getBean("controlBus", Lifecycle.class);
+		assertThat(controlBus.isRunning()).isFalse();
+
+		this.controlBusFlow.start();
+
+		assertThat(controlBusGateway.isRunning()).isTrue();
+		assertThat(controlBus.isRunning()).isTrue();
+	}
+
+
 	@After
 	public void cleanUpList() {
 		outputStringList.clear();
@@ -592,8 +615,8 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow controlBusFlow() {
-			return IntegrationFlows.from(ControlBusGateway.class)
-					.controlBus()
+			return IntegrationFlows.from(ControlBusGateway.class, (gateway) -> gateway.beanName("controlBusGateway"))
+					.controlBus((endpoint) -> endpoint.id("controlBus"))
 					.get();
 		}
 
@@ -941,6 +964,7 @@ public class IntegrationFlowTests {
 		public IntegrationFlow interceptorFlow(List<String> outputStringList) {
 			return IntegrationFlows.from("interceptorChannelIn")
 					.intercept(new ChannelInterceptor() {
+
 						@Override
 						public Message<?> preSend(Message<?> message, MessageChannel channel) {
 							outputStringList.add("Pre send transform: " + message.getPayload());
@@ -954,6 +978,7 @@ public class IntegrationFlowTests {
 					})
 					.transform((String s) -> s.toUpperCase())
 					.intercept(new ChannelInterceptor() {
+
 						@Override
 						public Message<?> preSend(Message<?> message, MessageChannel channel) {
 							outputStringList.add("Pre send handle: " + message.getPayload());
@@ -967,6 +992,7 @@ public class IntegrationFlowTests {
 					})
 					.handle(m -> outputStringList.add("Handle: " + m.getPayload())).get();
 		}
+
 	}
 
 	@Service