diff --git a/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java b/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java index 921814416..ec0d38ff4 100644 --- a/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java +++ b/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java @@ -20,6 +20,8 @@ import java.util.HashMap; import java.util.Map; import java.util.Properties; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.LinkedBlockingDeque; import org.springframework.cloud.stream.binder.Binder; @@ -47,6 +49,7 @@ public class TestSupportBinder implements Binder { private final MessageCollectorImpl messageCollector = new MessageCollectorImpl(); + private final ConcurrentMap messageChannels = new ConcurrentHashMap<>(); @Override public Binding bindConsumer(String name, String group, MessageChannel inboundBindTarget, Properties properties) { @@ -65,19 +68,25 @@ public class TestSupportBinder implements Binder { queue.add(message); } }); + this.messageChannels.put(name, outboundBindTarget); return null; } @Override public void unbind(Binding binding) { - if (Binding.Type.producer.equals(binding.getType())) - messageCollector.unregister(binding.getTarget()); + if (Binding.Type.producer.equals(binding.getType())) { + messageCollector.unregister(binding.getTarget()); + } } public MessageCollector messageCollector() { return messageCollector; } + public MessageChannel getChannelForName(String name) { + return this.messageChannels.get(name); + } + /** * Maintains mappings between channels and queues. * 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 26bc24c15..c344c2f34 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 @@ -16,17 +16,19 @@ package org.springframework.cloud.stream.binding; +import java.util.Arrays; import java.util.Properties; import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.config.ConfigurableBeanFactory; -import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.integration.channel.DirectChannel; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver; import org.springframework.messaging.core.DestinationResolutionException; +import org.springframework.util.Assert; /** * A {@link org.springframework.messaging.core.DestinationResolver} implementation that @@ -40,36 +42,41 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina private final BinderFactory binderFactory; - private final Properties producerProperties; + private final ChannelBindingServiceProperties channelBindingServiceProperties; - private DefaultListableBeanFactory beanFactory; + private ConfigurableListableBeanFactory beanFactory; - public BinderAwareChannelResolver(BinderFactory binderFactory, Properties producerProperties) { + public BinderAwareChannelResolver(BinderFactory binderFactory, + ChannelBindingServiceProperties channelBindingServiceProperties) { + Assert.notNull(binderFactory, "'binderFactory' cannot be null"); this.binderFactory = binderFactory; - this.producerProperties = producerProperties; + this.channelBindingServiceProperties = channelBindingServiceProperties; } @Override public void setBeanFactory(BeanFactory beanFactory) { super.setBeanFactory(beanFactory); - if (beanFactory instanceof ConfigurableBeanFactory) { - this.beanFactory = (DefaultListableBeanFactory) beanFactory; + if (beanFactory instanceof ConfigurableListableBeanFactory) { + this.beanFactory = (ConfigurableListableBeanFactory) beanFactory; } } @Override public MessageChannel resolveDestination(String name) { MessageChannel channel = null; + DestinationResolutionException destinationResolutionException; try { return super.resolveDestination(name); } catch (DestinationResolutionException e) { + destinationResolutionException = e; } synchronized (this) { try { return super.resolveDestination(name); } catch (DestinationResolutionException e) { + destinationResolutionException = e; } if (this.beanFactory != null && this.binderFactory != null) { channel = new DirectChannel(); @@ -86,8 +93,23 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina " [:]"); } } - Binder binder = binderFactory.getBinder(transport); - binder.bindProducer(name, channel, this.producerProperties); + 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); + } + boolean dynamicAllowed = dynamicDestinations == null + || dynamicDestinations.length == 0 + || Arrays.asList(dynamicDestinations).contains(name); + if (dynamicAllowed) { + Binder binder = binderFactory.getBinder(transport); + binder.bindProducer(name, channel, producerProperties); + } + else { + throw destinationResolutionException; + } } return channel; } 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 85f664ab2..e1d821030 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-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. @@ -20,7 +20,6 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Properties; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactoryUtils; @@ -31,7 +30,6 @@ import org.springframework.boot.context.properties.ConfigurationPropertiesBindin 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.MessageChannelConfigurer; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor; import org.springframework.cloud.stream.binding.ChannelBindingService; @@ -39,6 +37,7 @@ import org.springframework.cloud.stream.binding.CompositeMessageChannelConfigure import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener; import org.springframework.cloud.stream.binding.DefaultBindableChannelFactory; import org.springframework.cloud.stream.binding.InputBindingLifecycle; +import org.springframework.cloud.stream.binding.MessageChannelConfigurer; import org.springframework.cloud.stream.binding.MessageConverterConfigurer; import org.springframework.cloud.stream.binding.MessageHistoryTrackerConfigurer; import org.springframework.cloud.stream.binding.OutputBindingLifecycle; @@ -62,6 +61,7 @@ import org.springframework.messaging.core.DestinationResolver; * @author David Turanski * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Gary Russell */ @Configuration @EnableConfigurationProperties(ChannelBindingServiceProperties.class) @@ -124,8 +124,9 @@ public class ChannelBindingServiceConfiguration { @Bean public BinderAwareChannelResolver binderAwareChannelResolver( - BinderFactory binderFactory) { - return new BinderAwareChannelResolver(binderFactory, new Properties()); + BinderFactory binderFactory, + ChannelBindingServiceProperties channelBindingServiceProperties) { + return new BinderAwareChannelResolver(binderFactory, channelBindingServiceProperties); } @Bean diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java index f529dfd32..1e95a504b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java @@ -50,6 +50,8 @@ public class ChannelBindingServiceProperties { private String defaultBinder; + private String[] dynamicDestinations = new String[0]; + public Map getBindings() { return bindings; } @@ -90,6 +92,14 @@ public class ChannelBindingServiceProperties { this.instanceCount = instanceCount; } + public String[] getDynamicDestinations() { + return dynamicDestinations; + } + + public void setDynamicDestinations(String[] dynamicDestinations) { + this.dynamicDestinations = dynamicDestinations; + } + public String getBindingDestination(String channelName) { BindingProperties bindingProperties = bindings.get(channelName); // we may shortcut directly to the path 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 d261fb46b..0d4d401f6 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 @@ -25,9 +25,12 @@ import static org.mockito.Matchers.anyString; import static org.mockito.Matchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -40,6 +43,8 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.cloud.stream.binder.local.LocalMessageChannelBinder; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.context.support.StaticApplicationContext; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.scheduling.PollerMetadata; @@ -128,21 +133,26 @@ public class BinderAwareChannelResolverTests { @Test @SuppressWarnings("rawtypes") public void propertyPassthrough() { - Properties properties = new Properties(); + ChannelBindingServiceProperties bindingServiceProperties = new ChannelBindingServiceProperties(); + Map bindings = new HashMap(); + BindingProperties bindingProperties = new BindingProperties(); + bindingProperties.setContentType("text/plain"); + bindings.put("foo", bindingProperties); + bindingServiceProperties.setBindings(bindings); @SuppressWarnings("unchecked") - Binder binderFactory = mock(Binder.class); + Binder binder = mock(Binder.class); BinderFactory mockBinderFactory = Mockito.mock(BinderFactory.class); - Mockito.when(mockBinderFactory.getBinder(anyString())).thenReturn(binderFactory); + when(mockBinderFactory.getBinder(anyString())).thenReturn(binder); @SuppressWarnings("unchecked") BinderAwareChannelResolver resolver = - new BinderAwareChannelResolver(mockBinderFactory, properties); + new BinderAwareChannelResolver(mockBinderFactory, bindingServiceProperties); BeanFactory beanFactory = new DefaultListableBeanFactory(); resolver.setBeanFactory(beanFactory); MessageChannel resolved = resolver.resolveDestination("foo"); - verify(binderFactory).bindProducer(eq("foo"), any(MessageChannel.class), eq(properties)); + verify(binder).bindProducer(eq("foo"), any(MessageChannel.class), any(Properties.class)); assertSame(resolved, beanFactory.getBean("foo")); resolved = resolver.resolveDestination("someTransport:foo"); - verify(binderFactory).bindProducer(eq("someTransport:foo"), any(MessageChannel.class), eq(properties)); + verify(binder).bindProducer(eq("someTransport:foo"), any(MessageChannel.class), any(Properties.class)); assertSame(resolved, beanFactory.getBean("someTransport:foo")); } 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 41f761af8..b196d931e 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 @@ -16,18 +16,33 @@ package org.springframework.cloud.stream.binding; + import static org.hamcrest.CoreMatchers.sameInstance; +import static org.hamcrest.Matchers.containsString; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.fail; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; +import static org.mockito.Matchers.matches; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Properties; +import java.util.concurrent.atomic.AtomicReference; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; +import org.springframework.beans.factory.NoSuchBeanDefinitionException; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderConfiguration; import org.springframework.cloud.stream.binder.BinderType; @@ -39,6 +54,7 @@ import org.springframework.cloud.stream.utils.MockBinderConfiguration; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.core.DestinationResolutionException; /** * @author Gary Russell @@ -94,10 +110,51 @@ public class ChannelBindingServiceTests { MessageChannel inputChannel = new DirectChannel(); Binding mockBinding = Binding.forConsumer("foo", "fooGroup", Mockito.mock(AbstractEndpoint.class), inputChannel, null); - Mockito.when(binder.bindConsumer("foo", "fooGroup", inputChannel, new Properties())) + when(binder.bindConsumer("foo", "fooGroup", inputChannel, new Properties())) .thenReturn(mockBinding); Binding binding = service.bindConsumer(inputChannel, name); - Assert.assertThat(binding, sameInstance(mockBinding)); + assertThat(binding, sameInstance(mockBinding)); + + 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); + ConfigurableListableBeanFactory beanFactory = mock(ConfigurableListableBeanFactory.class); + when(beanFactory.getBean("mock:bar", MessageChannel.class)) + .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class)); + doAnswer(new Answer(){ + + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + dynamic.set(invocation.getArgumentAt(1, MessageChannel.class)); + return null; + } + + }).when(beanFactory).registerSingleton(eq("mock:bar"), any(MessageChannel.class)); + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + return dynamic.get(); + } + + }).when(beanFactory).initializeBean(any(MessageChannel.class), eq("mock: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)); + properties.setDynamicDestinations(new String[] { "mock:bar" }); + resolved = resolver.resolveDestination("mock:bar"); + assertThat(resolved, sameInstance(dynamic.get())); + properties.setDynamicDestinations(new String[] { "foo:bar" }); + try { + resolved = resolver.resolveDestination("mock:bar"); + fail(); + } + catch (DestinationResolutionException e) { + assertThat(e.getMessage(), containsString("Failed to find MessageChannel bean with name 'mock:bar'")); + } + service.unbindConsumers(name); verify(binder).bindConsumer(name, props.getGroup(), inputChannel, properties.getConsumerProperties(name)); verify(binder).unbind(binding);