Add per-binding message type-conversion support

- For the message channels being created, set the corresponding `dataType` and `message-converters` based on the `contentType` value set per-binding
   - This is applicable for both direct-binding and the channels that connect to the binder

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

spring-cloud-stream/pom.xml

@@ -30,6 +30,11 @@
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-messaging</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-tuple</artifactId>
+			<version>1.0.0.BUILD-SNAPSHOT</version>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream-binder-local</artifactId>

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<Obje
 			log.debug(String.format("Binding inputs for %s:%s", this.channelNamespace, this.type));
 		}
 		for (Map.Entry<String, ChannelHolder> 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<Obje
 			log.debug(String.format("Binding outputs for %s:%s", this.channelNamespace, this.type));
 		}
 		for (Map.Entry<String, ChannelHolder> 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);
 			}
 		}
 	}

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<MessageChannel> binder;
 
 	private 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 (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);
+			}
+		}
+	}
 }

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;
+	}
+
 }