Legacy cotent-type changes for the consumer

- Remove binding config property `legacyContentTypeHeaderEnabled` that was
  introduced to enable legacy content type handling
- Enable LegacyContentTypeInterceptor in 2.0 always, but bypass any
  legacy content type handling if the message received is from
  a 2.0 producer by checking on the version header
- Introdce a new BinderHeader property for version
- Fix the LegacyContentType related tests
- Remove the check for originalContentType in ReceivingHandler when
  the payload received is a byte[]
- Remove unnecessary deserializePayloadIfNecessary calls in ReceivingHandler
- Remove deprecated deserializePayload methods in AbstractBinder
  and MessageSerializationUtils

Partly fixes #1106
Fixes #1110

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

@@ -28,7 +28,6 @@ import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.messaging.Sink;
-import org.springframework.context.annotation.PropertySource;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
@@ -61,14 +60,16 @@ public class LegacyContentTypeTests {
 			}
 		};
 		testSink.input().subscribe(messageHandler);
-		testSink.input().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}".getBytes()).setHeader(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, "application/json").build());
+		testSink.input().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}".getBytes())
+							.setHeader(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, "application/json")
+							.setHeader(BinderHeaders.SCST_VERSION, "1.x")
+							.build());
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 		testSink.input().unsubscribe(messageHandler);
 	}
 
 	@EnableBinding(Sink.class)
 	@EnableAutoConfiguration
-	@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/legacy-sink-channel-configurers.properties")
 	public static class LegacyTestSink {
 
 	}

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

@@ -1,4 +0,0 @@
-spring.cloud.stream.bindings.input.destination=configure1
-spring.cloud.stream.bindings.input.legacyContentTypeHeaderEnabled=true
-spring.cloud.stream.bindings.input.contentType=application/x-spring-tuple
-

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java

@@ -151,16 +151,6 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
 		return MessageSerializationUtils.serializePayload(message);
 	}
 
-	@Deprecated
-	protected final MessageValues deserializePayloadIfNecessary(Message<?> message) {
-		return MessageSerializationUtils.deserializePayload(new MessageValues(message), this.contentTypeResolver);
-	}
-
-	@Deprecated
-	protected final MessageValues deserializePayloadIfNecessary(MessageValues messageValues) {
-		return MessageSerializationUtils.deserializePayload(messageValues, this.contentTypeResolver);
-	}
-
 	@Deprecated
 	protected String buildPartitionRoutingExpression(String expressionRoot) {
 		return "'" + expressionRoot + "-' + headers['" + BinderHeaders.PARTITION_HEADER + "']";

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -45,13 +45,12 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
-import org.springframework.util.MimeTypeUtils;
 
 /**
  * {@link AbstractBinder} that serves as base class for {@link MessageChannel} binders.
  * Implementors must implement the following methods:
  * <ul>
- * <li>{@link #createProducerMessageHandler(ProducerDestination, ProducerProperties)}</li>
+ * <li>{@link #createProducerMessageHandler(ProducerDestination, ProducerProperties, MessageChannel)}</li>
  * <li>{@link #createConsumerEndpoint(ConsumerDestination, String, ConsumerProperties)}
  * </li>
  * </ul>
@@ -94,7 +93,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	/**
 	 * Binds an outbound channel to a given destination. The implementation delegates to
 	 * {@link ProvisioningProvider#provisionProducerDestination(String, ProducerProperties)}
-	 * and {@link #createProducerMessageHandler(ProducerDestination, ProducerProperties)}
+	 * and {@link #createProducerMessageHandler(ProducerDestination, ProducerProperties, MessageChannel)}
 	 * for handling the middleware specific logic. If the returned producer message
 	 * handler is an {@link InitializingBean} then
 	 * {@link InitializingBean#afterPropertiesSet()} will be called on it. Similarly, if
@@ -535,14 +534,13 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		@Override
 		@SuppressWarnings("unchecked")
 		protected Object handleRequestMessage(Message<?> requestMessage) {
-			if (!(requestMessage.getPayload() instanceof byte[])
-					&& !requestMessage.getHeaders().containsKey(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)) {
+			if (!(requestMessage.getPayload() instanceof byte[])) {
 				return requestMessage;
 			}
-			MessageValues messageValues;
 			if (this.extractEmbeddedHeaders
 					&& !requestMessage.getHeaders().containsKey(BinderHeaders.NATIVE_HEADERS_PRESENT)
 					&& EmbeddedHeaderUtils.mayHaveEmbeddedHeaders((byte[]) requestMessage.getPayload())) {
+				MessageValues messageValues;
 				try {
 					messageValues = EmbeddedHeaderUtils.extractHeaders((Message<byte[]>) requestMessage,
 							true);
@@ -554,18 +552,11 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 							e);
 					messageValues = new MessageValues(requestMessage);
 				}
-				messageValues = deserializePayloadIfNecessary(messageValues);
+				return messageValues.toMessage();
 			}
 			else {
-				MimeType contentType = AbstractMessageChannelBinder.this.contentTypeResolver.resolve(requestMessage.getHeaders());
-				if (contentType != null && !MimeTypeUtils.APPLICATION_OCTET_STREAM.equals(contentType)) {
-					messageValues = deserializePayloadIfNecessary(requestMessage);
-				}
-				else {
-					return requestMessage;
-				}
+				return requestMessage;
 			}
-			return messageValues.toMessage();
 		}
 
 		@Override

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java

@@ -23,6 +23,7 @@ import org.springframework.messaging.MessageHeaders;
  * Spring Integration message headers for Spring Cloud Stream.
  * @author Gary Russell
  * @author David Turanski
+ * @author Soby Chacko
  */
 public final class BinderHeaders {
 
@@ -67,6 +68,13 @@ public final class BinderHeaders {
 	 */
 	public static final String NATIVE_HEADERS_PRESENT = PREFIX + "nativeHeadersPresent";
 
+	/**
+	 * Indicates the Spring Cloud Stream version.
+	 * Used for determining if legacy content type check supported or not.
+	 * @since 2.0
+	 */
+	public static final String SCST_VERSION = PREFIX + "version";
+
 	private BinderHeaders() {
 		super();
 	}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageSerializationUtils.java

@@ -16,13 +16,8 @@
 
 package org.springframework.cloud.stream.binder;
 
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
-import org.springframework.messaging.converter.ContentTypeResolver;
-import org.springframework.util.MimeType;
 
 /**
  * Utility class for serializing and de-serializing the message payload.
@@ -32,8 +27,6 @@ import org.springframework.util.MimeType;
  */
 public abstract class MessageSerializationUtils {
 
-	private static final Map<String, Class<?>> payloadTypeCache = new ConcurrentHashMap<>();
-
 	/**
 	 * Serialize the message payload unless it is a byte array.
 	 *
@@ -49,25 +42,4 @@ public abstract class MessageSerializationUtils {
 		return messageValues;
 	}
 
-
-
-	/**
-	 * De-serialize the message payload if necessary.
-	 *
-	 * @param messageValues message with the payload to deserialize
-	 * @param contentTypeResolver used for resolving the mime type.
-	 * @return Deserialized Message.
-	 */
-	public static MessageValues deserializePayload(MessageValues messageValues, ContentTypeResolver contentTypeResolver) {
-		Object payload = messageValues.getPayload();
-		MimeType contentType = contentTypeResolver.resolve(new MessageHeaders(messageValues.getHeaders()));
-		if (payload != null) {
-			messageValues.setPayload(payload);
-			messageValues.put(MessageHeaders.CONTENT_TYPE, contentType);
-		}
-		return messageValues;
-	}
-
-
-
 }

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

@@ -121,7 +121,7 @@ public class MessageConverterConfigurer
 					getPartitionKeyExtractorStrategy(producerProperties),
 					getPartitionSelectorStrategy(producerProperties)));
 		}
-		if (input && bindingProperties.isLegacyContentTypeHeaderEnabled()) {
+		if (input) {
 			messageChannel.addInterceptor(new LegacyContentTypeHeaderInterceptor());
 		}
 		// TODO: Set all interceptors in the correct order for input/output channels
@@ -306,11 +306,12 @@ public class MessageConverterConfigurer
 
 		@Override
 		public Message<?> preSend(Message<?> message, MessageChannel channel) {
-			Object originalContentType = message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE);
-			if (originalContentType != null) {
-				return MessageConverterConfigurer.this.messageBuilderFactory
+			if (!message.getHeaders().containsKey(BinderHeaders.SCST_VERSION) ||
+					!message.getHeaders().get(BinderHeaders.SCST_VERSION).equals("2.x")) {
+				Object originalContentType = message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE);
+				return originalContentType != null ? MessageConverterConfigurer.this.messageBuilderFactory
 						.fromMessage(message)
-						.setHeader(MessageHeaders.CONTENT_TYPE, originalContentType).build();
+						.setHeader(MessageHeaders.CONTENT_TYPE, originalContentType).build() : message;
 			}
 			return message;
 		}

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

@@ -58,8 +58,6 @@ public class BindingProperties {
 
 	private String contentType = MimeTypeUtils.APPLICATION_JSON_VALUE;
 
-	private boolean legacyContentTypeHeaderEnabled = false;
-
 	private String binder;
 
 	private ConsumerProperties consumer;
@@ -90,14 +88,6 @@ public class BindingProperties {
 		this.contentType = contentType;
 	}
 
-	public boolean isLegacyContentTypeHeaderEnabled() {
-		return legacyContentTypeHeaderEnabled;
-	}
-
-	public void setLegacyContentTypeHeaderEnabled(boolean legacyContentTypeHeaderEnabled) {
-		this.legacyContentTypeHeaderEnabled = legacyContentTypeHeaderEnabled;
-	}
-
 	public String getBinder() {
 		return binder;
 	}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java

@@ -31,6 +31,7 @@ import org.springframework.messaging.converter.AbstractMessageConverter;
 import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.MimeType;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -103,7 +104,6 @@ public class MessageConverterConfigurerTests {
 		BindingServiceProperties props = new BindingServiceProperties();
 		BindingProperties bindingProps = new BindingProperties();
 		bindingProps.setContentType("foo/bar");
-		bindingProps.setLegacyContentTypeHeaderEnabled(true);
 		props.setBindings(Collections.singletonMap("foo", bindingProps));
 		CompositeMessageConverterFactory converterFactory = new CompositeMessageConverterFactory(
 				Collections.<MessageConverter>emptyList(), null);
@@ -111,8 +111,11 @@ public class MessageConverterConfigurerTests {
 		QueueChannel in = new QueueChannel();
 		configurer.configureInputChannel(in, "foo");
 		Foo foo = new Foo();
-		in.send(new GenericMessage<>(foo,
-				Collections.singletonMap(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, "application/json")));
+		in.send(
+				MessageBuilder.withPayload(foo)
+					.setHeader(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, "application/json")
+					.setHeader(BinderHeaders.SCST_VERSION, "1.x")
+					.build());
 		Message<?> received = in.receive(0);
 		assertThat(received).isNotNull();
 		assertThat(received.getPayload()).isEqualTo(foo);