From 3afa3d8bd39988ac23af91963a2ca0c7962bae73 Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Fri, 4 Dec 2015 22:40:29 +0530 Subject: [PATCH] Remove configuring shared message channel Simplify ChannelFactory interface - Use separate methods for each message channel type (Subscribable/Pollable) --- .../aggregate/AggregateApplication.java | 8 +- ...ctory.java => BindableChannelFactory.java} | 33 ++++---- .../stream/binding/BindableProxyFactory.java | 83 ++++++++----------- .../cloud/stream/binding/ChannelFactory.java | 22 ++++- .../ChannelBindingServiceConfiguration.java | 6 +- .../aggregation/ModuleAggregationTest.java | 4 + 6 files changed, 86 insertions(+), 70 deletions(-) rename spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/{DefaultChannelFactory.java => BindableChannelFactory.java} (51%) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java index 2cc0c792e..198e15784 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java @@ -27,9 +27,13 @@ import org.springframework.cloud.stream.messaging.Source; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.integration.channel.DirectChannel; +import org.springframework.messaging.SubscribableChannel; /** + * Class that is responsible for embedding modules using shared channel registry. + * * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ public class AggregateApplication { @@ -84,7 +88,7 @@ public class AggregateApplication { .web(false) .headless(true) .properties("spring.jmx.default-domain=" - + AggregatorParentConfiguration.class.getName()); + + AggregatorParentConfiguration.class.getName()); return aggregatorParentConfiguration.run(args); } @@ -113,7 +117,7 @@ public class AggregateApplication { } private static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry, Class[] modules) { - DirectChannel sharedChannel = null; + SubscribableChannel sharedChannel = null; for (int i = 0; i < modules.length; i++) { Class module = modules[i]; String moduleClassName = module.getName(); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultChannelFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableChannelFactory.java similarity index 51% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultChannelFactory.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableChannelFactory.java index 0c75be947..b8e9f147c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultChannelFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableChannelFactory.java @@ -15,36 +15,37 @@ */ package org.springframework.cloud.stream.binding; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; /** - * Class that {@link BindableProxyFactory} uses to create message channels. + * Class that {@link BindableProxyFactory} uses to create and configure message channels. * * @author Marius Bogoevici * @author David Syer * @author Ilayaperumal Gopinathan */ -public class DefaultChannelFactory implements ChannelFactory { +public class BindableChannelFactory implements ChannelFactory { - @Autowired - MessageConverterConfigurer messageConverterConfigurer; + private final MessageConverterConfigurer messageConverterConfigurer; + + public BindableChannelFactory(MessageConverterConfigurer messageConverterConfigurer) { + this.messageConverterConfigurer = messageConverterConfigurer; + } @Override - public MessageChannel createChannel(String name, Class inputChannelType) throws Exception { - MessageChannel messageChannel = createMessageChannel(inputChannelType); - messageConverterConfigurer.configureMessageConverters(messageChannel, name); - return messageChannel; + public PollableChannel createPollableChannel(String name) { + PollableChannel pollableChannel = new QueueChannel(); + messageConverterConfigurer.configureMessageConverters(pollableChannel, name); + return pollableChannel; } - private MessageChannel createMessageChannel(Class messageChannelType) { - return isPollable(messageChannelType) ? new QueueChannel() : new DirectChannel(); - } - - private boolean isPollable(Class channelType) { - return PollableChannel.class.equals(channelType); + @Override + public SubscribableChannel createSubscribableChannel(String name) { + SubscribableChannel subscribableChannel = new DirectChannel(); + messageConverterConfigurer.configureMessageConverters(subscribableChannel, name); + return subscribableChannel; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java index efa2b2755..9516c4c3e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java @@ -41,6 +41,8 @@ import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.Output; import org.springframework.cloud.stream.binder.MessageChannelBinderSupport; import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.MessageChannel; @@ -79,9 +81,6 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean channelType = (Class) method.getReturnType(); + MessageChannel sharedChannel = locateSharedChannel(name); + if (sharedChannel == null) { + inputHolders.put(name, new ChannelHolder(createBindableChannel(name, channelType), true)); + } + else { + if (!channelType.isAssignableFrom(sharedChannel.getClass())) { + bridgeSharedChannel(channelType, sharedChannel); } } } - catch (Exception e) { - throw new RuntimeException(e); - } } }); ReflectionUtils.doWithMethods(type, new ReflectionUtils.MethodCallback() { @Override public void doWith(Method method) throws IllegalArgumentException { - try { - Output output = AnnotationUtils.findAnnotation(method, Output.class); - if (output != null) { - String name = BindingBeanDefinitionRegistryUtils.getChannelName(output, method); - MessageChannel sharedChannel = locateSharedChannel(name); - if (sharedChannel == null) { - outputHolders.put(name, new ChannelHolder( - channelFactory.createChannel(name, method.getReturnType()), true)); - } - else { - configureSharedMessageChannel(name, method.getReturnType(), sharedChannel); + Output output = AnnotationUtils.findAnnotation(method, Output.class); + if (output != null) { + String name = BindingBeanDefinitionRegistryUtils.getChannelName(output, method); + Class channelType = (Class) method.getReturnType(); + MessageChannel sharedChannel = locateSharedChannel(name); + if (sharedChannel == null) { + outputHolders.put(name, new ChannelHolder(createBindableChannel(name, channelType), true)); + } + else { + if (!channelType.isAssignableFrom(sharedChannel.getClass())) { + bridgeSharedChannel(channelType, sharedChannel); } } } - catch (Exception e) { - throw new RuntimeException(e); - } } }); } + private MessageChannel createBindableChannel(String name, Class channelType) { + return isPollable(channelType) ? this.channelFactory.createPollableChannel(name) : + this.channelFactory.createSubscribableChannel(name); + } + private MessageChannel locateSharedChannel(String name) { return this.sharedChannelRegistry != null ? this.sharedChannelRegistry.get(getNamespacePrefixedChannelName(name)) : null; @@ -176,22 +174,14 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean channelType, MessageChannel sharedChannel) - throws Exception { - if (channelType.isAssignableFrom(sharedChannel.getClass())) { - messageConverterConfigurer.configureMessageConverters(sharedChannel, name); + private void bridgeSharedChannel(Class channelType, MessageChannel sharedChannel) { + // handle the special case where the shared channel is of a different nature + // (i.e. pollable vs subscribable) than the target channel + if (isPollable(sharedChannel.getClass())) { + bridgePollableToSubscribableChannel(sharedChannel, new DirectChannel()); } else { - // handle the special case where the shared channel is of a different nature - // (i.e. pollable vs subscribable) than the target channel - final MessageChannel inputChannel = this.channelFactory.createChannel(name, channelType); - if (isPollable(sharedChannel.getClass())) { - bridgePollableToSubscribableChannel(sharedChannel, inputChannel); - } - else { - bridgeSubscribableToPollableChannel((SubscribableChannel) sharedChannel, inputChannel); - } - messageConverterConfigurer.configureMessageConverters(inputChannel, name); + bridgeSubscribableToPollableChannel((SubscribableChannel) sharedChannel, new QueueChannel()); } } @@ -200,8 +190,7 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean channelType) throws Exception; + /** + * Create a {@link SubscribableChannel} that will be bound via the message channel + * {@link org.springframework.cloud.stream.binder.Binder}. + * + * @param name name of the message channel + * @return Subscribable message channel + */ + SubscribableChannel createSubscribableChannel(String name); + + /** + * Create a {@link PollableChannel} that will be bound via the message channel + * {@link org.springframework.cloud.stream.binder.Binder}. + * + * @param name name of the message channel + * @return Pollable message channel + */ + PollableChannel createPollableChannel(String name); + } 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 f26d9ed1c..54e5219a5 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 @@ -33,7 +33,7 @@ import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcess import org.springframework.cloud.stream.binding.ChannelBindingService; import org.springframework.cloud.stream.binding.ChannelFactory; import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener; -import org.springframework.cloud.stream.binding.DefaultChannelFactory; +import org.springframework.cloud.stream.binding.BindableChannelFactory; import org.springframework.cloud.stream.binding.InputBindingLifecycle; import org.springframework.cloud.stream.binding.MessageConverterConfigurer; import org.springframework.cloud.stream.binding.OutputBindingLifecycle; @@ -79,8 +79,8 @@ public class ChannelBindingServiceConfiguration { } @Bean - public ChannelFactory channelFactory() { - return new DefaultChannelFactory(); + public ChannelFactory channelFactory(ChannelBindingServiceProperties channelBindingServiceProperties) { + return new BindableChannelFactory(messageConverterConfigurer(channelBindingServiceProperties)); } @Bean diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java index cd1161ec5..b14e33afa 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.aggregation; import static org.hamcrest.collection.IsCollectionWithSize.hasSize; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import org.junit.Ignore; @@ -26,6 +27,7 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.cloud.stream.aggregate.AggregateApplication; import org.springframework.cloud.stream.aggregate.SharedChannelRegistry; import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binding.ChannelFactory; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.cloud.stream.messaging.Source; import org.springframework.context.ConfigurableApplicationContext; @@ -42,6 +44,8 @@ public class ModuleAggregationTest { ConfigurableApplicationContext aggregatedApplicationContext = AggregateApplication.run(TestSource.class, TestProcessor.class); SharedChannelRegistry sharedChannelRegistry = aggregatedApplicationContext.getBean(SharedChannelRegistry.class); + ChannelFactory channelFactory = aggregatedApplicationContext.getBean(ChannelFactory.class); + assertNotNull(channelFactory); assertThat(sharedChannelRegistry.getAll().keySet(), hasSize(2)); }