From 53df5f5747be834ef2a26a6d46d0fcd5ff1ef418 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 10 Nov 2017 11:11:36 -0500 Subject: [PATCH] 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 --- .../config/MessageChannelConfigurerTests.java | 13 ++- ...MessageChannelWithNativeDecodingTests.java | 83 +++++++++++++++++++ ...MessageChannelWithNativeEncodingTests.java | 68 +++++++++++++++ .../config/contentType/ContentTypeTests.java | 33 +------- .../cloud/stream/config/contentType/User.java | 55 ++++++++++++ .../channel/native-decoding-sink.properties | 3 + .../channel/native-encoding-source.properties | 3 + .../stream/test/binder/TestSupportBinder.java | 12 ++- .../stream/binder/ConsumerProperties.java | 11 +++ .../binding/MessageConverterConfigurer.java | 20 +++-- 10 files changed, 249 insertions(+), 52 deletions(-) create mode 100644 spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeDecodingTests.java create mode 100644 spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeEncodingTests.java create mode 100644 spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/User.java create mode 100644 spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/native-decoding-sink.properties create mode 100644 spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/native-encoding-source.properties diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java index 99842618f..d05bc0cac 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java +++ b/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()); diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeDecodingTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeDecodingTests.java new file mode 100644 index 000000000..55d30e651 --- /dev/null +++ b/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 { + + } +} diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeEncodingTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeEncodingTests.java new file mode 100644 index 000000000..9aa43a3bb --- /dev/null +++ b/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 { + + } +} diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/ContentTypeTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/ContentTypeTests.java index f0e96783b..090c654b0 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/ContentTypeTests.java +++ b/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(); - } - } } diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/User.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/User.java new file mode 100644 index 000000000..bfadca1c3 --- /dev/null +++ b/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(); + } +} diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/native-decoding-sink.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/native-decoding-sink.properties new file mode 100644 index 000000000..86577d5ea --- /dev/null +++ b/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 diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/native-encoding-source.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/native-encoding-source.properties new file mode 100644 index 000000000..27d5bf0bd --- /dev/null +++ b/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 diff --git a/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java b/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java index a165e012c..d5823eb53 100644 --- a/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java +++ b/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 { @@ -71,7 +72,7 @@ public class TestSupportBinder implements Binder bindProducer(String name, MessageChannel outboundBindTarget, ProducerProperties properties) { - final BlockingQueue> queue = messageCollector.register(outboundBindTarget); + final BlockingQueue> 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>> results = new HashMap<>(); - private BlockingQueue> 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> 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> result = new LinkedBlockingDeque<>(); Assert.isTrue(!results.containsKey(channel), "Channel [" + channel + "] was already bound"); results.put(channel, result); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java index fb801ceef..50d3a314c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java +++ b/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; + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index 3cbfb4aef..c6e9ef506 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/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); }