diff --git a/spring-cloud-stream-samples/source/src/main/resources/application.yml b/spring-cloud-stream-samples/source/src/main/resources/application.yml index 04d5e2be3..72f476dff 100644 --- a/spring-cloud-stream-samples/source/src/main/resources/application.yml +++ b/spring-cloud-stream-samples/source/src/main/resources/application.yml @@ -5,7 +5,9 @@ spring: cloud: stream: bindings: - output: testtock + output: + destination: testtock + contentType: text/plain # uncomment below to use the last digit of the seconds as a partition key # hashcode(key) % N is then applied with N being the partitionCount value # thus, even seconds should go to the 0 queue, odd seconds to the 1 queue diff --git a/spring-cloud-stream/pom.xml b/spring-cloud-stream/pom.xml index ff98ee3bf..5f0b0cd3e 100644 --- a/spring-cloud-stream/pom.xml +++ b/spring-cloud-stream/pom.xml @@ -30,6 +30,11 @@ org.springframework spring-messaging + + org.springframework.cloud + spring-cloud-stream-tuple + 1.0.0.BUILD-SNAPSHOT + org.springframework.cloud spring-cloud-stream-binder-local 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 2c42ce4e2..c1cc96426 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 @@ -57,6 +57,7 @@ import org.springframework.util.ReflectionUtils; * * @author Marius Bogoevici * @author David Syer + * @author Ilayaperumal Gopinathan * * @see EnableBinding */ @@ -264,13 +265,14 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean channelHolderEntry : inputs.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, channelHolderEntry.getKey())); + log.debug(String.format("Binding %s:%s:%s", this.channelNamespace, this.type, inputChannelName)); } - channelBindingService.bindConsumer( - channelHolder.getMessageChannel(), channelHolderEntry.getKey()); + channelBindingService.bindConsumer(channelHolder.getMessageChannel(), inputChannelName); } } } @@ -281,12 +283,14 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean channelHolderEntry : outputs.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, channelHolderEntry.getKey())); + log.debug(String.format("Binding %s:%s:%s", this.channelNamespace, this.type, outputChannelName)); } - channelBindingService.bindProducer(channelHolderEntry.getValue() - .getMessageChannel(), channelHolderEntry.getKey()); + channelBindingService.bindProducer(channelHolder.getMessageChannel(), outputChannelName); } } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java index 9f8a6bcef..35fc8416e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java @@ -16,31 +16,69 @@ package org.springframework.cloud.stream.binding; +import java.util.HashSet; +import java.util.Set; + +import org.springframework.beans.factory.InitializingBean; import org.springframework.cloud.stream.binder.Binder; +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; /** - * Handles the binding of input/output channels by delegating to an underlying - * {@link Binder}. + * Handles the operations related to channel binding including binding of input/output channels by delegating + * to an underlying {@link Binder}, setting up data type conversion for binding channel. * * @author Mark Fisher * @author Dave Syer * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ -public class ChannelBindingService { +public class ChannelBindingService implements InitializingBean { private Binder binder; private 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 (isChannelPubSub(channelBindingTarget)) { @@ -82,4 +120,26 @@ public class ChannelBindingService { public void unbindProducers(String outputChannelName) { this.binder.unbindProducers(outputChannelName); } + + /** + * Setup data-type and message converters for the given message channel. + * + * @param messageChannel message channel to set the data-type and message converters + * @param channelName the channel name + */ + public void configureMessageConverters(MessageChannel messageChannel, String channelName) { + Assert.isAssignable(AbstractMessageChannel.class, messageChannel.getClass()); + BindingProperties bindingProperties = channelBindingServiceProperties.getBindings().get(channelName); + if (bindingProperties != null) { + String contentType = bindingProperties.getContentType(); + if (StringUtils.hasText(contentType)) { + MimeType mimeType = MessageConverterUtils.getMimeType(contentType); + MessageConverter messageConverter = messageConverterFactory.newInstance(mimeType); + Class dataType = MessageConverterUtils.getJavaTypeForContentType(mimeType, + Thread.currentThread().getContextClassLoader()); + ((AbstractMessageChannel)messageChannel).setDatatypes(dataType); + ((AbstractMessageChannel)messageChannel).setMessageConverter(messageConverter); + } + } + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index 0315d9ded..fc18e0a3e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -20,6 +20,7 @@ package org.springframework.cloud.stream.config; * Contains the properties of a binding. * * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ public class BindingProperties { @@ -37,6 +38,8 @@ public class BindingProperties { private String partitionSelectorExpression; + private String contentType; + public String getDestination() { return destination; } @@ -92,4 +95,13 @@ public class BindingProperties { public void setPartitionSelectorExpression(String partitionSelectorExpression) { this.partitionSelectorExpression = partitionSelectorExpression; } + + public String getContentType() { + return this.contentType; + } + + public void setContentType(String contentType) { + this.contentType = contentType; + } + }