GH-1291 Ensured converters honor default content type

- Added DefaultContentTypeResolver with explicitly set default content type to all MessageConverters
- Added default content type constant to BindingProperties
- Test and minor polishing

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

@@ -23,6 +23,8 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include;
 
 import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
 import org.springframework.validation.annotation.Validated;
 
 /**
@@ -36,6 +38,8 @@ import org.springframework.validation.annotation.Validated;
 @JsonInclude(Include.NON_DEFAULT)
 @Validated
 public class BindingProperties {
+	
+	public static final MimeType DEFAULT_CONTENT_TYPE = MimeTypeUtils.APPLICATION_JSON;
 
 	private static final String COMMA = ",";
 
@@ -60,7 +64,7 @@ public class BindingProperties {
 	 * Specifies content-type that will be used by this binding in the event 
 	 * it is not specified in Message headers. Default: 'application/json'.
 	 */
-	private String contentType = "application/json";
+	private String contentType = DEFAULT_CONTENT_TYPE.toString();
 
 	/**
 	 * The name of the binder to use for this binding in the event multiple binders available (e.g., 'rabbit');

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java

@@ -47,6 +47,12 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 	
 	private final Map<ParameterizedTypeReference<?>, JavaType> typeCache = new ConcurrentHashMap<>();
 	
+	ApplicationJsonMessageMarshallingConverter(@Nullable ObjectMapper objectMapper) {
+		if (objectMapper != null) {
+			this.setObjectMapper(objectMapper);
+		}	
+	}
+	
 	@Override
 	protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) {
 		if (payload instanceof byte[]) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java

@@ -25,9 +25,11 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.messaging.converter.AbstractMessageConverter;
 import org.springframework.messaging.converter.ByteArrayMessageConverter;
 import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.messaging.converter.DefaultContentTypeResolver;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.MimeType;
@@ -66,20 +68,24 @@ public class CompositeMessageConverterFactory {
 			this.converters = new ArrayList<>();
 		}
 		initDefaultConverters();
+		
+		DefaultContentTypeResolver resolver = new DefaultContentTypeResolver();
+		resolver.setDefaultMimeType(BindingProperties.DEFAULT_CONTENT_TYPE);
+		this.converters.stream()
+			.filter(mc -> mc instanceof AbstractMessageConverter)
+			.forEach(mc -> ((AbstractMessageConverter)mc).setContentTypeResolver(resolver));
 	}
 
 	@SuppressWarnings("deprecation")
 	private void initDefaultConverters() {
-		ApplicationJsonMessageMarshallingConverter applicationJson = new ApplicationJsonMessageMarshallingConverter();
-		applicationJson.setStrictContentTypeMatch(true);
-		if (this.objectMapper != null) {
-			applicationJson.setObjectMapper(this.objectMapper);
-		}
-		this.converters.add(applicationJson);
-		
+		ApplicationJsonMessageMarshallingConverter applicationJsonConverter = new ApplicationJsonMessageMarshallingConverter(this.objectMapper);
+		applicationJsonConverter.setStrictContentTypeMatch(true);
+		this.converters.add(applicationJsonConverter);		
 		this.converters.add(new TupleJsonMessageConverter(this.objectMapper));
 		this.converters.add(new ByteArrayMessageConverter());
 		this.converters.add(new ObjectStringMessageConverter());
+		
+		// Deprecated converters
 		this.converters.add(new JavaSerializationMessageConverter());
 		this.converters.add(new KryoMessageConverter(null,true));
 		this.converters.add(new JsonUnmarshallingConverter(this.objectMapper));

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -31,7 +31,9 @@ import org.springframework.messaging.converter.AbstractMessageConverter;
 /**
  * @author Marius Bogoevici
  * @author Oleg Zhurakousky
+ * @deprecated as of 2.0. Will be removed in 2.1
  */
+@Deprecated
 public class JavaSerializationMessageConverter extends AbstractMessageConverter {
 
 	public JavaSerializationMessageConverter() {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonUnmarshallingConverter.java

@@ -30,7 +30,11 @@ import org.springframework.messaging.converter.MessageConversionException;
  * as input.
  *
  * @author Marius Bogoevici
+ * 
+ * @deprecated as of 2.0. 
  */
+// NOTE we need to revisit as to why do we need it in the first place, given that our first converter already handles JSON
+@Deprecated
 public class JsonUnmarshallingConverter extends AbstractMessageConverter {
 
 	private final ObjectMapper objectMapper;

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java

@@ -41,8 +41,10 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.AbstractMessageConverter;
 import org.springframework.messaging.converter.MessageConversionException;
@@ -69,6 +71,19 @@ import static org.junit.Assert.assertTrue;
  */
 public class ContentTypeTckTests {
 	
+	@Test
+	public void withInternalPipeline() {
+		ApplicationContext context = new SpringApplicationBuilder(InternalPipeLine.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false");
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		String jsonPayload = "{\"name\":\"oleg\"}";
+		source.send(new GenericMessage<byte[]>(jsonPayload.getBytes()));
+		Message<byte[]> outputMessage = target.receive();
+		assertEquals("OLEG", new String(outputMessage.getPayload()));
+	}
+	
 	@Test
 	public void pojoToPojo() {
 		ApplicationContext context = new SpringApplicationBuilder(PojoToPojoStreamListener.class)
@@ -533,6 +548,27 @@ public class ContentTypeTckTests {
 		}
 	}
 	
+	@EnableBinding(Processor.class)
+	@Import(TestChannelBinderConfiguration.class)
+	public static class InternalPipeLine {
+		@StreamListener(Processor.INPUT)
+		@SendTo("internalChannel")
+		public String handleA(Person value)  {
+			return "{\"name\":\"" + value.getName().toUpperCase() + "\"}";
+		}
+		
+		@Bean
+		public MessageChannel internalChannel() {
+			return new DirectChannel();
+		}
+		
+		@StreamListener("internalChannel")
+		@SendTo(Processor.OUTPUT)
+		public String handleB(Person value)  {
+			return value.toString();
+		}
+	}
+	
 	public static class Employee<P> {
 		private P person;
 		private int id;