SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove configuring shared message channel

Browse Source 
Simplify ChannelFactory interface

 - Use separate methods for each message channel type (Subscribable/Pollable)
This commit is contained in:
Ilayaperumal Gopinathan
2015-12-04 22:40:29 +05:30
committed by Mark Fisher
parent 13247b191a
commit 3afa3d8bd3
6 changed files with 86 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java
View File

@@ -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();

33
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultChannelFactory.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableChannelFactory.java
View File

@@ -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;
}
}

83
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java
View File

@@ -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<Obje
@Autowired
private ChannelFactory channelFactory;
@Autowired
private MessageConverterConfigurer messageConverterConfigurer;
@Autowired(required = false)
private SharedChannelRegistry sharedChannelRegistry;
@@ -123,50 +122,49 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
ReflectionUtils.doWithMethods(type, new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException {
try {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(input, method);
MessageChannel sharedChannel = locateSharedChannel(name);
if (sharedChannel == null) {
inputHolders.put(name, new ChannelHolder(
channelFactory.createChannel(name, method.getReturnType()), true));
}
else {
configureSharedMessageChannel(name, method.getReturnType(), sharedChannel);
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(input, method);
Class<? extends MessageChannel> channelType = (Class<? extends MessageChannel>) 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<? extends MessageChannel> channelType = (Class<? extends MessageChannel>) 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<? extends MessageChannel> 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<Obje
return this.channelNamespace + "." + name;
}
private void configureSharedMessageChannel(String name, Class<?> channelType, MessageChannel sharedChannel)
throws Exception {
if (channelType.isAssignableFrom(sharedChannel.getClass())) {
messageConverterConfigurer.configureMessageConverters(sharedChannel, name);
private void bridgeSharedChannel(Class<? extends MessageChannel> 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<Obje
}
private void bridgeSubscribableToPollableChannel(SubscribableChannel sharedChannel, MessageChannel inputChannel) {
sharedChannel.subscribe(new MessageChannelBinderSupport.DirectHandler(
inputChannel));
sharedChannel.subscribe(new MessageChannelBinderSupport.DirectHandler(inputChannel));
}
private void bridgePollableToSubscribableChannel(MessageChannel pollableChannel,

22
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelFactory.java
View File

@@ -15,7 +15,8 @@
*/
package org.springframework.cloud.stream.binding;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* Defines methods to create/configure the {@link org.springframework.messaging.MessageChannel} defined
@@ -25,5 +26,22 @@ import org.springframework.messaging.MessageChannel;
*/
public interface ChannelFactory {
MessageChannel createChannel(String name, Class<?> 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);
}

6
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java
View File

@@ -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

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java
View File

@@ -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));
}