Set message header using BindingProperties' contentType

- Update MessageConverterConfigurer that uses a channel interceptor to check if the message contentType header is missing and if there is a binding property for `contentType` header is provided, then set that value as the message contentType header

This resolves #397

Add test

spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeConfigurerTests.java

@@ -0,0 +1,66 @@
+/*
+ * 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.config;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.SpringApplicationConfiguration;
+import org.springframework.cloud.stream.annotation.Bindings;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.BinderFactory;
+import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.test.binder.TestSupportBinder;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Ilayaperumal Gopinathan
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringApplicationConfiguration({ContentTypeConfigurerTests.TestSource.class})
+public class ContentTypeConfigurerTests {
+
+	@Autowired @Bindings(TestSource.class)
+	private Source testSource;
+
+	@Autowired
+	private BinderFactory binderFactory;
+
+	@Test
+	public void testMessageHeaderWhenNoExplicitContentType() throws Exception {
+		testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}").build());
+		Message<String> received = (Message<String>) ((TestSupportBinder) binderFactory.getBinder(null)).messageCollector().forChannel(testSource.output()).poll();
+		assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString(), equalTo("application/json"));
+		assertThat(received.getPayload(), equalTo("{\"message\":\"Hi\"}"));
+	}
+
+	@EnableBinding(Source.class)
+	@EnableAutoConfiguration
+	@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties")
+	public static class TestSource {
+
+	}
+}
+

spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java

@@ -52,7 +52,7 @@ public class MessageChannelConfigurerTests {
 	private Sink testSink;
 
 	@Test
-	public void testContentTypeConfigurer() throws Exception {
+	public void testMessageConverterConfigurer() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
 		MessageHandler messageHandler = new MessageHandler() {
 			@Override

spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties

@@ -0,0 +1,2 @@
+spring.cloud.stream.bindings.output.destination=interceptor-test
+spring.cloud.stream.bindings.output.contentType=application/json

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java

@@ -40,15 +40,23 @@ import org.springframework.cloud.stream.converter.SerializedToJavaMessageConvert
 import org.springframework.cloud.stream.converter.StringToByteArrayMessageConverter;
 import org.springframework.cloud.stream.converter.TupleToJsonMessageConverter;
 import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.ChannelInterceptorAware;
+import org.springframework.integration.support.MessageBuilderFactory;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.support.ChannelInterceptorAdapter;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.StringUtils;
 
 /**
- * A {@link MessageChannelConfigurer} that sets the datatype and message converters for the message channel.
+ * A {@link MessageChannelConfigurer} that sets the datatypes and message converters based on the ContentType set via
+ * {@link BindingProperties}. This also adds a {@link org.springframework.messaging.support.ChannelInterceptor} to
+ * the message channel to set the `ContentType` header for the message (if not already set) based on the `ContentType` binding
+ * property of the channel.
  *
  * @author Ilayaperumal Gopinathan
  */
@@ -62,10 +70,14 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea
 
 	private final Collection<AbstractFromMessageConverter> customMessageConverters;
 
+	private final MessageBuilderFactory messageBuilderFactory;
+
 	public MessageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties,
-			Collection<AbstractFromMessageConverter> customMessageConverters) {
+			Collection<AbstractFromMessageConverter> customMessageConverters,
+			MessageBuilderFactory messageBuilderFactory) {
 		this.channelBindingServiceProperties = channelBindingServiceProperties;
 		this.customMessageConverters = customMessageConverters;
+		this.messageBuilderFactory = messageBuilderFactory;
 	}
 
 	@Override
@@ -103,14 +115,27 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea
 		Assert.isAssignable(AbstractMessageChannel.class, channel.getClass());
 		AbstractMessageChannel messageChannel = (AbstractMessageChannel) channel;
 		BindingProperties bindingProperties = this.channelBindingServiceProperties.getBindingProperties(channelName);
-		if (bindingProperties != null) {
-			String contentType = bindingProperties.getContentType();
-			if (StringUtils.hasText(contentType)) {
+		final String contentType = bindingProperties.getContentType();
+		if (bindingProperties != null && StringUtils.hasText(contentType)) {
 				MimeType mimeType = MessageConverterUtils.getMimeType(contentType);
 				MessageConverter messageConverter = this.messageConverterFactory.newInstance(mimeType);
 				Class<?>[] supportedDataTypes = this.messageConverterFactory.supportedDataTypes(mimeType);
 				messageChannel.setDatatypes(supportedDataTypes);
 				messageChannel.setMessageConverter(messageConverter);
+			if (messageChannel instanceof ChannelInterceptorAware) {
+				((ChannelInterceptorAware) messageChannel).addInterceptor(new ChannelInterceptorAdapter() {
+					@Override
+					public Message<?> preSend(Message<?> message, MessageChannel messageChannel) {
+						Object contentTypeFromMessage = message.getHeaders().get(MessageHeaders.CONTENT_TYPE);
+						if (contentTypeFromMessage == null) {
+							return messageBuilderFactory
+									.fromMessage(message)
+									.setHeader(MessageHeaders.CONTENT_TYPE, contentType)
+									.build();
+						}
+						return message;
+					}
+				});
 			}
 		}
 	}

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

@@ -98,7 +98,8 @@ public class ChannelBindingServiceConfiguration {
 	@Bean
 	public MessageConverterConfigurer messageConverterConfigurer
 			(ChannelBindingServiceProperties channelBindingServiceProperties) {
-		return new MessageConverterConfigurer(channelBindingServiceProperties, customMessageConverters);
+		return new MessageConverterConfigurer(channelBindingServiceProperties, customMessageConverters,
+				messageBuilderFactory);
 	}
 
 	@Bean