From ae770cec6997c1b3914e73d26b540c83e171396e Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Fri, 23 Oct 2015 18:09:37 +0530 Subject: [PATCH] Use ChannelFactory - ChannelFactory to create message channels and configure message converters Autowire ChannelFactory Create message channel on demand ChannelFactory implementation to use SharedChannelRegistry Create ChannelFactory interface --- .../aggregate/AggregateApplication.java | 7 +- .../stream/binding/BindableProxyFactory.java | 253 ++++++++---------- .../stream/binding/ChannelBindingService.java | 76 +----- .../cloud/stream/binding/ChannelFactory.java | 29 ++ .../stream/binding/DefaultChannelFactory.java | 50 ++++ .../binding/MessageConverterConfigurer.java | 106 ++++++++ .../ChannelBindingServiceConfiguration.java | 13 + 7 files changed, 323 insertions(+), 211 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelFactory.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultChannelFactory.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java 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 2d232489e..2e6b0df5b 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 @@ -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); } 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 bcc6c9df7..efa2b2755 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 @@ -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, - BeanFactoryAware, Bindable, InitializingBean { +public class BindableProxyFactory implements MethodInterceptor, FactoryBean, 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 inputs = new HashMap<>(); - - private Map 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 inputHolders = new HashMap<>(); + + private Map 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 getInputs() { - return this.inputs.keySet(); - } - - @Override - public Set 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 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 channelHolderEntry : this.inputs.entrySet()) { + for (Map.Entry 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 channelHolderEntry : this.outputs.entrySet()) { + for (Map.Entry 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 channelHolderEntry : this.inputs.entrySet()) { + for (Map.Entry 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 channelHolderEntry : this.outputs.entrySet()) { + for (Map.Entry 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 getInputs() { + return this.inputHolders.keySet(); + } + + @Override + public Set 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 binder; private final ChannelBindingServiceProperties channelBindingServiceProperties; - private CompositeMessageConverterFactory messageConverterFactory; - public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties, Binder binder) { this.channelBindingServiceProperties = channelBindingServiceProperties; this.binder = binder; } - @Override - public void afterPropertiesSet() throws Exception { - Set 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; - } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelFactory.java new file mode 100644 index 000000000..82e045952 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelFactory.java @@ -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; +} 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/DefaultChannelFactory.java new file mode 100644 index 000000000..0c75be947 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DefaultChannelFactory.java @@ -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); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java new file mode 100644 index 000000000..deb5ff544 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -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 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); + } + } + } +} 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 fbd0cf534..3ceeec658 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 @@ -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() {