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

Use ChannelFactory

Browse Source 
- ChannelFactory to create message channels and configure message converters

Autowire ChannelFactory

Create message channel on demand

ChannelFactory implementation to use SharedChannelRegistry

Create ChannelFactory interface
This commit is contained in:
Ilayaperumal Gopinathan
2015-10-23 18:09:37 +05:30
committed by Marius Bogoevici
parent d3c87daf64
commit ae770cec69
7 changed files with 323 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,7 +20,6 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binding.BindableProxyFactory;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
@@ -33,6 +32,10 @@ import org.springframework.integration.channel.DirectChannel;
*/
public class AggregateApplication {
private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";
private static final String CHANNEL_NAMESPACE_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
public static final String INPUT_CHANNEL_NAME = "input";
public static final String OUTPUT_CHANNEL_NAME = "output";
@@ -103,7 +106,7 @@ public class AggregateApplication {
return new SpringApplicationBuilder(module)
.web(false)
.showBanner(false)
.properties(BindableProxyFactory.CHANNEL_NAMESPACE_PROPERTY_NAME + "=" + namespace)
.properties(CHANNEL_NAMESPACE_PROPERTY_NAME + "=" + namespace)
.registerShutdownHook(false)
.parent(applicationContext);
}

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

@@ -25,6 +25,7 @@ import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
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;
@@ -40,15 +41,12 @@ 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;
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,18 +59,16 @@ import org.springframework.util.ReflectionUtils;
*
* @see EnableBinding
*/
public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Object>,
BeanFactoryAware, Bindable, InitializingBean {
public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Object>, Bindable, BeanFactoryAware,
InitializingBean {
private static Log log = LogFactory.getLog(BindableProxyFactory.class);
public static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";
private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";
public static final String CHANNEL_NAMESPACE_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
private static final String POLLABLE_BRIDGE_INTERVAL_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".pollableBridge.interval";
public static final String POLLABLE_BRIDGE_INTERVAL_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".pollableBridge.interval";
private Class<?> type;
private static final String CHANNEL_NAMESPACE_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
@Value("${" + CHANNEL_NAMESPACE_PROPERTY_NAME + ":}")
private String channelNamespace;
@@ -80,123 +76,130 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
@Value("${" + POLLABLE_BRIDGE_INTERVAL_PROPERTY_NAME + ":1000}")
private int pollableBridgeDefaultFrequency;
private Object proxy = null;
@Autowired
private ChannelFactory channelFactory;
private Map<String, ChannelHolder> inputs = new HashMap<>();
private Map<String, ChannelHolder> outputs = new HashMap<>();
private ConfigurableListableBeanFactory beanFactory;
@Autowired
private MessageConverterConfigurer messageConverterConfigurer;
@Autowired(required = false)
private SharedChannelRegistry sharedChannelRegistry;
private ConfigurableListableBeanFactory beanFactory;
private Class<?> type;
private Object proxy = null;
private Map<String, ChannelHolder> inputHolders = new HashMap<>();
private Map<String, ChannelHolder> outputHolders = new HashMap<>();
public BindableProxyFactory(Class<?> type) {
this.type = type;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
public synchronized Object invoke(MethodInvocation invocation) throws Throwable {
Method method = invocation.getMethod();
if (MessageChannel.class.isAssignableFrom(method.getReturnType())) {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(input, method);
return this.inputHolders.get(name).getMessageChannel();
}
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(output, method);
return this.outputHolders.get(name).getMessageChannel();
}
}
//ignore
return null;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(this.beanFactory, "Bean factory cannot be empty");
}
@Override
public Set<String> getInputs() {
return this.inputs.keySet();
}
@Override
public Set<String> getOutputs() {
return this.outputs.keySet();
}
private void createChannels(Class<?> type) throws Exception {
ReflectionUtils.doWithMethods(type, new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException,
IllegalAccessException {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(
input, method);
Class<?> inputChannelType = method.getReturnType();
MessageChannel sharedChannel = locateSharedChannel(name);
if (sharedChannel == null) {
MessageChannel inputChannel = createMessageChannel(inputChannelType);
BindableProxyFactory.this.inputs.put(name, new ChannelHolder(inputChannel, true));
}
else {
if (inputChannelType.isAssignableFrom(sharedChannel.getClass())) {
BindableProxyFactory.this.inputs.put(name, new ChannelHolder(sharedChannel, false));
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 {
// 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 = createMessageChannel(inputChannelType);
if (isPollable(sharedChannel.getClass())) {
bridgePollableToSubscribableChannel(sharedChannel,
inputChannel);
}
else {
bridgeSubscribableToPollableChannel(
(SubscribableChannel) sharedChannel, inputChannel);
}
BindableProxyFactory.this.inputs.put(name, new ChannelHolder(inputChannel, false));
configureSharedMessageChannel(name, method.getReturnType(), sharedChannel);
}
}
}
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(
output, method);
Class<?> messageChannelType = method.getReturnType();
MessageChannel sharedChannel = locateSharedChannel(name);
if (sharedChannel == null) {
MessageChannel outputChannel = createMessageChannel(messageChannelType);
BindableProxyFactory.this.outputs.put(name, new ChannelHolder(outputChannel, true));
}
else {
if (messageChannelType.isAssignableFrom(sharedChannel.getClass())) {
BindableProxyFactory.this.outputs.put(name, new ChannelHolder(sharedChannel, false));
}
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 outputChannel = createMessageChannel(messageChannelType);
if (isPollable(messageChannelType)) {
bridgePollableToSubscribableChannel(outputChannel,
sharedChannel);
}
else {
bridgeSubscribableToPollableChannel(
(SubscribableChannel) outputChannel,
sharedChannel);
}
BindableProxyFactory.this.outputs.put(name, new ChannelHolder(outputChannel, false));
}
}
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);
}
}
}
catch (Exception e) {
throw new RuntimeException(e);
}
}
});
}
private MessageChannel locateSharedChannel(String name) {
return this.sharedChannelRegistry != null ? this.sharedChannelRegistry.get(getNamespacePrefixedChannelName(name)) : null;
return this.sharedChannelRegistry != null ?
this.sharedChannelRegistry.get(getNamespacePrefixedChannelName(name)) : null;
}
private String getNamespacePrefixedChannelName(String name) {
return this.channelNamespace + "." + name;
}
private void bridgeSubscribableToPollableChannel(SubscribableChannel sharedChannel,
MessageChannel inputChannel) {
private void configureSharedMessageChannel(String name, Class<?> channelType, MessageChannel sharedChannel)
throws Exception {
if (channelType.isAssignableFrom(sharedChannel.getClass())) {
messageConverterConfigurer.configureMessageConverters(sharedChannel, name);
}
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);
}
}
private boolean isPollable(Class<?> channelType) {
return PollableChannel.class.equals(channelType);
}
private void bridgeSubscribableToPollableChannel(SubscribableChannel sharedChannel, MessageChannel inputChannel) {
sharedChannel.subscribe(new MessageChannelBinderSupport.DirectHandler(
inputChannel));
}
@@ -214,45 +217,16 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
consumerEndpointFactoryBean.setBeanFactory(this.beanFactory);
try {
consumerEndpointFactoryBean.afterPropertiesSet();
} catch (Exception e) {
}
catch (Exception e) {
throw new IllegalStateException(e);
}
consumerEndpointFactoryBean.start();
}
private MessageChannel createMessageChannel(Class<?> messageChannelType) {
return isPollable(messageChannelType) ? new QueueChannel() : new DirectChannel();
}
private boolean isPollable(Class<?> channelType) {
return PollableChannel.class.equals(channelType);
}
@Override
public synchronized Object invoke(MethodInvocation invocation) throws Throwable {
Method method = invocation.getMethod();
if (MessageChannel.class.isAssignableFrom(method.getReturnType())) {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(input,
method);
return this.inputs.get(name).getMessageChannel();
}
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(output,
method);
return this.outputs.get(name).getMessageChannel();
}
}
// ignore
return null;
}
@Override
public synchronized Object getObject() throws Exception {
if (this.proxy == null) {
createChannels(this.type);
ProxyFactory factory = new ProxyFactory(this.type, this);
this.proxy = factory.getProxy();
}
@@ -274,10 +248,9 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
if (log.isDebugEnabled()) {
log.debug(String.format("Binding inputs for %s:%s", this.channelNamespace, this.type));
}
for (Map.Entry<String, ChannelHolder> channelHolderEntry : this.inputs.entrySet()) {
for (Map.Entry<String, ChannelHolder> channelHolderEntry : this.inputHolders.entrySet()) {
String inputChannelName = channelHolderEntry.getKey();
ChannelHolder channelHolder = channelHolderEntry.getValue();
channelBindingService.configureMessageConverters(channelHolder.getMessageChannel(), inputChannelName);
if (channelHolder.isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.channelNamespace, this.type, inputChannelName));
@@ -292,10 +265,9 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
if (log.isDebugEnabled()) {
log.debug(String.format("Binding outputs for %s:%s", this.channelNamespace, this.type));
}
for (Map.Entry<String, ChannelHolder> channelHolderEntry : this.outputs.entrySet()) {
for (Map.Entry<String, ChannelHolder> channelHolderEntry : this.outputHolders.entrySet()) {
ChannelHolder channelHolder = channelHolderEntry.getValue();
String outputChannelName = channelHolderEntry.getKey();
channelBindingService.configureMessageConverters(channelHolder.getMessageChannel(), outputChannelName);
if (channelHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.channelNamespace, this.type, outputChannelName));
@@ -310,7 +282,7 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding inputs for %s:%s", this.channelNamespace, this.type));
}
for (Map.Entry<String, ChannelHolder> channelHolderEntry : this.inputs.entrySet()) {
for (Map.Entry<String, ChannelHolder> channelHolderEntry : this.inputHolders.entrySet()) {
if (channelHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding %s:%s:%s", this.channelNamespace, this.type, channelHolderEntry.getKey()));
@@ -325,7 +297,7 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding outputs for %s:%s", this.channelNamespace, this.type));
}
for (Map.Entry<String, ChannelHolder> channelHolderEntry : this.outputs.entrySet()) {
for (Map.Entry<String, ChannelHolder> channelHolderEntry : this.outputHolders.entrySet()) {
if (channelHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.channelNamespace, this.type, channelHolderEntry.getKey()));
@@ -335,12 +307,26 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
}
}
@Override
public Set<String> getInputs() {
return this.inputHolders.keySet();
}
@Override
public Set<String> getOutputs() {
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.
*
*/
static class ChannelHolder {
class ChannelHolder {
private MessageChannel messageChannel;
@@ -359,5 +345,4 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
return this.bindable;
}
}
}

76
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java
View File

@@ -16,33 +16,12 @@
package org.springframework.cloud.stream.binding;
import java.util.HashSet;
import java.util.Set;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderUtils;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
import org.springframework.cloud.stream.converter.AbstractFromMessageConverter;
import org.springframework.cloud.stream.converter.ByteArrayToStringMessageConverter;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.cloud.stream.converter.JavaToSerializedMessageConverter;
import org.springframework.cloud.stream.converter.JsonToPojoMessageConverter;
import org.springframework.cloud.stream.converter.JsonToTupleMessageConverter;
import org.springframework.cloud.stream.converter.MessageConverterUtils;
import org.springframework.cloud.stream.converter.PojoToJsonMessageConverter;
import org.springframework.cloud.stream.converter.PojoToStringMessageConverter;
import org.springframework.cloud.stream.converter.SerializedToJavaMessageConverter;
import org.springframework.cloud.stream.converter.StringToByteArrayMessageConverter;
import org.springframework.cloud.stream.converter.TupleToJsonMessageConverter;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.StringUtils;
/**
@@ -54,34 +33,17 @@ import org.springframework.util.StringUtils;
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class ChannelBindingService implements InitializingBean {
public class ChannelBindingService {
private final Binder<MessageChannel> binder;
private final ChannelBindingServiceProperties channelBindingServiceProperties;
private CompositeMessageConverterFactory messageConverterFactory;
public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties, Binder<MessageChannel> binder) {
this.channelBindingServiceProperties = channelBindingServiceProperties;
this.binder = binder;
}
@Override
public void afterPropertiesSet() throws Exception {
Set<AbstractFromMessageConverter> messageConverters = new HashSet<>();
messageConverters.add(new JsonToTupleMessageConverter());
messageConverters.add(new TupleToJsonMessageConverter());
messageConverters.add(new JsonToPojoMessageConverter());
messageConverters.add(new PojoToJsonMessageConverter());
messageConverters.add(new ByteArrayToStringMessageConverter());
messageConverters.add(new StringToByteArrayMessageConverter());
messageConverters.add(new PojoToStringMessageConverter());
messageConverters.add(new JavaToSerializedMessageConverter());
messageConverters.add(new SerializedToJavaMessageConverter());
this.messageConverterFactory = new CompositeMessageConverterFactory(messageConverters);
}
public void bindConsumer(MessageChannel inputChannel, String inputChannelName) {
String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(inputChannelName);
if (BinderUtils.isChannelPubSub(channelBindingTarget)) {
@@ -122,40 +84,4 @@ public class ChannelBindingService implements InitializingBean {
public void unbindProducers(String outputChannelName) {
this.binder.unbindProducers(outputChannelName);
}
/**
* Setup data-type and message converters for the given message channel.
*
* @param channel message channel to set the data-type and message converters
* @param channelName the channel name
*/
public void configureMessageConverters(Object channel, String channelName) {
AbstractMessageChannel messageChannel = null;
try {
messageChannel = getMessageChannel(channel);
}
catch (Exception e) {
throw new IllegalStateException("Could not get the message channel to configure message converters" + e);
}
BindingProperties bindingProperties = this.channelBindingServiceProperties.getBindings().get(channelName);
if (bindingProperties != null) {
String contentType = bindingProperties.getContentType();
if (StringUtils.hasText(contentType)) {
MimeType mimeType = MessageConverterUtils.getMimeType(contentType);
MessageConverter messageConverter = this.messageConverterFactory.newInstance(mimeType);
Class<?> dataType = MessageConverterUtils.getJavaTypeForContentType(mimeType,
Thread.currentThread().getContextClassLoader());
messageChannel.setDatatypes(dataType);
messageChannel.setMessageConverter(messageConverter);
}
}
}
private AbstractMessageChannel getMessageChannel(Object channel) throws Exception {
if (AopUtils.isJdkDynamicProxy(channel)) {
return (AbstractMessageChannel) (((Advised) channel).getTargetSource().getTarget());
}
Assert.isAssignable(AbstractMessageChannel.class, channel.getClass());
return (AbstractMessageChannel) channel;
}
}

29
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelFactory.java Normal file
View File

@@ -0,0 +1,29 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binding;
import org.springframework.messaging.MessageChannel;
/**
* Defines methods to create/configure the {@link org.springframework.messaging.MessageChannel} defined
* in {@link org.springframework.cloud.stream.annotation.EnableBinding}.
*
* @author Ilayaperumal Gopinathan
*/
public interface ChannelFactory {
MessageChannel createChannel(String name, Class<?> channelType) throws Exception;
}

50
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultChannelFactory.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
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;
/**
* Class that {@link BindableProxyFactory} uses to create message channels.
*
* @author Marius Bogoevici
* @author David Syer
* @author Ilayaperumal Gopinathan
*/
public class DefaultChannelFactory implements ChannelFactory {
@Autowired
MessageConverterConfigurer messageConverterConfigurer;
@Override
public MessageChannel createChannel(String name, Class<?> inputChannelType) throws Exception {
MessageChannel messageChannel = createMessageChannel(inputChannelType);
messageConverterConfigurer.configureMessageConverters(messageChannel, name);
return messageChannel;
}
private MessageChannel createMessageChannel(Class<?> messageChannelType) {
return isPollable(messageChannelType) ? new QueueChannel() : new DirectChannel();
}
private boolean isPollable(Class<?> channelType) {
return PollableChannel.class.equals(channelType);
}
}

106
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java Normal file
View File

@@ -0,0 +1,106 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binding;
import java.util.HashSet;
import java.util.Set;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
import org.springframework.cloud.stream.converter.AbstractFromMessageConverter;
import org.springframework.cloud.stream.converter.ByteArrayToStringMessageConverter;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.cloud.stream.converter.JavaToSerializedMessageConverter;
import org.springframework.cloud.stream.converter.JsonToPojoMessageConverter;
import org.springframework.cloud.stream.converter.JsonToTupleMessageConverter;
import org.springframework.cloud.stream.converter.MessageConverterUtils;
import org.springframework.cloud.stream.converter.PojoToJsonMessageConverter;
import org.springframework.cloud.stream.converter.PojoToStringMessageConverter;
import org.springframework.cloud.stream.converter.SerializedToJavaMessageConverter;
import org.springframework.cloud.stream.converter.StringToByteArrayMessageConverter;
import org.springframework.cloud.stream.converter.TupleToJsonMessageConverter;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.StringUtils;
/**
*
* @author Ilayaperumal Gopinathan
*/
public class MessageConverterConfigurer implements BeanFactoryAware, InitializingBean {
private ConfigurableListableBeanFactory beanFactory;
private CompositeMessageConverterFactory messageConverterFactory;
private final ChannelBindingServiceProperties channelBindingServiceProperties;
public MessageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties) {
this.channelBindingServiceProperties = channelBindingServiceProperties;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(this.beanFactory, "Bean factory cannot be empty");
Set<AbstractFromMessageConverter> messageConverters = new HashSet<>();
messageConverters.add(new JsonToTupleMessageConverter());
messageConverters.add(new TupleToJsonMessageConverter());
messageConverters.add(new JsonToPojoMessageConverter());
messageConverters.add(new PojoToJsonMessageConverter());
messageConverters.add(new ByteArrayToStringMessageConverter());
messageConverters.add(new StringToByteArrayMessageConverter());
messageConverters.add(new PojoToStringMessageConverter());
messageConverters.add(new JavaToSerializedMessageConverter());
messageConverters.add(new SerializedToJavaMessageConverter());
this.messageConverterFactory = new CompositeMessageConverterFactory(messageConverters);
}
/**
* Setup data-type and message converters for the given message channel.
*
* @param channel message channel to set the data-type and message converters
* @param channelName the channel name
*/
void configureMessageConverters(MessageChannel channel, String channelName) {
Assert.isAssignable(AbstractMessageChannel.class, channel.getClass());
AbstractMessageChannel messageChannel = (AbstractMessageChannel) channel;
BindingProperties bindingProperties = this.channelBindingServiceProperties.getBindings().get(channelName);
if (bindingProperties != null) {
String contentType = bindingProperties.getContentType();
if (StringUtils.hasText(contentType)) {
MimeType mimeType = MessageConverterUtils.getMimeType(contentType);
MessageConverter messageConverter = this.messageConverterFactory.newInstance(mimeType);
Class<?> dataType = MessageConverterUtils.getJavaTypeForContentType(mimeType,
Thread.currentThread().getContextClassLoader());
messageChannel.setDatatypes(dataType);
messageChannel.setMessageConverter(messageConverter);
}
}
}
}

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

@@ -31,8 +31,11 @@ import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor;
import org.springframework.cloud.stream.binding.ChannelBindingService;
import org.springframework.cloud.stream.binding.ChannelFactory;
import org.springframework.cloud.stream.binding.DefaultChannelFactory;
import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener;
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
import org.springframework.cloud.stream.tuple.spel.TuplePropertyAccessor;
import org.springframework.context.annotation.Bean;
@@ -70,6 +73,16 @@ public class ChannelBindingServiceConfiguration {
return new ChannelBindingService(channelBindingServiceProperties, binder);
}
@Bean
public MessageConverterConfigurer messageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties) {
return new MessageConverterConfigurer(channelBindingServiceProperties);
}
@Bean
public ChannelFactory channelFactory() {
return new DefaultChannelFactory();
}
@Bean
@DependsOn("bindingService")
public OutputBindingLifecycle outputBindingLifecycle() {