From 5e0ad40113e83cdd416f8c2a01f6ee1b88d99f3b Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Wed, 27 Oct 2021 19:27:09 +0200
Subject: [PATCH] GH-2212 Additional fixes related to lifecycle control over
 components that rely on SPCA

---
 .../cloud/stream/binder/DefaultBinding.java       | 12 ++++++------
 .../function/SourceToFunctionsSupportTests.java   | 15 ++++++++++++---
 2 files changed, 18 insertions(+), 9 deletions(-)

diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java
index 2d1bae052..e3d7c0387 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java
@@ -131,12 +131,12 @@ public class DefaultBinding<T> implements Binding<T> {
 
 	@Override
 	public synchronized void start() {
+		if (this.companion != null) {
+			this.companion.start();
+		}
 		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");
@@ -146,11 +146,11 @@ public class DefaultBinding<T> implements Binding<T> {
 
 	@Override
 	public synchronized void stop() {
+		if (this.companion != null) {
+			this.companion.stop();
+		}
 		if (this.isRunning()) {
 			this.lifecycle.stop();
-			if (this.companion != null) {
-				this.companion.stop();
-			}
 		}
 	}
 
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
index 17a8b846f..d3c99c718 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
@@ -115,7 +115,7 @@ public class SourceToFunctionsSupportTests {
 	}
 
 	@Test
-	public void testImperativeSupplier() {
+	public void testImperativeSupplier() throws Exception {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
 				TestChannelBinderConfiguration.getCompleteConfiguration(
 						FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
@@ -126,8 +126,14 @@ public class SourceToFunctionsSupportTests {
 			String result = new String(target.receive(5000).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");
+
+			lifecycle.stop("number-out-0");
+			for (int i = 0; i < 2; i++) { //drain
+				target.receive(1000);
+			}
+
+			Thread.sleep(2000);
 			assertThat(target.receive(1000)).isNull();
 		}
 	}
@@ -320,7 +326,10 @@ public class SourceToFunctionsSupportTests {
 
 		@Bean
 		public Supplier<String> number() {
-			return () -> String.valueOf(this.counter.incrementAndGet());
+			return () -> {
+				System.out.println("Supplying");
+				return String.valueOf(this.counter.incrementAndGet());
+			};
 		}
 
 		@Bean