diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java index 388ea6499..bfeb6e509 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java +++ b/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 binding = this.channelBindingService.bindProducer(channel, channelName); + this.dynamicDestinationsBindable.addOutputBinding(channelName, binding); } else { throw destinationResolutionException; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java index 7e0c8fb1d..a9f5fcd3f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java +++ b/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 binderFactory; @@ -61,8 +68,11 @@ public class ChannelBindingService { BinderFactory binderFactory) { this.channelBindingServiceProperties = channelBindingServiceProperties; this.binderFactory = binderFactory; + this.validator = new CustomValidatorBean(); + this.validator.afterPropertiesSet(); } + @SuppressWarnings("unchecked") public Collection> 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 binding = binder.bindConsumer(target, channelBindingServiceProperties.getGroup(inputChannelName), inputChannel, consumerProperties); bindings.add(binding); @@ -92,10 +104,12 @@ public class ChannelBindingService { (Binder) 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 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()); + } + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicDestinationsBindable.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicDestinationsBindable.java new file mode 100644 index 000000000..63a8a0ec7 --- /dev/null +++ b/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 outputBindings = new HashMap<>(); + + public void addOutputBinding(String name, Binding binding) { + this.outputBindings.put(name, binding); + } + + @Override + public Set getOutputs() { + return Collections.unmodifiableSet(outputBindings.keySet()); + } + + @Override + public void unbindOutputs(ChannelBindingService adapter) { + for (Map.Entry entry: outputBindings.entrySet()) { + entry.getValue().unbind(); + } + outputBindings.clear(); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ProducerConsumerPropertiesUtil.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ProducerConsumerPropertiesUtil.java deleted file mode 100644 index d99e5c11a..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ProducerConsumerPropertiesUtil.java +++ /dev/null @@ -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; - } -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java index 993c793a8..a48892f44 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java +++ b/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 messageConverters = new ArrayList<>(); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DynamicDestinationResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java similarity index 84% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DynamicDestinationResolverTests.java rename to spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java index f0de280cd..cccc4f668 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DynamicDestinationResolverTests.java +++ b/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 producerBindings; + @Before public void setupContext() throws Exception { + producerBindings = new ArrayList<>(); this.binder = new TestBinder(); BinderFactory binderFactory = new BinderFactory() { @@ -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> 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 { @@ -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; + } } } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDynamicDestinationResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinderAwareChannelResolverTests.java similarity index 78% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDynamicDestinationResolverTests.java rename to spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinderAwareChannelResolverTests.java index a491e0e21..01fa6ad44 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesDynamicDestinationResolverTests.java +++ b/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 binder; + private volatile List producerBindings; + @Before @Override public void setupContext() throws Exception { + producerBindings = new ArrayList<>(); this.binder = new TestBinder(); BinderFactory binderFactory = new BinderFactory() { @@ -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> 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 { + class TestBinder implements ExtendedPropertiesBinder { private final Map 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; + } } } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java index efa10da34..4ce2ca982 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java +++ b/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))