From a0cdbdc1486521c4c15b923f75b1e2e03532cf3d Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Tue, 5 Apr 2016 13:02:44 +0530 Subject: [PATCH] Restrict Message channel types on input/output more explicitly - 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 --- .../binding/BindableChannelFactory.java | 10 -- .../stream/binding/BindableProxyFactory.java | 96 +++---------------- .../DefaultBindableChannelFactory.java | 9 -- 3 files changed, 15 insertions(+), 100 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableChannelFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableChannelFactory.java index fdd5aea33..fa9a454bc 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableChannelFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableChannelFactory.java @@ -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); - } 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 e2c340cd3..27776221d 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 @@ -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, Bindable, BeanFactoryAware, - InitializingBean { +public class BindableProxyFactory implements MethodInterceptor, FactoryBean, 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 channelType = (Class) 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 channelType = (Class) 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 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 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