Native endoding/decoding changes for interceptors

Fixes #1109

- Disable outbound content type interceptor when nativeEndoding is enabled on the producer
- Disable inbound content type interceptor when nativeDecoding is enabled on the consumer

- Add a new flag in ConsumerProperties for useNativeDecoding

- Apply ContentType interceptor before the InboundInterceptor to ensure
  that content type set on the channel is applied on the message if content type
  is missing on the Message.

- Add tests for verifying interceptors are not applied when native encoding/decoding is enabled
- Fix a kryo integration test
- Remove Inbound interceptor from MessageCollector when native encoding is enabled on the producer
- Polishing

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

@@ -38,7 +38,6 @@ import org.springframework.context.annotation.PropertySource;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
-import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -50,6 +49,7 @@ import static org.junit.Assert.assertNull;
 /**
  * @author Ilayaperumal Gopinathan
  * @author Gary Russell
+ * @author Soby Chacko
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @SpringBootTest(classes = { MessageChannelConfigurerTests.TestSink.class,
@@ -74,13 +74,10 @@ public class MessageChannelConfigurerTests {
 	@Test
 	public void testMessageConverterConfigurer() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
-		MessageHandler messageHandler = new MessageHandler() {
-			@Override
-			public void handleMessage(Message<?> message) throws MessagingException {
-				assertThat(message.getPayload()).isInstanceOf(byte[].class);
-				assertThat(message.getPayload()).isEqualTo("{\"message\":\"Hi\"}".getBytes());
-				latch.countDown();
-			}
+		MessageHandler messageHandler = message -> {
+			assertThat(message.getPayload()).isInstanceOf(byte[].class);
+			assertThat(message.getPayload()).isEqualTo("{\"message\":\"Hi\"}".getBytes());
+			latch.countDown();
 		};
 		testSink.input().subscribe(messageHandler);
 		testSink.input().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}".getBytes()).build());

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

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2015-2017 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 java.io.ByteArrayOutputStream;
+import java.io.ObjectOutput;
+import java.io.ObjectOutputStream;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+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.context.SpringBootTest;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Soby Chacko
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = { MessageChannelWithNativeDecodingTests.NativeDecodingSink.class})
+public class MessageChannelWithNativeDecodingTests {
+
+	@Autowired
+	private Sink nativeDecodingSink;
+
+	@Test
+	public void testMessageConverterInterceptorsAreSkippedWhenNativeDecodingIsEnabled() throws Exception {
+		final CountDownLatch latch = new CountDownLatch(1);
+
+		byte[] serializedData;
+		ObjectOutput out;
+		try(ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
+			out = new ObjectOutputStream(bos);
+			out.writeObject(123);
+			out.flush();
+			serializedData = bos.toByteArray();
+		}
+
+		MessageHandler messageHandler = message -> {
+			//ensure that the data is not deserialized becasue of native decoding
+			//and the content type set in the properties file didn't take any effect
+			assertThat(message.getPayload()).isInstanceOf(byte[].class);
+			assertThat(message.getPayload()).isEqualTo(serializedData);
+			latch.countDown();
+		};
+		nativeDecodingSink.input().subscribe(messageHandler);
+
+		nativeDecodingSink.input().send(MessageBuilder.withPayload(serializedData).build());
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		nativeDecodingSink.input().unsubscribe(messageHandler);
+	}
+
+	@EnableBinding(Sink.class)
+	@EnableAutoConfiguration
+	@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/native-decoding-sink.properties")
+	public static class NativeDecodingSink {
+
+	}
+}

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

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2015-2017 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 java.util.concurrent.TimeUnit;
+
+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.context.SpringBootTest;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.test.binder.MessageCollector;
+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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Soby Chacko
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = { MessageChannelWithNativeEncodingTests.NativeEncodingSource.class})
+public class MessageChannelWithNativeEncodingTests {
+
+	@Autowired
+	private Source nativeEncodingSource;
+
+	@Autowired
+	private MessageCollector messageCollector;
+
+	@Test
+	public void testOutboundContentTypeInterceptorIsSkippedWhenNativeEncodingIsEnabled() throws Exception {
+		this.nativeEncodingSource.output().send(MessageBuilder.withPayload("hello foobar!").build());
+		Message<?> message = this.messageCollector.forChannel(this.nativeEncodingSource.output()).poll(1, TimeUnit.SECONDS);
+		//should not convert the payload to byte[] even though we set a contentType on the channel.
+		//This is becasue, we are using native encoding.
+		assertThat(message.getPayload()).isInstanceOf(String.class);
+		assertThat(message.getPayload()).isEqualTo("hello foobar!");
+		assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isNull();
+	}
+
+	@EnableBinding(Source.class)
+	@EnableAutoConfiguration
+	@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/native-encoding-source.properties")
+	public static class NativeEncodingSource {
+
+	}
+}

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

@@ -18,17 +18,13 @@ package org.springframework.cloud.stream.config.contentType;
 
 import java.io.ByteArrayOutputStream;
 import java.io.ObjectOutputStream;
-import java.io.Serializable;
 import java.util.LinkedList;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.io.Output;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
-
 import org.junit.Test;
 
 import org.springframework.boot.SpringApplication;
@@ -297,7 +293,7 @@ public class ContentTypeTests {
 		try (ConfigurableApplicationContext context = SpringApplication.run(
 				SinkApplication.class, "--server.port=0",
 				"--spring.jmx.enabled=false",
-				"--spring.cloud.stream.bindings.pojo_input.contentType=application/x-java-object"
+				"--spring.cloud.stream.bindings.pojo_input.contentType=application/x-java-object;type=org.springframework.cloud.stream.config.contentType.User"
 				)) {
 			TestSink testSink = context.getBean(TestSink.class);
 			SinkApplication sourceApp = context.getBean(SinkApplication.class);
@@ -412,33 +408,6 @@ public class ContentTypeTests {
 
 	}
 
-	@SuppressWarnings("serial")
-	public static class User implements Serializable {
 
-		private String name;
-
-		public User(){}
-
-		@JsonCreator
-		public User(@JsonProperty("name") String name) {
-			this.name = name;
-		}
-
-		public String getName() {
-			return name;
-		}
-
-		public void setName(String name) {
-			this.name = name;
-		}
-
-		@Override
-		public String toString() {
-			final StringBuffer sb = new StringBuffer("User{");
-			sb.append("name='").append(name).append('\'');
-			sb.append('}');
-			return sb.toString();
-		}
-	}
 
 }

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2017 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.contentType;
+
+import java.io.Serializable;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * @author Vinicius Carvalho
+ * @author Oleg Zhurakousky
+ */
+@SuppressWarnings("serial")
+public class User implements Serializable {
+
+	private String name;
+
+	public User(){}
+
+	@JsonCreator
+	public User(@JsonProperty("name") String name) {
+		this.name = name;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	@Override
+	public String toString() {
+		final StringBuffer sb = new StringBuffer("User{");
+		sb.append("name='").append(name).append('\'');
+		sb.append('}');
+		return sb.toString();
+	}
+}

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

@@ -0,0 +1,3 @@
+spring.cloud.stream.bindings.input.destination=foobar
+spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true
+spring.cloud.stream.bindings.input.contentType=application/x-java-serialized-object

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

@@ -0,0 +1,3 @@
+spring.cloud.stream.bindings.output.destination=foobar-x
+spring.cloud.stream.bindings.output.producer.useNativeEncoding=true
+spring.cloud.stream.bindings.output.contentType=application/x-java-serialized-object

spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java

@@ -50,6 +50,7 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Soby Chacko
  * @see MessageQueueMatcher
  */
 public class TestSupportBinder implements Binder<MessageChannel, ConsumerProperties, ProducerProperties> {
@@ -71,7 +72,7 @@ public class TestSupportBinder implements Binder<MessageChannel, ConsumerPropert
 	@Override
 	public Binding<MessageChannel> bindProducer(String name, MessageChannel outboundBindTarget,
 			ProducerProperties properties) {
-		final BlockingQueue<Message<?>> queue = messageCollector.register(outboundBindTarget);
+		final BlockingQueue<Message<?>> queue = messageCollector.register(outboundBindTarget, properties.isUseNativeEncoding());
 		((SubscribableChannel) outboundBindTarget).subscribe(new MessageHandler() {
 			@Override
 			public void handleMessage(Message<?> message) throws MessagingException {
@@ -99,9 +100,12 @@ public class TestSupportBinder implements Binder<MessageChannel, ConsumerPropert
 
 		private final Map<MessageChannel, BlockingQueue<Message<?>>> results = new HashMap<>();
 
-		private BlockingQueue<Message<?>> register(MessageChannel channel) {
-			// we need to add this intercepter to ensure MessageCollector's compatibility with previous versions of SCSt
-			((AbstractMessageChannel)channel).addInterceptor(new MessageConverterConfigurer.InboundMessageConvertingInterceptor());
+		private BlockingQueue<Message<?>> register(MessageChannel channel, boolean useNativeEncoding) {
+			// we need to add this interceptor to ensure MessageCollector's compatibility with
+			// previous versions of SCSt when native encoding is disabled.
+			if (!useNativeEncoding) {
+				((AbstractMessageChannel) channel).addInterceptor(new MessageConverterConfigurer.InboundMessageConvertingInterceptor());
+			}
 			LinkedBlockingDeque<Message<?>> result = new LinkedBlockingDeque<>();
 			Assert.isTrue(!results.containsKey(channel), "Channel [" + channel + "] was already bound");
 			results.put(channel, result);

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

@@ -26,6 +26,7 @@ import com.fasterxml.jackson.annotation.JsonInclude;
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
  * @author Gary Russell
+ * @author Soby Chacko
  */
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public class ConsumerProperties {
@@ -48,6 +49,8 @@ public class ConsumerProperties {
 
 	private HeaderMode headerMode;
 
+	private boolean useNativeDecoding = false;
+
 	@Min(value = 1, message = "Concurrency should be greater than zero.")
 	public int getConcurrency() {
 		return concurrency;
@@ -126,4 +129,12 @@ public class ConsumerProperties {
 	public void setHeaderMode(HeaderMode headerMode) {
 		this.headerMode = headerMode;
 	}
+
+	public boolean isUseNativeDecoding() {
+		return useNativeDecoding;
+	}
+
+	public void setUseNativeDecoding(boolean useNativeDecoding) {
+		this.useNativeDecoding = useNativeDecoding;
+	}
 }

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

@@ -25,6 +25,7 @@ import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.cloud.stream.binder.BinderException;
 import org.springframework.cloud.stream.binder.BinderHeaders;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.PartitionHandler;
 import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
 import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
@@ -125,13 +126,16 @@ public class MessageConverterConfigurer
 					getPartitionKeyExtractorStrategy(producerProperties),
 					getPartitionSelectorStrategy(producerProperties)));
 		}
-		if (input) {
-			messageChannel.addInterceptor(new InboundMessageConvertingInterceptor());
-		}
-		// TODO: Set all interceptors in the correct order for input/output channels
+		ConsumerProperties consumerProperties = bindingProperties.getConsumer();
 		if (StringUtils.hasText(contentType)) {
-			messageChannel.addInterceptor(
-					new ContentTypeConvertingInterceptor(contentType, input));
+			if ((!input && (producerProperties == null || !producerProperties.isUseNativeEncoding())) ||
+					(input && (consumerProperties == null || !consumerProperties.isUseNativeDecoding()))) {
+				messageChannel.addInterceptor(
+						new ContentTypeConvertingInterceptor(contentType, input));
+			}
+		}
+		if (input && (consumerProperties == null || !consumerProperties.isUseNativeDecoding())) {
+			messageChannel.addInterceptor(new InboundMessageConvertingInterceptor());
 		}
 	}
 
@@ -331,8 +335,8 @@ public class MessageConverterConfigurer
 			Object payload;
 			if (converter != null){
 				Assert.isTrue(!(equalTypeAndSubType(MessageConverterUtils.X_JAVA_OBJECT, contentType) && targetClass == null),
-						"Can not deserialize into message since 'contentType` has not "
-							+ "being encoded with the actual target type."
+						"Cannot deserialize into message since 'contentType` is not "
+							+ "encoded with the actual target type."
 							+ "Consider 'application/x-java-object; type=foo.bar.MyClass'");
 				payload = converter.fromMessage(message, targetClass);
 			}