diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index f0dcbf599..0cf4825de 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder; +import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; @@ -134,6 +135,13 @@ public abstract class AbstractMessageChannelBinder consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(), + new ConsumerProperties()); + DirectFieldAccessor consumerBindingAccessor = new DirectFieldAccessor(consumerBinding); + Object messageProducer = consumerBindingAccessor.getPropertyValue("lifecycle"); + Mockito.verify((Lifecycle) messageProducer).start(); + Mockito.verify((InitializingBean) messageProducer).afterPropertiesSet(); + Mockito.verify((MessageProducer) messageProducer).setOutputChannel(Mockito.any(MessageChannel.class)); + Mockito.verifyNoMoreInteractions(messageProducer); + consumerBinding.unbind(); + Mockito.verify((Lifecycle) messageProducer).stop(); + Mockito.verify((DisposableBean) messageProducer).destroy(); + Mockito.verifyNoMoreInteractions(messageProducer); + + Binding producerBinding = binder.bindProducer("bar", new DirectChannel(), + new ProducerProperties()); + DirectFieldAccessor producerBindingAccessor = new DirectFieldAccessor(producerBinding); + Object messageHandler = producerBindingAccessor.getPropertyValue("lifecycle"); + Mockito.verify((Lifecycle) messageHandler).start(); + Mockito.verify((InitializingBean) messageHandler).afterPropertiesSet(); + Mockito.verifyNoMoreInteractions(messageHandler); + producerBinding.unbind(); + Mockito.verify((Lifecycle) messageHandler).stop(); + Mockito.verify((DisposableBean) messageHandler).destroy(); + Mockito.verifyNoMoreInteractions(messageHandler); + } + + private static class StubMessageChannelBinder extends + AbstractMessageChannelBinder> { + + @SuppressWarnings("unchecked") + private StubMessageChannelBinder() { + super(true, null, Mockito.mock(ProvisioningProvider.class)); + } + + @Override + protected MessageHandler createProducerMessageHandler(ProducerDestination destination, + ProducerProperties producerProperties) throws Exception { + MessageHandler mock = Mockito.mock(MessageHandler.class, Mockito.withSettings() + .extraInterfaces(Lifecycle.class, InitializingBean.class, DisposableBean.class)); + return mock; + } + + @Override + protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, + ConsumerProperties properties) throws Exception { + return Mockito.mock(MessageProducer.class, + Mockito.withSettings().extraInterfaces(Lifecycle.class, InitializingBean.class, + DisposableBean.class)); + } + } + +}