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 bde9a86a6..ad73c4584 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 @@ -35,8 +35,10 @@ import org.springframework.util.ObjectUtils; * resolves the channel from the bean factory and, if not present, creates a new channel * and adds it to the factory after binding it to the binder. The binder is optionally * determined with a prefix preceding a colon. + * * @author Mark Fisher * @author Gary Russell + * @author Ilayaperumal Gopinathan */ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestinationResolver { @@ -44,13 +46,16 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina private final ChannelBindingServiceProperties channelBindingServiceProperties; + private final DynamicBindable dynamicBindable; + private ConfigurableListableBeanFactory beanFactory; - public BinderAwareChannelResolver(BinderFactory binderFactory, - ChannelBindingServiceProperties channelBindingServiceProperties) { + public BinderAwareChannelResolver(BinderFactory binderFactory, + ChannelBindingServiceProperties channelBindingServiceProperties, DynamicBindable dynamicBindable) { Assert.notNull(binderFactory, "'binderFactory' cannot be null"); this.binderFactory = binderFactory; this.channelBindingServiceProperties = channelBindingServiceProperties; + this.dynamicBindable = dynamicBindable; } @Override @@ -62,49 +67,45 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina } @Override - public MessageChannel resolveDestination(String name) { + public MessageChannel resolveDestination(String destinationName) { MessageChannel channel = null; DestinationResolutionException destinationResolutionException; try { - return super.resolveDestination(name); + return super.resolveDestination(destinationName); } catch (DestinationResolutionException e) { destinationResolutionException = e; } synchronized (this) { - try { - return super.resolveDestination(name); - } - catch (DestinationResolutionException e) { - destinationResolutionException = e; - } if (this.beanFactory != null && this.binderFactory != null) { String[] dynamicDestinations = null; Properties producerProperties = null; if (this.channelBindingServiceProperties != null) { dynamicDestinations = this.channelBindingServiceProperties.getDynamicDestinations(); // TODO: need the props to return some defaults if not found - producerProperties = this.channelBindingServiceProperties.getProducerProperties(name); + producerProperties = this.channelBindingServiceProperties.getProducerProperties(destinationName); } boolean dynamicAllowed = ObjectUtils.isEmpty(dynamicDestinations) - || ObjectUtils.containsElement(dynamicDestinations, name); + || ObjectUtils.containsElement(dynamicDestinations, destinationName); if (dynamicAllowed) { - channel = new DirectChannel(); - this.beanFactory.registerSingleton(name, channel); - channel = (MessageChannel) this.beanFactory.initializeBean(channel, name); String transport = null; - if (name.contains(":")) { - String[] tokens = name.split(":", 2); + String beanName = destinationName; + if (destinationName.contains(":")) { + String[] tokens = destinationName.split(":", 2); if (tokens.length == 2) { transport = tokens[0]; + destinationName = tokens[1]; } else if (tokens.length != 1) { - throw new IllegalArgumentException("Unrecognized channel naming scheme: " + name + " , should be" + - " [:]"); + throw new IllegalArgumentException("Unrecognized channel naming scheme: " + destinationName + " , should be" + + " [:]"); } } + channel = new DirectChannel(); + this.beanFactory.registerSingleton(beanName, channel); + channel = (MessageChannel) this.beanFactory.initializeBean(channel, beanName); Binder binder = binderFactory.getBinder(transport); - binder.bindProducer(name, channel, producerProperties); + this.dynamicBindable.addDynamicOutputs(beanName, binder.bindProducer(destinationName, channel, producerProperties)); } else { throw destinationResolutionException; @@ -113,5 +114,4 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina return channel; } } - } 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/DynamicBindable.java new file mode 100644 index 000000000..53e4ad0cc --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DynamicBindable.java @@ -0,0 +1,52 @@ +/* + * 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 handle their unbinding. + * + * @author Ilayaperumal Gopinathan + */ +public final class DynamicBindable extends BindableAdapter { + + /** + * Map containing dynamic destination names and their bindings. + */ + private Map outputBindings = new HashMap<>(); + + void addDynamicOutputs(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(); + } + } +} 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 ea6131fab..a5b3fa014 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,6 +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.InputBindingLifecycle; import org.springframework.cloud.stream.binding.MessageChannelConfigurer; import org.springframework.cloud.stream.binding.MessageConverterConfigurer; @@ -134,10 +135,9 @@ public class ChannelBindingServiceConfiguration { } @Bean - public BinderAwareChannelResolver binderAwareChannelResolver( - BinderFactory binderFactory, + public BinderAwareChannelResolver binderAwareChannelResolver(BinderFactory binderFactory, ChannelBindingServiceProperties channelBindingServiceProperties) { - return new BinderAwareChannelResolver(binderFactory, channelBindingServiceProperties); + return new BinderAwareChannelResolver(binderFactory, channelBindingServiceProperties, dynamicBindable()); } @Bean @@ -153,6 +153,11 @@ public class ChannelBindingServiceConfiguration { return new SingleChannelBindable(ERROR_CHANNEL_NAME, errorChannel); } + @Bean + public DynamicBindable dynamicBindable() { + return new DynamicBindable(); + } + // IMPORTANT: Nested class to avoid instantiating all of the above early @Configuration protected static class PostProcessorConfiguration { 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 618d7fd1a..062ccaf47 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 @@ -21,8 +21,8 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Matchers.any; -import static org.mockito.Matchers.anyString; import static org.mockito.Matchers.eq; +import static org.mockito.Matchers.matches; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -43,6 +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.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.context.support.StaticApplicationContext; @@ -59,6 +60,7 @@ import org.springframework.messaging.SubscribableChannel; /** * @author Mark Fisher * @author Gary Russell + * @author Ilayaperumal Gopinathan */ public class BinderAwareChannelResolverTests { @@ -71,12 +73,13 @@ public class BinderAwareChannelResolverTests { @Before public void setupContext() throws Exception { this.binder = new TestBinder(); - this.resolver = new BinderAwareChannelResolver(new BinderFactory() { + BinderFactory binderFactory = new BinderFactory() { @Override public Binder getBinder(String configurationName) { return binder; } - }, null); + }; + this.resolver = new BinderAwareChannelResolver(binderFactory, null, new DynamicBindable()); this.resolver.setBeanFactory(context.getBeanFactory()); context.getBeanFactory().registerSingleton("channelResolver", this.resolver); @@ -125,6 +128,7 @@ public class BinderAwareChannelResolverTests { @SuppressWarnings("rawtypes") public void propertyPassthrough() { ChannelBindingServiceProperties bindingServiceProperties = new ChannelBindingServiceProperties(); + DynamicBindable dynamicBindable = new DynamicBindable(); Map bindings = new HashMap(); BindingProperties bindingProperties = new BindingProperties(); bindingProperties.setContentType("text/plain"); @@ -132,19 +136,30 @@ public class BinderAwareChannelResolverTests { bindingServiceProperties.setBindings(bindings); @SuppressWarnings("unchecked") Binder binder = mock(Binder.class); + Binder binder2 = mock(Binder.class); BinderFactory mockBinderFactory = Mockito.mock(BinderFactory.class); - when(mockBinderFactory.getBinder(anyString())).thenReturn(binder); + Binding fooBinding = Mockito.mock(Binding.class); + Binding barBinding = Mockito.mock(Binding.class); + when(binder.bindProducer( + matches("foo"), any(DirectChannel.class), any(Properties.class))).thenReturn(fooBinding); + when(binder2.bindProducer( + matches("bar"), any(DirectChannel.class), any(Properties.class))).thenReturn(barBinding); + when(mockBinderFactory.getBinder(null)).thenReturn(binder); + when(mockBinderFactory.getBinder("someTransport")).thenReturn(binder2); @SuppressWarnings("unchecked") BinderAwareChannelResolver resolver = - new BinderAwareChannelResolver(mockBinderFactory, bindingServiceProperties); + new BinderAwareChannelResolver(mockBinderFactory, bindingServiceProperties, dynamicBindable); BeanFactory beanFactory = new DefaultListableBeanFactory(); resolver.setBeanFactory(beanFactory); MessageChannel resolved = resolver.resolveDestination("foo"); verify(binder).bindProducer(eq("foo"), any(MessageChannel.class), any(Properties.class)); assertSame(resolved, beanFactory.getBean("foo")); - resolved = resolver.resolveDestination("someTransport:foo"); - verify(binder).bindProducer(eq("someTransport:foo"), any(MessageChannel.class), any(Properties.class)); - assertSame(resolved, beanFactory.getBean("someTransport:foo")); + 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")); } /** @@ -156,7 +171,7 @@ public class BinderAwareChannelResolverTests { @Override public Binding bindConsumer(String name, String group, MessageChannel inboundBindTarget, - Properties properties) { + Properties properties) { synchronized (destinations) { if (!destinations.containsKey(name)) { destinations.put(name, new DirectChannel()); 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 1c927fc38..e303b9e9b 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 @@ -61,6 +61,7 @@ import org.springframework.messaging.core.DestinationResolutionException; * @author Gary Russell * @author Mark Fisher * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ public class ChannelBindingServiceTests { @@ -181,6 +182,7 @@ public class ChannelBindingServiceTests { public void checkDynamicBinding () { ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties(); + DynamicBindable dynamicBindable = new DynamicBindable(); DefaultBinderFactory binderFactory = new DefaultBinderFactory<>(Collections.singletonMap("mock", new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), @@ -193,8 +195,8 @@ public class ChannelBindingServiceTests { @SuppressWarnings("unchecked") final AtomicReference dynamic = new AtomicReference<>(); when(binder.bindProducer( - matches("mock:bar"), any(DirectChannel.class), any(Properties.class))).thenReturn(mockBinding); - BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binderFactory, properties); + matches("bar"), any(DirectChannel.class), any(Properties.class))).thenReturn(mockBinding); + BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binderFactory, properties, dynamicBindable); ConfigurableListableBeanFactory beanFactory = mock(ConfigurableListableBeanFactory.class); when(beanFactory.getBean("mock:bar", MessageChannel.class)) .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class)); @@ -206,7 +208,7 @@ public class ChannelBindingServiceTests { return null; } - }).when(beanFactory).registerSingleton(eq("mock:bar"), any(MessageChannel.class)); + }).when(beanFactory).registerSingleton(eq("bar"), any(MessageChannel.class)); doAnswer(new Answer() { @Override @@ -214,11 +216,11 @@ public class ChannelBindingServiceTests { return dynamic.get(); } - }).when(beanFactory).initializeBean(any(MessageChannel.class), eq("mock:bar")); + }).when(beanFactory).initializeBean(any(MessageChannel.class), eq("bar")); resolver.setBeanFactory(beanFactory); MessageChannel resolved = resolver.resolveDestination("mock:bar"); assertThat(resolved, sameInstance(dynamic.get())); - verify(binder).bindProducer(eq("mock:bar"), eq(dynamic.get()), any(Properties.class)); + verify(binder).bindProducer(eq("bar"), eq(dynamic.get()), any(Properties.class)); properties.setDynamicDestinations(new String[] { "mock:bar" }); resolved = resolver.resolveDestination("mock:bar"); assertThat(resolved, sameInstance(dynamic.get()));