From 08504952e69ae8f286ab1a54c10208d15abdff88 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 6 Oct 2017 18:49:51 -0400 Subject: [PATCH] Default interceptor for legacy contentType Fixes #1057 Enable consumers to convert from originalContentType to contentType in the case of legacy producers sending messages with originalContentType header. By default, this conversion will not happen and only be acitvated by setting `spring.cloud.stream.bindings.input.legacyContentTypeHeaderEnabled` to true. --- .../stream/config/LegacyContentTypeTests.java | 75 +++++++++++++++++++ ...legacy-sink-channel-configurers.properties | 4 + .../binding/MessageConverterConfigurer.java | 19 +++++ .../stream/config/BindingProperties.java | 11 +++ .../MessageConverterConfigurerTests.java | 22 ++++++ 5 files changed, 131 insertions(+) create mode 100644 spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/LegacyContentTypeTests.java create mode 100644 spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/legacy-sink-channel-configurers.properties 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";