diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/LegacyContentTypeTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/LegacyContentTypeTests.java new file mode 100644 index 000000000..0209ec9ae --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/LegacyContentTypeTests.java @@ -0,0 +1,75 @@ +/* + * 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.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.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; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Soby Chacko + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringBootTest(classes = { LegacyContentTypeTests.LegacyTestSink.class}) +public class LegacyContentTypeTests { + + @Autowired + private Sink testSink; + + @Test + public void testOriginalContentTypeIsRetrievedForLegacyContentHeaderType() 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()); + assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo("application/json"); + latch.countDown(); + } + }; + testSink.input().subscribe(messageHandler); + testSink.input().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}".getBytes()).setHeader(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, "application/json").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 { + + } +} diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/legacy-sink-channel-configurers.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/legacy-sink-channel-configurers.properties new file mode 100644 index 000000000..78c379dd0 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/legacy-sink-channel-configurers.properties @@ -0,0 +1,4 @@ +spring.cloud.stream.bindings.input.destination=configure1 +spring.cloud.stream.bindings.input.legacyContentTypeHeaderEnabled=true +spring.cloud.stream.bindings.input.contentType=application/x-spring-tuple + 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 abde5094b..418a4951c 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 @@ -60,6 +60,7 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author Maxim Kirilov * @author Gary Russell + * @author Soby Chacko */ public class MessageConverterConfigurer implements MessageChannelConfigurer, BeanFactoryAware, InitializingBean { @@ -120,6 +121,9 @@ public class MessageConverterConfigurer getPartitionKeyExtractorStrategy(producerProperties), getPartitionSelectorStrategy(producerProperties))); } + if (input && bindingProperties.isLegacyContentTypeHeaderEnabled()) { + messageChannel.addInterceptor(new LegacyContentTypeHeaderInterceptor()); + } // TODO: Set all interceptors in the correct order for input/output channels if (StringUtils.hasText(contentType)) { messageChannel.addInterceptor( @@ -298,4 +302,19 @@ public class MessageConverterConfigurer } } + private final class LegacyContentTypeHeaderInterceptor extends ChannelInterceptorAdapter { + + @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 + .fromMessage(message) + .setHeader(MessageHeaders.CONTENT_TYPE, originalContentType).build(); + } + return message; + } + + } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index 72924de66..676de5d54 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -31,6 +31,7 @@ import org.springframework.validation.annotation.Validated; * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Gary Russell + * @author Soby Chacko */ @JsonInclude(Include.NON_DEFAULT) @Validated @@ -57,6 +58,8 @@ public class BindingProperties { private String contentType = MimeTypeUtils.APPLICATION_JSON_VALUE; + private boolean legacyContentTypeHeaderEnabled = false; + private String binder; private ConsumerProperties consumer; @@ -87,6 +90,14 @@ public class BindingProperties { this.contentType = contentType; } + public boolean isLegacyContentTypeHeaderEnabled() { + return legacyContentTypeHeaderEnabled; + } + + public void setLegacyContentTypeHeaderEnabled(boolean legacyContentTypeHeaderEnabled) { + this.legacyContentTypeHeaderEnabled = legacyContentTypeHeaderEnabled; + } + public String getBinder() { return binder; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java index dccbdb99d..386df1582 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java @@ -20,6 +20,7 @@ import java.util.Collections; import org.junit.Test; +import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; @@ -97,6 +98,27 @@ public class MessageConverterConfigurerTests { } } + @Test + public void testConfigureInputChannelWithLegacyContentType() { + 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.emptyList(), null); + MessageConverterConfigurer configurer = new MessageConverterConfigurer(props, converterFactory); + 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"))); + Message received = in.receive(0); + assertThat(received).isNotNull(); + assertThat(received.getPayload()).isEqualTo(foo); + assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo("application/json"); + } + public static class Foo { private String bar = "bar";