GH-3234 Ensure there are no duplicate subscribers in TestChannelBinder

Resolves #3234

Signed-off-by: Oleg Zhurakousky <ozhurakousky@vmware.com>

Author: olegz

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

@@ -18,6 +18,7 @@
 
 import java.util.function.Consumer;
 
+import org.reactivestreams.Subscription;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
@@ -28,6 +29,8 @@
 import org.springframework.cloud.stream.binder.test.TestChannelBinderProvisioner.SpringIntegrationProducerDestination;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationListener;
 import org.springframework.core.retry.RetryException;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.acks.AcknowledgmentCallback;
@@ -103,7 +106,8 @@
  *
  */
 public class TestChannelBinder extends
-	AbstractMessageChannelBinder<ConsumerProperties, ProducerProperties, TestChannelBinderProvisioner> {
+	AbstractMessageChannelBinder<ConsumerProperties, ProducerProperties, TestChannelBinderProvisioner>
+	implements ApplicationListener<ApplicationEvent> {
 
 	@Autowired
 	private BeanFactory beanFactory;
@@ -157,7 +161,7 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
 			.getChannel();
 
 		IntegrationMessageListeningContainer messageListenerContainer = new IntegrationMessageListeningContainer();
-		IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(
+		IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(siBinderInputChannel,
 			messageListenerContainer);
 		adapter.setBeanFactory(this.beanFactory);
 		String groupName = StringUtils.hasText(group) ? group : "anonymous";
@@ -173,7 +177,6 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
 		}
 
 		siBinderInputChannel.subscribe(messageListenerContainer);
-
 		return adapter;
 	}
 
@@ -225,9 +228,12 @@ private static class IntegrationBinderInboundChannelAdapter
 
 		private RecoveryCallback<?> recoveryCallback;
 
-		IntegrationBinderInboundChannelAdapter(
+		private final SubscribableChannel subscribableChannel;
+
+		IntegrationBinderInboundChannelAdapter(SubscribableChannel subscribableChannel,
 			IntegrationMessageListeningContainer listenerContainer) {
 			this.listenerContainer = listenerContainer;
+			this.subscribableChannel = subscribableChannel;
 		}
 
 		// Temporarily unused until DLQ strategy for this binder becomes a requirement
@@ -239,6 +245,12 @@ public void setRetryTemplate(RetryTemplate retryTemplate) {
 			this.retryTemplate = retryTemplate;
 		}
 
+		@Override
+		protected void doStop() {
+			super.doStop();
+			this.subscribableChannel.unsubscribe(listenerContainer);
+		}
+
 		@Override
 		protected void onInit() {
 			if (this.retryTemplate != null) {
@@ -294,4 +306,9 @@ private void processMessage(Message<?> message) {
 
 	}
 
+	@Override
+	public void onApplicationEvent(ApplicationEvent event) {
+		System.out.println("Event: " + event);
+	}
+
 }