GH-2212 Fix Lifecycle control over components that rely on SPCA

This fixes the condition when startiing/stopping message producers which themselves are controlled by SourcsPollingChannelAdapter which we must also start/stop

Resolves #2212

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -255,6 +255,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 								.equals(producerProperties.getHeaderMode()),
 						this.headersToEmbed, useNativeEncoding(producerProperties)));
 
+
 		Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination,
 				outputChannel, producerMessageHandler instanceof Lifecycle
 						? (Lifecycle) producerMessageHandler : null) {
@@ -285,6 +286,13 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			}
 		};
 
+		Lifecycle companion = null;
+		String companionLifecycleName = destination + "_spca";
+		if (this.getApplicationContext().containsBean(companionLifecycleName)) {
+			companion = this.getApplicationContext().getBean(companionLifecycleName, Lifecycle.class);
+		}
+		((DefaultBinding<?>) binding).setCompanion(companion);
+
 		doPublishEvent(new BindingCreatedEvent(binding));
 		return binding;
 	}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java

@@ -61,6 +61,8 @@ public class DefaultBinding<T> implements Binding<T> {
 
 	private boolean restartable;
 
+	private Lifecycle companion;
+
 	/**
 	 * Creates an instance that associates a given name, group and binding target with an
 	 * optional {@link Lifecycle} component, which will be stopped during unbinding.
@@ -84,10 +86,12 @@ public class DefaultBinding<T> implements Binding<T> {
 		this.restartable = true;
 	}
 
+	@Override
 	public String getName() {
 		return this.name;
 	}
 
+	@Override
 	public String getBindingName() {
 		String resolvedName = (this.target instanceof IntegrationObjectSupport)
 				? ((IntegrationObjectSupport) this.target).getComponentName() : getName();
@@ -111,6 +115,7 @@ public class DefaultBinding<T> implements Binding<T> {
 		return state;
 	}
 
+	@Override
 	public boolean isRunning() {
 		return this.lifecycle != null && this.lifecycle.isRunning();
 	}
@@ -129,6 +134,9 @@ public class DefaultBinding<T> implements Binding<T> {
 		if (!this.isRunning()) {
 			if (this.lifecycle != null && this.restartable) {
 				this.lifecycle.start();
+				if (this.companion != null) {
+					this.companion.start();
+				}
 			}
 			else {
 				this.logger.warn("Can not re-bind an anonymous binding");
@@ -140,6 +148,9 @@ public class DefaultBinding<T> implements Binding<T> {
 	public synchronized void stop() {
 		if (this.isRunning()) {
 			this.lifecycle.stop();
+			if (this.companion != null) {
+				this.companion.stop();
+			}
 		}
 	}
 
@@ -199,4 +210,16 @@ public class DefaultBinding<T> implements Binding<T> {
 		return isRunning() ? "running" : "stopped";
 	}
 
+	/**
+	 * Sets the companion Lifecycle.
+	 * In most cases, when dealing with message producer (e.g., Supplier), performing
+	 * any lifecycle operation on it does nothing as we may need to also perform the same operation on its companion
+	 * object (e.g., SourcePollingChannelAdapter)
+	 *
+	 * @param companion instance of companion {@link Lifecycle} object
+	 */
+	public void setCompanion(Lifecycle companion) {
+		this.companion = companion;
+	}
+
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java

@@ -40,6 +40,10 @@ import org.springframework.expression.Expression;
 @JsonInclude(Include.NON_DEFAULT)
 public class ProducerProperties {
 
+	public ProducerProperties() {
+		System.out.println();
+	}
+
 	/**
 	 * Signals if this producer needs to be started automatically. Default: true
 	 */

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -229,7 +229,7 @@ public class FunctionConfiguration {
 							if (functionWrapper != null) {
 								Type functionType = functionWrapper.getFunctionType();
 								IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(functionWrapper, context, producerProperties),
-										beginPublishingTrigger, pollable, context, taskScheduler, functionType)
+										beginPublishingTrigger, pollable, context, taskScheduler, functionType, producerProperties, outputName)
 										.route(Message.class, message -> {
 											if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
 												String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination");
@@ -244,7 +244,7 @@ public class FunctionConfiguration {
 							else {
 								Type functionType = ((FunctionInvocationWrapper) supplier).getFunctionType();
 								IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(supplier, context, producerProperties),
-										beginPublishingTrigger, pollable, context, taskScheduler, functionType)
+										beginPublishingTrigger, pollable, context, taskScheduler, functionType, producerProperties, outputName)
 										.channel(c -> c.direct())
 										.fluxTransform((Function<? super Flux<Message<Object>>, ? extends Publisher<Object>>) function)
 										.route(Message.class, message -> {
@@ -288,7 +288,7 @@ public class FunctionConfiguration {
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier<?> supplier,
 			Publisher<Object> beginPublishingTrigger, PollableBean pollable, GenericApplicationContext context,
-			TaskScheduler taskScheduler, Type functionType) {
+			TaskScheduler taskScheduler, Type functionType, ProducerProperties producerProperties, String bindingName) {
 
 		IntegrationFlowBuilder integrationFlowBuilder;
 
@@ -308,7 +308,10 @@ public class FunctionConfiguration {
 			taskScheduler.schedule(() -> { }, Instant.now()); // will keep AC alive
 		}
 		else { // implies pollable
-			integrationFlowBuilder = IntegrationFlows.fromSupplier(supplier);
+
+			boolean autoStartup = producerProperties != null ? producerProperties.isAutoStartup() : true;
+			integrationFlowBuilder = IntegrationFlows
+					.fromSupplier(supplier, spca -> spca.id(bindingName + "_spca").autoStartup(autoStartup));
 			//only apply the PollableBean attributes if this is a reactive function.
 			if (splittable && reactive) {
 				integrationFlowBuilder = integrationFlowBuilder.split();

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java

@@ -36,6 +36,7 @@ import org.springframework.cloud.function.context.PollableBean;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.test.OutputDestination;
 import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.cloud.stream.binding.BindingsLifecycleController;
 import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
@@ -118,10 +119,14 @@ public class SourceToFunctionsSupportTests {
 						FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
 								"--spring.cloud.stream.function.definition=number",
 								"--spring.jmx.enabled=false")) {
-
+			BindingsLifecycleController lifecycle = context .getBean(BindingsLifecycleController.class);
 			OutputDestination target = context.getBean(OutputDestination.class);
 			String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
 			assertThat(result).isEqualTo("1");
+			result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
+			lifecycle.stop("number-out-0");
+			assertThat(result).isEqualTo("2");
+			assertThat(target.receive(1000)).isNull();
 		}
 	}