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 5f556c783..a03080b8c 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 @@ -22,7 +22,6 @@ import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.ProducerProperties; 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; @@ -46,16 +45,22 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina private final DynamicDestinationsBindable dynamicDestinationsBindable; + private final BindableChannelFactory bindableChannelFactory; + private ConfigurableListableBeanFactory beanFactory; - public BinderAwareChannelResolver(BinderFactory binderFactory, ChannelBindingServiceProperties channelBindingServiceProperties, - DynamicDestinationsBindable dynamicDestinationsBindable) { + public BinderAwareChannelResolver(BinderFactory binderFactory, + ChannelBindingServiceProperties channelBindingServiceProperties, + DynamicDestinationsBindable dynamicDestinationsBindable, + BindableChannelFactory bindableChannelFactory) { Assert.notNull(binderFactory, "'binderFactory' cannot be null"); Assert.notNull(channelBindingServiceProperties, "'channelBindingServiceProperties' cannot be null"); Assert.notNull(dynamicDestinationsBindable, "'dynamicDestinationBindable' cannot be null"); + Assert.notNull(bindableChannelFactory, "'bindableChannelFactory' cannot be null"); this.binderFactory = binderFactory; this.channelBindingServiceProperties = channelBindingServiceProperties; this.dynamicDestinationsBindable = dynamicDestinationsBindable; + this.bindableChannelFactory = bindableChannelFactory; } @Override @@ -98,7 +103,7 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina " [:]"); } } - channel = new DirectChannel(); + channel = this.bindableChannelFactory.createSubscribableChannel(channelName); this.beanFactory.registerSingleton(beanName, channel); channel = (MessageChannel) this.beanFactory.initializeBean(channel, beanName); @SuppressWarnings("unchecked") 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 7d4b5912b..65fa4983d 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 @@ -142,8 +142,10 @@ public class ChannelBindingServiceConfiguration { @Bean public BinderAwareChannelResolver binderAwareChannelResolver(BinderFactory binderFactory, - ChannelBindingServiceProperties channelBindingServiceProperties) { - return new BinderAwareChannelResolver(binderFactory, channelBindingServiceProperties, dynamicBindable()); + ChannelBindingServiceProperties channelBindingServiceProperties, + BindableChannelFactory bindableChannelFactory) { + return new BinderAwareChannelResolver(binderFactory, channelBindingServiceProperties, dynamicBindable(), + bindableChannelFactory); } 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 9f580c4c0..3e81f02f9 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 @@ -39,10 +39,15 @@ import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; +import org.springframework.beans.DirectFieldAccessor; 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.DefaultBindableChannelFactory; import org.springframework.cloud.stream.binding.DynamicDestinationsBindable; +import org.springframework.cloud.stream.binding.MessageConverterConfigurer; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.context.support.StaticApplicationContext; @@ -55,6 +60,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.SubscribableChannel; +import org.springframework.util.Assert; /** * @author Mark Fisher @@ -69,6 +75,10 @@ public class BinderAwareChannelResolverTests { private volatile Binder binder; + private volatile BindableChannelFactory bindableChannelFactory; + + private volatile ChannelBindingServiceProperties channelBindingServiceProperties; + @Before public void setupContext() throws Exception { this.binder = new TestBinder(); @@ -79,7 +89,19 @@ public class BinderAwareChannelResolverTests { return binder; } }; - this.resolver = new BinderAwareChannelResolver(binderFactory, new ChannelBindingServiceProperties(), new DynamicDestinationsBindable()); + this.channelBindingServiceProperties = new ChannelBindingServiceProperties(); + Map bindings = new HashMap(); + BindingProperties bindingProperties = new BindingProperties(); + bindingProperties.setContentType("text/plain"); + bindings.put("foo", bindingProperties); + this.channelBindingServiceProperties.setBindings(bindings); + MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer( + this.channelBindingServiceProperties, null, new DefaultMessageBuilderFactory()); + messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class)); + messageConverterConfigurer.afterPropertiesSet(); + this.bindableChannelFactory = new DefaultBindableChannelFactory(messageConverterConfigurer); + this.resolver = new BinderAwareChannelResolver(binderFactory, this.channelBindingServiceProperties, + new DynamicDestinationsBindable(), bindableChannelFactory); this.resolver.setBeanFactory(context.getBeanFactory()); context.getBeanFactory().registerSingleton("channelResolver", this.resolver); @@ -127,12 +149,12 @@ public class BinderAwareChannelResolverTests { @Test @SuppressWarnings({"rawtypes", "unchecked"}) public void propertyPassthrough() { - ChannelBindingServiceProperties bindingServiceProperties = new ChannelBindingServiceProperties(); DynamicDestinationsBindable dynamicDestinationsBindable = new DynamicDestinationsBindable(); Map bindings = new HashMap<>(); BindingProperties genericProperties = new BindingProperties(); + genericProperties.setContentType("text/plain"); bindings.put("foo", genericProperties); - bindingServiceProperties.setBindings(bindings); + this.channelBindingServiceProperties.setBindings(bindings); @SuppressWarnings("unchecked") Binder binder = mock(Binder.class); Binder binder2 = mock(Binder.class); @@ -147,13 +169,18 @@ public class BinderAwareChannelResolverTests { when(mockBinderFactory.getBinder("someTransport")).thenReturn(binder2); @SuppressWarnings("unchecked") BinderAwareChannelResolver resolver = - new BinderAwareChannelResolver(mockBinderFactory, bindingServiceProperties, dynamicDestinationsBindable); + new BinderAwareChannelResolver(mockBinderFactory, this.channelBindingServiceProperties, dynamicDestinationsBindable, + this.bindableChannelFactory); BeanFactory beanFactory = new DefaultListableBeanFactory(); resolver.setBeanFactory(beanFactory); - MessageChannel resolved = resolver.resolveDestination("foo"); + SubscribableChannel resolved = (SubscribableChannel) resolver.resolveDestination("foo"); + DirectFieldAccessor accessor = new DirectFieldAccessor(resolved); + Class[] dataTypes = (Class[]) accessor.getPropertyValue("datatypes"); + Assert.isTrue(dataTypes.length == 1, "Data type must be set for the Foo Channel"); + Assert.isTrue(dataTypes[0].equals(String.class), "Data type should be of type String"); verify(binder).bindProducer(eq("foo"), any(MessageChannel.class), any(ProducerProperties.class)); assertSame(resolved, beanFactory.getBean("foo")); - resolved = resolver.resolveDestination("someTransport:bar"); + resolved = (SubscribableChannel) resolver.resolveDestination("someTransport:bar"); verify(binder2).bindProducer(eq("bar"), any(MessageChannel.class), any(ProducerProperties.class)); assertSame(resolved, beanFactory.getBean("someTransport:bar")); assertTrue("Dynamic bindable should have two destination names", dynamicDestinationsBindable.getOutputs().size() == 2); 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 1c8d1baed..506c8784e 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 @@ -58,6 +58,7 @@ import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.cloud.stream.utils.MockBinderConfiguration; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.DestinationResolutionException; @@ -198,8 +199,10 @@ public class ChannelBindingServiceTests { Binding mockBinding = Mockito.mock(Binding.class); @SuppressWarnings("unchecked") final AtomicReference dynamic = new AtomicReference<>(); - when(binder.bindProducer(matches("bar"), any(DirectChannel.class), any(ProducerProperties.class))).thenReturn(mockBinding); - BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binderFactory, properties, dynamicDestinationsBindable); + when(binder.bindProducer( + matches("bar"), any(DirectChannel.class), any(ProducerProperties.class))).thenReturn(mockBinding); + BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binderFactory, properties, dynamicDestinationsBindable, + new DefaultBindableChannelFactory(new MessageConverterConfigurer(properties, null, new DefaultMessageBuilderFactory()))); ConfigurableListableBeanFactory beanFactory = mock(ConfigurableListableBeanFactory.class); when(beanFactory.getBean("mock:bar", MessageChannel.class)) .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class));