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

Restrict Message channel types on input/output more explicitly

Browse Source 
- Add necessary assertions on the type of message channels for input/output
  - Make sure to support message channel of type `MessageChannel` for @Output-annotated channels
  - support message channel of type `SubscribableChannel` for @Input-annotated channels and if the type is exactly equal to `MessageChannel` type then create subscribable channel
 - Remove references of creating PollableChannel for binding

This resolves #469

Remove bridging of shared channel
This commit is contained in:
Ilayaperumal Gopinathan
2016-04-05 13:02:44 +05:30
committed by Marius Bogoevici
parent 0b4b916fe6
commit a0cdbdc148
3 changed files with 15 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableChannelFactory.java
View File

@@ -16,7 +16,6 @@
package org.springframework.cloud.stream.binding;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
/**
@@ -36,13 +35,4 @@ public interface BindableChannelFactory {
*/
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);
}

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

@@ -27,28 +27,18 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.cloud.stream.aggregate.SharedChannelRegistry;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binder.DirectHandler;
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;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
@@ -61,31 +51,23 @@ import org.springframework.util.ReflectionUtils;
*
* @see EnableBinding
*/
public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Object>, Bindable, BeanFactoryAware,
InitializingBean {
public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Object>, Bindable, InitializingBean {
private static Log log = LogFactory.getLog(BindableProxyFactory.class);
private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";
private static final String POLLABLE_BRIDGE_INTERVAL_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".pollableBridge.interval";
private static final String CHANNEL_NAMESPACE_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
@Value("${" + CHANNEL_NAMESPACE_PROPERTY_NAME + ":}")
private String channelNamespace;
@Value("${" + POLLABLE_BRIDGE_INTERVAL_PROPERTY_NAME + ":1000}")
private int pollableBridgeDefaultFrequency;
@Autowired
private BindableChannelFactory channelFactory;
@Autowired(required = false)
private SharedChannelRegistry sharedChannelRegistry;
private ConfigurableListableBeanFactory beanFactory;
private Class<?> type;
private Object proxy = null;
@@ -122,20 +104,18 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
ReflectionUtils.doWithMethods(type, new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException {
Assert.notNull(channelFactory, "Channel Factory cannot be null");
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(input, method);
@SuppressWarnings("unchecked")
Class<? extends MessageChannel> channelType = (Class<? extends MessageChannel>) method.getReturnType();
Assert.isTrue(MessageChannel.class.isAssignableFrom(method.getReturnType()),
"Input channel should be of type 'MessageChannel'");
MessageChannel sharedChannel = locateSharedChannel(name);
if (sharedChannel == null) {
inputHolders.put(name, new ChannelHolder(createBindableChannel(name, channelType), true));
inputHolders.put(name, new ChannelHolder(channelFactory.createSubscribableChannel(name), true));
}
else {
inputHolders.put(name, new ChannelHolder(sharedChannel, false));
if (!channelType.isAssignableFrom(sharedChannel.getClass())) {
bridgeSharedChannel(channelType, sharedChannel);
}
}
}
}
@@ -146,17 +126,20 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(output, method);
@SuppressWarnings("unchecked")
Class<? extends MessageChannel> channelType = (Class<? extends MessageChannel>) method.getReturnType();
if (method.getReturnType().equals(MessageChannel.class)) {
log.debug("Output channel is a MessageChannel. Creating a Subscribable Channel for binding.");
}
// Make sure any type other than `MessageChannel` should be of type `SubscribableChannel`
else {
Assert.isTrue(SubscribableChannel.class.isAssignableFrom(method.getReturnType()),
"Output channel should be of type 'SubscribableChannel'");
}
MessageChannel sharedChannel = locateSharedChannel(name);
if (sharedChannel == null) {
outputHolders.put(name, new ChannelHolder(createBindableChannel(name, channelType), true));
outputHolders.put(name, new ChannelHolder(channelFactory.createSubscribableChannel(name), true));
}
else {
outputHolders.put(name, new ChannelHolder(sharedChannel, false));
if (!channelType.isAssignableFrom(sharedChannel.getClass())) {
bridgeSharedChannel(channelType, sharedChannel);
}
}
}
}
@@ -164,11 +147,6 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
});
}
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;
@@ -178,45 +156,6 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
return this.channelNamespace + "." + 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((PollableChannel) sharedChannel, new DirectChannel());
}
else {
bridgeSubscribableToPollableChannel((SubscribableChannel) sharedChannel, new QueueChannel());
}
}
private boolean isPollable(Class<?> channelType) {
return PollableChannel.class.equals(channelType);
}
private void bridgeSubscribableToPollableChannel(SubscribableChannel sharedChannel, PollableChannel pollableChannel) {
sharedChannel.subscribe(new DirectHandler(pollableChannel));
}
private void bridgePollableToSubscribableChannel(PollableChannel pollableChannel,
SubscribableChannel subscribableChannel) {
ConsumerEndpointFactoryBean consumerEndpointFactoryBean = new ConsumerEndpointFactoryBean();
consumerEndpointFactoryBean.setInputChannel(pollableChannel);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(this.pollableBridgeDefaultFrequency));
consumerEndpointFactoryBean.setPollerMetadata(pollerMetadata);
consumerEndpointFactoryBean
.setHandler(new DirectHandler(
subscribableChannel));
consumerEndpointFactoryBean.setBeanFactory(this.beanFactory);
try {
consumerEndpointFactoryBean.afterPropertiesSet();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
consumerEndpointFactoryBean.start();
}
@Override
public synchronized Object getObject() throws Exception {
if (this.proxy == null) {
@@ -310,11 +249,6 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
return this.outputHolders.keySet();
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
}
/**
* Holds information about the channels exposed by the interface proxy, as well as
* their status.

9
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultBindableChannelFactory.java
View File

@@ -16,8 +16,6 @@
package org.springframework.cloud.stream.binding;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
/**
@@ -35,13 +33,6 @@ public class DefaultBindableChannelFactory implements BindableChannelFactory {
this.messageChannelConfigurer = messageChannelConfigurer;
}
@Override
public PollableChannel createPollableChannel(String name) {
PollableChannel pollableChannel = new QueueChannel();
messageChannelConfigurer.configureMessageChannel(pollableChannel, name);
return pollableChannel;
}
@Override
public SubscribableChannel createSubscribableChannel(String name) {
SubscribableChannel subscribableChannel = new DirectChannel();