From 9abf36879d31add7e773253ebd92548c1ee46f90 Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Fri, 4 Mar 2016 21:24:59 +0530 Subject: [PATCH] Set message header using BindingProperties' contentType - Update MessageConverterConfigurer that uses a channel interceptor to check if the message contentType header is missing and if there is a binding property for `contentType` header is provided, then set that value as the message contentType header This resolves #397 Add test --- .../config/ContentTypeConfigurerTests.java | 66 +++++++++++++++++++ .../config/MessageChannelConfigurerTests.java | 2 +- .../source-channel-configurers.properties | 2 + .../binding/MessageConverterConfigurer.java | 35 ++++++++-- .../ChannelBindingServiceConfiguration.java | 3 +- 5 files changed, 101 insertions(+), 7 deletions(-) create mode 100644 spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeConfigurerTests.java create mode 100644 spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeConfigurerTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeConfigurerTests.java new file mode 100644 index 000000000..d17d9717e --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeConfigurerTests.java @@ -0,0 +1,66 @@ +/* + * Copyright 2015 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 static org.hamcrest.CoreMatchers.equalTo; +import static org.junit.Assert.assertThat; + +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.SpringApplicationConfiguration; +import org.springframework.cloud.stream.annotation.Bindings; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.cloud.stream.test.binder.TestSupportBinder; +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; + +/** + * @author Ilayaperumal Gopinathan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration({ContentTypeConfigurerTests.TestSource.class}) +public class ContentTypeConfigurerTests { + + @Autowired @Bindings(TestSource.class) + private Source testSource; + + @Autowired + private BinderFactory binderFactory; + + @Test + public void testMessageHeaderWhenNoExplicitContentType() throws Exception { + testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}").build()); + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null)).messageCollector().forChannel(testSource.output()).poll(); + assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString(), equalTo("application/json")); + assertThat(received.getPayload(), equalTo("{\"message\":\"Hi\"}")); + } + + @EnableBinding(Source.class) + @EnableAutoConfiguration + @PropertySource("classpath:/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties") + public static class TestSource { + + } +} + 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 c70090c8a..11553c9c1 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 @@ -52,7 +52,7 @@ public class MessageChannelConfigurerTests { private Sink testSink; @Test - public void testContentTypeConfigurer() throws Exception { + public void testMessageConverterConfigurer() throws Exception { final CountDownLatch latch = new CountDownLatch(1); MessageHandler messageHandler = new MessageHandler() { @Override diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties new file mode 100644 index 000000000..1064e292c --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties @@ -0,0 +1,2 @@ +spring.cloud.stream.bindings.output.destination=interceptor-test +spring.cloud.stream.bindings.output.contentType=application/json 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 3c204da67..779dfeb11 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 @@ -40,15 +40,23 @@ import org.springframework.cloud.stream.converter.SerializedToJavaMessageConvert import org.springframework.cloud.stream.converter.StringToByteArrayMessageConverter; import org.springframework.cloud.stream.converter.TupleToJsonMessageConverter; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.ChannelInterceptorAware; +import org.springframework.integration.support.MessageBuilderFactory; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; import org.springframework.util.StringUtils; /** - * A {@link MessageChannelConfigurer} that sets the datatype and message converters for the message channel. + * A {@link MessageChannelConfigurer} that sets the datatypes and message converters based on the ContentType set via + * {@link BindingProperties}. This also adds a {@link org.springframework.messaging.support.ChannelInterceptor} to + * the message channel to set the `ContentType` header for the message (if not already set) based on the `ContentType` binding + * property of the channel. * * @author Ilayaperumal Gopinathan */ @@ -62,10 +70,14 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea private final Collection customMessageConverters; + private final MessageBuilderFactory messageBuilderFactory; + public MessageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties, - Collection customMessageConverters) { + Collection customMessageConverters, + MessageBuilderFactory messageBuilderFactory) { this.channelBindingServiceProperties = channelBindingServiceProperties; this.customMessageConverters = customMessageConverters; + this.messageBuilderFactory = messageBuilderFactory; } @Override @@ -103,14 +115,27 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea Assert.isAssignable(AbstractMessageChannel.class, channel.getClass()); AbstractMessageChannel messageChannel = (AbstractMessageChannel) channel; BindingProperties bindingProperties = this.channelBindingServiceProperties.getBindingProperties(channelName); - if (bindingProperties != null) { - String contentType = bindingProperties.getContentType(); - if (StringUtils.hasText(contentType)) { + final String contentType = bindingProperties.getContentType(); + if (bindingProperties != null && StringUtils.hasText(contentType)) { MimeType mimeType = MessageConverterUtils.getMimeType(contentType); MessageConverter messageConverter = this.messageConverterFactory.newInstance(mimeType); Class[] supportedDataTypes = this.messageConverterFactory.supportedDataTypes(mimeType); messageChannel.setDatatypes(supportedDataTypes); messageChannel.setMessageConverter(messageConverter); + if (messageChannel instanceof ChannelInterceptorAware) { + ((ChannelInterceptorAware) messageChannel).addInterceptor(new ChannelInterceptorAdapter() { + @Override + public Message preSend(Message message, MessageChannel messageChannel) { + Object contentTypeFromMessage = message.getHeaders().get(MessageHeaders.CONTENT_TYPE); + if (contentTypeFromMessage == null) { + return messageBuilderFactory + .fromMessage(message) + .setHeader(MessageHeaders.CONTENT_TYPE, contentType) + .build(); + } + return message; + } + }); } } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java index 04d4c807e..7d4b5912b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java @@ -98,7 +98,8 @@ public class ChannelBindingServiceConfiguration { @Bean public MessageConverterConfigurer messageConverterConfigurer (ChannelBindingServiceProperties channelBindingServiceProperties) { - return new MessageConverterConfigurer(channelBindingServiceProperties, customMessageConverters); + return new MessageConverterConfigurer(channelBindingServiceProperties, customMessageConverters, + messageBuilderFactory); } @Bean