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 ad73c4584..a2a522454 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 @@ -46,16 +46,16 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina private final ChannelBindingServiceProperties channelBindingServiceProperties; - private final DynamicBindable dynamicBindable; + private final DynamicDestinationsBindable dynamicDestinationsBindable; private ConfigurableListableBeanFactory beanFactory; public BinderAwareChannelResolver(BinderFactory binderFactory, - ChannelBindingServiceProperties channelBindingServiceProperties, DynamicBindable dynamicBindable) { + ChannelBindingServiceProperties channelBindingServiceProperties, DynamicDestinationsBindable dynamicDestinationsBindable) { Assert.notNull(binderFactory, "'binderFactory' cannot be null"); this.binderFactory = binderFactory; this.channelBindingServiceProperties = channelBindingServiceProperties; - this.dynamicBindable = dynamicBindable; + this.dynamicDestinationsBindable = dynamicDestinationsBindable; } @Override @@ -105,7 +105,7 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina this.beanFactory.registerSingleton(beanName, channel); channel = (MessageChannel) this.beanFactory.initializeBean(channel, beanName); Binder binder = binderFactory.getBinder(transport); - this.dynamicBindable.addDynamicOutputs(beanName, binder.bindProducer(destinationName, channel, producerProperties)); + this.dynamicDestinationsBindable.addOutputBinding(beanName, binder.bindProducer(destinationName, channel, producerProperties)); } else { throw destinationResolutionException; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicBindable.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicDestinationsBindable.java similarity index 85% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicBindable.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicDestinationsBindable.java index 53e4ad0cc..c03201b30 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicBindable.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicDestinationsBindable.java @@ -23,18 +23,20 @@ import java.util.Set; import org.springframework.cloud.stream.binder.Binding; /** - * A {@link BindableAdapter} that stores the dynamic destination names and handle their unbinding. + * 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 DynamicBindable extends BindableAdapter { +public final class DynamicDestinationsBindable extends BindableAdapter { /** * Map containing dynamic destination names and their bindings. */ private Map outputBindings = new HashMap<>(); - void addDynamicOutputs(String name, Binding binding) { + public void addOutputBinding(String name, Binding binding) { this.outputBindings.put(name, binding); } @@ -48,5 +50,6 @@ public final class DynamicBindable extends BindableAdapter { for (Map.Entry entry: outputBindings.entrySet()) { entry.getValue().unbind(); } + outputBindings.clear(); } } 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 a5b3fa014..1da49f774 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 @@ -39,7 +39,7 @@ 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.DynamicBindable; +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; @@ -154,8 +154,8 @@ public class ChannelBindingServiceConfiguration { } @Bean - public DynamicBindable dynamicBindable() { - return new DynamicBindable(); + public DynamicDestinationsBindable dynamicBindable() { + return new DynamicDestinationsBindable(); } // IMPORTANT: Nested class to avoid instantiating all of the above early diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java index 062ccaf47..f00bac5c0 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java @@ -43,7 +43,7 @@ import org.mockito.Mockito; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; -import org.springframework.cloud.stream.binding.DynamicBindable; +import org.springframework.cloud.stream.binding.DynamicDestinationsBindable; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.context.support.StaticApplicationContext; @@ -79,7 +79,7 @@ public class BinderAwareChannelResolverTests { return binder; } }; - this.resolver = new BinderAwareChannelResolver(binderFactory, null, new DynamicBindable()); + this.resolver = new BinderAwareChannelResolver(binderFactory, null, new DynamicDestinationsBindable()); this.resolver.setBeanFactory(context.getBeanFactory()); context.getBeanFactory().registerSingleton("channelResolver", this.resolver); @@ -128,7 +128,7 @@ public class BinderAwareChannelResolverTests { @SuppressWarnings("rawtypes") public void propertyPassthrough() { ChannelBindingServiceProperties bindingServiceProperties = new ChannelBindingServiceProperties(); - DynamicBindable dynamicBindable = new DynamicBindable(); + DynamicDestinationsBindable dynamicDestinationsBindable = new DynamicDestinationsBindable(); Map bindings = new HashMap(); BindingProperties bindingProperties = new BindingProperties(); bindingProperties.setContentType("text/plain"); @@ -148,7 +148,7 @@ public class BinderAwareChannelResolverTests { when(mockBinderFactory.getBinder("someTransport")).thenReturn(binder2); @SuppressWarnings("unchecked") BinderAwareChannelResolver resolver = - new BinderAwareChannelResolver(mockBinderFactory, bindingServiceProperties, dynamicBindable); + new BinderAwareChannelResolver(mockBinderFactory, bindingServiceProperties, dynamicDestinationsBindable); BeanFactory beanFactory = new DefaultListableBeanFactory(); resolver.setBeanFactory(beanFactory); MessageChannel resolved = resolver.resolveDestination("foo"); @@ -157,9 +157,9 @@ public class BinderAwareChannelResolverTests { resolved = resolver.resolveDestination("someTransport:bar"); verify(binder2).bindProducer(eq("bar"), any(MessageChannel.class), any(Properties.class)); assertSame(resolved, beanFactory.getBean("someTransport:bar")); - assertTrue("Dynamic bindable should have two destination names", dynamicBindable.getOutputs().size() == 2); - assertTrue("Dynamic bindable should have the destination name 'foo'", dynamicBindable.getOutputs().contains("foo")); - assertTrue("Dynamic bindable should have the destination name 'bar'", dynamicBindable.getOutputs().contains("someTransport:bar")); + assertTrue("Dynamic bindable should have two destination names", dynamicDestinationsBindable.getOutputs().size() == 2); + assertTrue("Dynamic bindable should have the destination name 'foo'", dynamicDestinationsBindable.getOutputs().contains("foo")); + assertTrue("Dynamic bindable should have the destination name 'bar'", dynamicDestinationsBindable.getOutputs().contains("someTransport:bar")); } /** 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 e303b9e9b..6da1fb1b2 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 @@ -182,7 +182,7 @@ public class ChannelBindingServiceTests { public void checkDynamicBinding () { ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties(); - DynamicBindable dynamicBindable = new DynamicBindable(); + DynamicDestinationsBindable dynamicDestinationsBindable = new DynamicDestinationsBindable(); DefaultBinderFactory binderFactory = new DefaultBinderFactory<>(Collections.singletonMap("mock", new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), @@ -196,7 +196,7 @@ public class ChannelBindingServiceTests { final AtomicReference dynamic = new AtomicReference<>(); when(binder.bindProducer( matches("bar"), any(DirectChannel.class), any(Properties.class))).thenReturn(mockBinding); - BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binderFactory, properties, dynamicBindable); + BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binderFactory, properties, dynamicDestinationsBindable); ConfigurableListableBeanFactory beanFactory = mock(ConfigurableListableBeanFactory.class); when(beanFactory.getBean("mock:bar", MessageChannel.class)) .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class));