GH-3234 TestChannelBinder subscriber leak fix

core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java

@@ -100,6 +100,7 @@
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Mattia Gualtieri
  *
  */
 public class TestChannelBinder extends
@@ -158,7 +159,7 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
 
 		IntegrationMessageListeningContainer messageListenerContainer = new IntegrationMessageListeningContainer();
 		IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(
-			messageListenerContainer);
+			siBinderInputChannel, messageListenerContainer);
 		adapter.setBeanFactory(this.beanFactory);
 		String groupName = StringUtils.hasText(group) ? group : "anonymous";
 		ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination,
@@ -172,8 +173,6 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
 			adapter.setErrorChannel(errorInfrastructure.getErrorChannel());
 		}
 
-		siBinderInputChannel.subscribe(messageListenerContainer);
-
 		return adapter;
 	}
 
@@ -219,14 +218,17 @@ public void setMessageListener(Consumer<Message<?>> listener) {
 	private static class IntegrationBinderInboundChannelAdapter
 		extends MessageProducerSupport {
 
+		private final SubscribableChannel siBinderInputChannel;
 		private final IntegrationMessageListeningContainer listenerContainer;
 
 		private RetryTemplate retryTemplate;
 
 		private RecoveryCallback<?> recoveryCallback;
 
 		IntegrationBinderInboundChannelAdapter(
+			SubscribableChannel siBinderInputChannel,
 			IntegrationMessageListeningContainer listenerContainer) {
+			this.siBinderInputChannel = siBinderInputChannel;
 			this.listenerContainer = listenerContainer;
 		}
 
@@ -239,6 +241,17 @@ public void setRetryTemplate(RetryTemplate retryTemplate) {
 			this.retryTemplate = retryTemplate;
 		}
 
+		@Override
+		protected void doStart() {
+			this.siBinderInputChannel.subscribe(this.listenerContainer);
+		}
+
+		@Override
+		protected void doStop() {
+			super.doStop();
+			this.siBinderInputChannel.unsubscribe(this.listenerContainer);
+		}
+
 		@Override
 		protected void onInit() {
 			if (this.retryTemplate != null) {

core/spring-cloud-stream-test-binder/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderTests.java

@@ -17,19 +17,27 @@
 package org.springframework.cloud.stream.binder.test;
 
 import java.lang.reflect.Method;
+import java.util.function.Function;
 
 import org.junit.jupiter.api.Test;
 
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.binding.BindingsLifecycleController;
+import org.springframework.cloud.stream.binding.BindingsLifecycleController.State;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.ReflectionUtils;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 /**
  *
  * @author Oleg Zhurakousky
+ * @author Mattia Gualtieri
  *
  */
 class TestChannelBinderTests {
@@ -72,8 +80,48 @@ public String getName() {
 		}
 	}
 
+	@Test
+	void stoppedBindingDoesNotConsumeAndRestartedBindingConsumesAgain() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(EchoConfiguration.class))
+			.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
+				"--spring.cloud.function.definition=echo",
+				"--spring.cloud.stream.bindings.echo-in-0.group=test-group")) {
+
+			InputDestination input = context.getBean(InputDestination.class);
+			OutputDestination output = context.getBean(OutputDestination.class);
+			BindingsLifecycleController controller = context.getBean(BindingsLifecycleController.class);
+
+			input.send(new GenericMessage<>("first".getBytes()), "echo-in-0");
+			assertThat(output.receive(1000, "echo-out-0").getPayload()).isEqualTo("first".getBytes());
+
+			controller.changeState("echo-in-0", State.STOPPED);
+			assertThat(controller.queryState("echo-in-0").get(0).isRunning()).isFalse();
+
+			input.send(new GenericMessage<>("while-stopped".getBytes()), "echo-in-0");
+			// A stale subscription left behind by an incomplete stop() would still deliver
+			// this message and produce an echo, even though the binding is reported as stopped.
+			assertThat(output.receive(200, "echo-out-0")).isNull();
+
+			controller.changeState("echo-in-0", State.STARTED);
+			assertThat(controller.queryState("echo-in-0").get(0).isRunning()).isTrue();
+
+			input.send(new GenericMessage<>("after-restart".getBytes()), "echo-in-0");
+			assertThat(output.receive(1000, "echo-out-0").getPayload()).isEqualTo("after-restart".getBytes());
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class SampleConfiguration {
 
 	}
+
+	@EnableAutoConfiguration
+	public static class EchoConfiguration {
+
+		@Bean
+		public Function<String, String> echo() {
+			return value -> value;
+		}
+	}
 }