Reinstate DynamicDestinationsBindable, remove utilities

- add DynamicDestinationsBindable back
- add tests validating that the unbind of the dynamic channels takes place
- Rename tests (original test renamed back) to use the `BinderAwareChannelResolver`
- Remove utility class for property validation and extraction

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binding;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver;
@@ -40,11 +41,14 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
 
 	private final BindableChannelFactory bindableChannelFactory;
 
+	private final DynamicDestinationsBindable dynamicDestinationsBindable;
+
 	private ConfigurableListableBeanFactory beanFactory;
 
 	@SuppressWarnings("unchecked")
 	public BinderAwareChannelResolver(ChannelBindingService channelBindingService,
-			BindableChannelFactory bindableChannelFactory) {
+			BindableChannelFactory bindableChannelFactory, DynamicDestinationsBindable dynamicDestinationsBindable) {
+		this.dynamicDestinationsBindable = dynamicDestinationsBindable;
 		Assert.notNull(channelBindingService, "'channelBindingService' cannot be null");
 		Assert.notNull(bindableChannelFactory, "'bindableChannelFactory' cannot be null");
 		this.channelBindingService = channelBindingService;
@@ -83,7 +87,8 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
 					channel = this.bindableChannelFactory.createSubscribableChannel(channelName);
 					this.beanFactory.registerSingleton(channelName, channel);
 					channel = (MessageChannel) this.beanFactory.initializeBean(channel, channelName);
-					this.channelBindingService.bindProducer(channel, channelName);
+					Binding<MessageChannel> binding = this.channelBindingService.bindProducer(channel, channelName);
+					this.dynamicDestinationsBindable.addOutputBinding(channelName, binding);
 				}
 				else {
 					throw destinationResolutionException;

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java

@@ -25,16 +25,21 @@ import java.util.Map;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.beans.BeanUtils;
+import org.springframework.boot.bind.RelaxedDataBinder;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.BinderFactory;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
+import org.springframework.validation.beanvalidation.CustomValidatorBean;
 
 /**
  * Handles the operations related to channel binding including binding of input/output channels by delegating
@@ -47,6 +52,8 @@ import org.springframework.util.StringUtils;
  */
 public class ChannelBindingService {
 
+	private final CustomValidatorBean validator;
+
 	private final Log log = LogFactory.getLog(ChannelBindingService.class);
 
 	private BinderFactory<MessageChannel> binderFactory;
@@ -61,8 +68,11 @@ public class ChannelBindingService {
 			BinderFactory<MessageChannel> binderFactory) {
 		this.channelBindingServiceProperties = channelBindingServiceProperties;
 		this.binderFactory = binderFactory;
+		this.validator = new CustomValidatorBean();
+		this.validator.afterPropertiesSet();
 	}
 
+
 	@SuppressWarnings("unchecked")
 	public Collection<Binding<MessageChannel>> bindConsumer(MessageChannel inputChannel, String inputChannelName) {
 		String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(inputChannelName);
@@ -73,10 +83,12 @@ public class ChannelBindingService {
 		ConsumerProperties consumerProperties =
 				this.channelBindingServiceProperties.getConsumerProperties(inputChannelName);
 		if (binder instanceof ExtendedPropertiesBinder) {
-			consumerProperties = ProducerConsumerPropertiesUtil.getExtendedConsumerProperties((ExtendedPropertiesBinder) binder,
-					inputChannelName, consumerProperties);
+			Object extension = ((ExtendedPropertiesBinder) binder).getExtendedConsumerProperties(inputChannelName);
+			ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties(extension);
+			BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties);
+			consumerProperties = extendedConsumerProperties;
 		}
-		ProducerConsumerPropertiesUtil.validate(consumerProperties);
+		validate(consumerProperties);
 		for (String target : channelBindingTargets) {
 			Binding<MessageChannel> binding = binder.bindConsumer(target, channelBindingServiceProperties.getGroup(inputChannelName), inputChannel, consumerProperties);
 			bindings.add(binding);
@@ -92,10 +104,12 @@ public class ChannelBindingService {
 				(Binder<MessageChannel, ?, ProducerProperties>) getBinderForChannel(outputChannelName);
 		ProducerProperties producerProperties = this.channelBindingServiceProperties.getProducerProperties(outputChannelName);
 		if (binder instanceof ExtendedPropertiesBinder) {
-			producerProperties = ProducerConsumerPropertiesUtil.getExtendedProducerProperties((ExtendedPropertiesBinder) binder,
-					outputChannelName, producerProperties);
+			Object extension = ((ExtendedPropertiesBinder) binder).getExtendedProducerProperties(outputChannelName);
+			ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(extension);
+			BeanUtils.copyProperties(producerProperties, extendedProducerProperties);
+			producerProperties = extendedProducerProperties;
 		}
-		ProducerConsumerPropertiesUtil.validate(producerProperties);
+		validate(producerProperties);
 		Binding<MessageChannel> binding = binder.bindProducer(channelBindingTarget, outputChannel, producerProperties);
 		this.producerBindings.put(outputChannelName, binding);
 		return binding;
@@ -131,4 +145,13 @@ public class ChannelBindingService {
 	public ChannelBindingServiceProperties getChannelBindingServiceProperties() {
 		return this.channelBindingServiceProperties;
 	}
+
+	private void validate(Object properties) {
+		RelaxedDataBinder dataBinder = new RelaxedDataBinder(properties);
+		dataBinder.setValidator(validator);
+		dataBinder.validate();
+		if (dataBinder.getBindingResult().hasErrors()) {
+			throw new IllegalStateException(dataBinder.getBindingResult().toString());
+		}
+	}
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicDestinationsBindable.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cloud.stream.binding;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.springframework.cloud.stream.binder.Binding;
+
+/**
+ * A {@link BindableAdapter} that stores the dynamic destination names and handles their unbinding.
+ *
+ * This class is not thread-safe.
+ *
+ * @author Ilayaperumal Gopinathan
+ */
+public final class DynamicDestinationsBindable extends BindableAdapter {
+
+	/**
+	 * Map containing dynamic channel names and their bindings.
+	 */
+	private Map<String, Binding> outputBindings = new HashMap<>();
+
+	public void addOutputBinding(String name, Binding binding) {
+		this.outputBindings.put(name, binding);
+	}
+
+	@Override
+	public Set<String> getOutputs() {
+		return Collections.unmodifiableSet(outputBindings.keySet());
+	}
+
+	@Override
+	public void unbindOutputs(ChannelBindingService adapter) {
+		for (Map.Entry<String, Binding> entry: outputBindings.entrySet()) {
+			entry.getValue().unbind();
+		}
+		outputBindings.clear();
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ProducerConsumerPropertiesUtil.java

@@ -1,65 +0,0 @@
-/*
- * Copyright 2016 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.cloud.stream.binding;
-
-import org.springframework.beans.BeanUtils;
-import org.springframework.boot.bind.RelaxedDataBinder;
-import org.springframework.cloud.stream.binder.ConsumerProperties;
-import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
-import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
-import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
-import org.springframework.cloud.stream.binder.ProducerProperties;
-import org.springframework.validation.beanvalidation.CustomValidatorBean;
-
-/**
- * Helper class for operations involving producer/consumer properties.
- *
- * @author Ilayaperumal Gopinathan
- */
-public class ProducerConsumerPropertiesUtil {
-
-		private static CustomValidatorBean validator;
-
-		static {
-			validator = new CustomValidatorBean();
-			validator.afterPropertiesSet();
-		}
-
-		static void validate(Object properties) {
-			RelaxedDataBinder dataBinder = new RelaxedDataBinder(properties);
-			dataBinder.setValidator(validator);
-			dataBinder.validate();
-			if (dataBinder.getBindingResult().hasErrors()) {
-				throw new IllegalStateException(dataBinder.getBindingResult().toString());
-			}
-		}
-
-		static ProducerProperties getExtendedProducerProperties(ExtendedPropertiesBinder extendedPropertiesBinder, String outputChannelName,
-				ProducerProperties producerProperties) {
-			Object extension = extendedPropertiesBinder.getExtendedProducerProperties(outputChannelName);
-			ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(extension);
-			BeanUtils.copyProperties(producerProperties, extendedProducerProperties);
-			return extendedProducerProperties;
-		}
-
-		static ConsumerProperties getExtendedConsumerProperties(ExtendedPropertiesBinder extendedPropertiesBinder, String inputChannelName,
-				ConsumerProperties consumerProperties) {
-			Object extension = extendedPropertiesBinder.getExtendedConsumerProperties(inputChannelName);
-			ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties(extension);
-			BeanUtils.copyProperties(consumerProperties, extendedConsumerProperties);
-			return extendedConsumerProperties;
-		}
-}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.config;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -33,12 +35,13 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.BinderFactory;
 import org.springframework.cloud.stream.binding.BindableChannelFactory;
+import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
 import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor;
 import org.springframework.cloud.stream.binding.ChannelBindingService;
 import org.springframework.cloud.stream.binding.CompositeMessageChannelConfigurer;
 import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener;
 import org.springframework.cloud.stream.binding.DefaultBindableChannelFactory;
-import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
+import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
 import org.springframework.cloud.stream.binding.InputBindingLifecycle;
 import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
 import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
@@ -65,8 +68,6 @@ import org.springframework.messaging.handler.annotation.support.MessageHandlerMe
 import org.springframework.tuple.spel.TuplePropertyAccessor;
 import org.springframework.util.CollectionUtils;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-
 /**
  * Configuration class that provides necessary beans for {@link MessageChannel} binding.
  *
@@ -145,8 +146,8 @@ public class ChannelBindingServiceConfiguration {
 
 	@Bean
 	public BinderAwareChannelResolver binderAwareChannelResolver(ChannelBindingService channelBindingService,
-			BindableChannelFactory bindableChannelFactory) {
-		return new BinderAwareChannelResolver(channelBindingService, bindableChannelFactory);
+			BindableChannelFactory bindableChannelFactory, DynamicDestinationsBindable dynamicDestinationsBindable) {
+		return new BinderAwareChannelResolver(channelBindingService, bindableChannelFactory, dynamicDestinationsBindable);
 	}
 
 	@Bean
@@ -156,6 +157,11 @@ public class ChannelBindingServiceConfiguration {
 		return new SingleChannelBindable(ERROR_CHANNEL_NAME, errorChannel);
 	}
 
+	@Bean
+	public DynamicDestinationsBindable dynamicDestinationsBindable() {
+		return new DynamicDestinationsBindable();
+	}
+
 	@Bean
 	public CompositeMessageConverterFactory compositeMessageConverterFactory() {
 		List<AbstractFromMessageConverter> messageConverters = new ArrayList<>();

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

@@ -16,8 +16,10 @@
 
 package org.springframework.cloud.stream.binder;
 
+import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
@@ -44,10 +46,13 @@ import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.cloud.stream.binding.BindableChannelFactory;
+import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
 import org.springframework.cloud.stream.binding.ChannelBindingService;
 import org.springframework.cloud.stream.binding.DefaultBindableChannelFactory;
-import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
+import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
+import org.springframework.cloud.stream.binding.InputBindingLifecycle;
 import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
+import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
@@ -68,7 +73,7 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @author Ilayaperumal Gopinathan
  */
-public class DynamicDestinationResolverTests {
+public class BinderAwareChannelResolverTests {
 
 	protected final StaticApplicationContext context = new StaticApplicationContext();
 
@@ -80,8 +85,13 @@ public class DynamicDestinationResolverTests {
 
 	protected volatile ChannelBindingServiceProperties channelBindingServiceProperties;
 
+	protected volatile DynamicDestinationsBindable dynamicDestinationsBindable;
+
+	private volatile List<TestBinder.TestBinding> producerBindings;
+
 	@Before
 	public void setupContext() throws Exception {
+		producerBindings = new ArrayList<>();
 		this.binder = new TestBinder();
 		BinderFactory binderFactory = new BinderFactory<MessageChannel>() {
 
@@ -103,19 +113,28 @@ public class DynamicDestinationResolverTests {
 		messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class));
 		messageConverterConfigurer.afterPropertiesSet();
 		this.bindableChannelFactory = new DefaultBindableChannelFactory(messageConverterConfigurer);
-		this.resolver = new BinderAwareChannelResolver(channelBindingService, this.bindableChannelFactory);
+		dynamicDestinationsBindable = new DynamicDestinationsBindable();
+		this.resolver = new BinderAwareChannelResolver(channelBindingService, this.bindableChannelFactory,
+				dynamicDestinationsBindable);
 		this.resolver.setBeanFactory(context.getBeanFactory());
-		context.getBeanFactory().registerSingleton("channelResolver",
-				this.resolver);
+		context.getBeanFactory().registerSingleton("channelResolver", this.resolver);
+		context.getBeanFactory().registerSingleton("dynamicDestinationBindable", this.dynamicDestinationsBindable);
 		context.registerSingleton("other", DirectChannel.class);
 		context.registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
 				DefaultMessageBuilderFactory.class);
+		context.getBeanFactory().registerSingleton("channelBindingService", channelBindingService);
+		context.registerSingleton("inputBindingLifecycle", InputBindingLifecycle.class);
+		context.registerSingleton("outputBindingLifecycle", OutputBindingLifecycle.class);
 		context.refresh();
 	}
 
 	@Test
 	public void resolveChannel() {
+		assertThat(producerBindings, hasSize(0));
 		MessageChannel registered = resolver.resolveDestination("foo");
+		assertThat(producerBindings, hasSize(1));
+		TestBinder.TestBinding binding = producerBindings.get(0);
+		assertTrue("Must be bound", binding.isBound());
 		DirectChannel testChannel = new DirectChannel();
 		final CountDownLatch latch = new CountDownLatch(1);
 		final List<Message<?>> received = new ArrayList<>();
@@ -140,6 +159,8 @@ public class DynamicDestinationResolverTests {
 		assertEquals(1, received.size());
 		assertEquals("hello", received.get(0).getPayload());
 		context.close();
+		assertThat(producerBindings, hasSize(1));
+		assertTrue("Must not be bound", !binding.isBound());
 	}
 
 	@Test
@@ -171,7 +192,7 @@ public class DynamicDestinationResolverTests {
 		ChannelBindingService channelBindingService = new ChannelBindingService(channelBindingServiceProperties, mockBinderFactory);
 		@SuppressWarnings("unchecked")
 		BinderAwareChannelResolver resolver =
-				new BinderAwareChannelResolver(channelBindingService, this.bindableChannelFactory);
+				new BinderAwareChannelResolver(channelBindingService, this.bindableChannelFactory, new DynamicDestinationsBindable());
 		BeanFactory beanFactory = new DefaultListableBeanFactory();
 		resolver.setBeanFactory(beanFactory);
 		SubscribableChannel resolved = (SubscribableChannel) resolver.resolveDestination("foo");
@@ -215,7 +236,9 @@ public class DynamicDestinationResolverTests {
 			DirectHandler directHandler = new DirectHandler(destinations.get(name));
 			// for test purposes we can assume it is a SubscribableChannel
 			((SubscribableChannel) outboundBindTarget).subscribe(directHandler);
-			return new TestBinding(name, directHandler);
+			TestBinding binding = new TestBinding(name, directHandler);
+			producerBindings.add(binding);
+			return binding;
 		}
 
 		private class TestBinding implements Binding<MessageChannel> {
@@ -224,6 +247,8 @@ public class DynamicDestinationResolverTests {
 
 			private final DirectHandler directHandler;
 
+			private boolean bound = true;
+
 			private TestBinding(String name, DirectHandler directHandler) {
 				this.name = name;
 				this.directHandler = directHandler;
@@ -231,8 +256,13 @@ public class DynamicDestinationResolverTests {
 
 			@Override
 			public void unbind() {
+				bound = false;
 				destinations.get(name).unsubscribe(directHandler);
 			}
+
+			public boolean isBound() {
+				return bound;
+			}
 		}
 	}
 }

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

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -28,15 +30,19 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
 
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
 import org.springframework.cloud.stream.binding.ChannelBindingService;
 import org.springframework.cloud.stream.binding.DefaultBindableChannelFactory;
-import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
+import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
+import org.springframework.cloud.stream.binding.InputBindingLifecycle;
 import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
+import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
@@ -55,13 +61,16 @@ import org.springframework.messaging.SubscribableChannel;
  * @author Gary Russell
  * @author Ilayaperumal Gopinathan
  */
-public class ExtendedPropertiesDynamicDestinationResolverTests extends DynamicDestinationResolverTests {
+public class ExtendedPropertiesBinderAwareChannelResolverTests extends BinderAwareChannelResolverTests {
 
 	private volatile ExtendedPropertiesBinder<MessageChannel, ExtendedConsumerProperties, ExtendedProducerProperties> binder;
 
+	private volatile List<TestBinder.TestBinding> producerBindings;
+
 	@Before
 	@Override
 	public void setupContext() throws Exception {
+		producerBindings = new ArrayList<>();
 		this.binder = new TestBinder();
 		BinderFactory binderFactory = new BinderFactory<MessageChannel>() {
 
@@ -82,21 +91,31 @@ public class ExtendedPropertiesDynamicDestinationResolverTests extends DynamicDe
 		messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class));
 		messageConverterConfigurer.afterPropertiesSet();
 		this.bindableChannelFactory = new DefaultBindableChannelFactory(messageConverterConfigurer);
-		ChannelBindingService channelBindingService = new ChannelBindingService(channelBindingServiceProperties, binderFactory);
-		this.resolver = new BinderAwareChannelResolver(channelBindingService, bindableChannelFactory);
+		dynamicDestinationsBindable = new DynamicDestinationsBindable();
+		ChannelBindingService channelBindingService = new ChannelBindingService(channelBindingServiceProperties,
+				binderFactory);
+		this.resolver = new BinderAwareChannelResolver(channelBindingService, this.bindableChannelFactory,
+				dynamicDestinationsBindable);
 		this.resolver.setBeanFactory(context.getBeanFactory());
-		context.getBeanFactory().registerSingleton("channelResolver",
-				this.resolver);
+		context.getBeanFactory().registerSingleton("channelResolver", this.resolver);
+		context.getBeanFactory().registerSingleton("dynamicDestinationBindable", this.dynamicDestinationsBindable);
 		context.registerSingleton("other", DirectChannel.class);
 		context.registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
 				DefaultMessageBuilderFactory.class);
+		context.getBeanFactory().registerSingleton("channelBindingService", channelBindingService);
+		context.registerSingleton("inputBindingLifecycle", InputBindingLifecycle.class);
+		context.registerSingleton("outputBindingLifecycle", OutputBindingLifecycle.class);
 		context.refresh();
 	}
 
 	@Test
 	@Override
 	public void resolveChannel() {
+		assertThat(producerBindings, hasSize(0));
 		MessageChannel registered = resolver.resolveDestination("foo");
+		Assert.assertThat(producerBindings, hasSize(1));
+		TestBinder.TestBinding binding = producerBindings.get(0);
+		assertTrue("Must be bound", binding.isBound());
 		DirectChannel testChannel = new DirectChannel();
 		final CountDownLatch latch = new CountDownLatch(1);
 		final List<Message<?>> received = new ArrayList<>();
@@ -121,12 +140,14 @@ public class ExtendedPropertiesDynamicDestinationResolverTests extends DynamicDe
 		assertEquals(1, received.size());
 		assertEquals("hello", received.get(0).getPayload());
 		context.close();
+		Assert.assertThat(producerBindings, hasSize(1));
+		assertTrue("Must not be bound", !binding.isBound());
 	}
 
 	/**
 	 * A simple test binder that creates queues for the destinations. Ignores groups.
 	 */
-	private class TestBinder implements ExtendedPropertiesBinder<MessageChannel, ExtendedConsumerProperties, ExtendedProducerProperties> {
+	class TestBinder implements ExtendedPropertiesBinder<MessageChannel, ExtendedConsumerProperties, ExtendedProducerProperties> {
 
 		private final Map<String, DirectChannel> destinations = new ConcurrentHashMap<>();
 
@@ -155,7 +176,9 @@ public class ExtendedPropertiesDynamicDestinationResolverTests extends DynamicDe
 			DirectHandler directHandler = new DirectHandler(destinations.get(name));
 			// for test purposes we can assume it is a SubscribableChannel
 			((SubscribableChannel) outboundBindTarget).subscribe(directHandler);
-			return new TestBinding(name, directHandler);
+			TestBinding binding = new TestBinding(name, directHandler);
+			producerBindings.add(binding);
+			return binding;
 		}
 
 		@Override
@@ -174,6 +197,8 @@ public class ExtendedPropertiesDynamicDestinationResolverTests extends DynamicDe
 
 			private final DirectHandler directHandler;
 
+			private boolean bound = true;
+
 			private TestBinding(String name, DirectHandler directHandler) {
 				this.name = name;
 				this.directHandler = directHandler;
@@ -181,8 +206,13 @@ public class ExtendedPropertiesDynamicDestinationResolverTests extends DynamicDe
 
 			@Override
 			public void unbind() {
+				bound = false;
 				destinations.get(name).unsubscribe(directHandler);
 			}
+
+			public boolean isBound() {
+				return bound;
+			}
 		}
 	}
 }

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java

@@ -224,7 +224,7 @@ public class ChannelBindingServiceTests {
 				channelBindingService,
 				new DefaultBindableChannelFactory(new MessageConverterConfigurer(
 						properties, new DefaultMessageBuilderFactory(),
-						new CompositeMessageConverterFactory())));
+						new CompositeMessageConverterFactory())), new DynamicDestinationsBindable());
 		ConfigurableListableBeanFactory beanFactory = mock(
 				ConfigurableListableBeanFactory.class);
 		when(beanFactory.getBean("foo", MessageChannel.class))