diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomMessageConverterTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomMessageConverterTests.java new file mode 100644 index 000000000..fbb6d49b8 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomMessageConverterTests.java @@ -0,0 +1,164 @@ +package org.springframework.cloud.stream.config; + +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.isA; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.hasItem; +import static org.junit.Assert.assertTrue; + +import java.util.List; + +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.converter.AbstractFromMessageConverter; +import org.springframework.cloud.stream.converter.MessageConverterUtils; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.cloud.stream.test.binder.TestSupportBinder; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +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 org.springframework.util.MimeType; + +/** + * @author Ilayaperumal Gopinathan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(CustomMessageConverterTests.TestSource.class) +public class CustomMessageConverterTests { + + @Autowired @Bindings(TestSource.class) + private Source testSource; + + @Autowired + private BinderFactory binderFactory; + + @Autowired + private List customMessageConverters; + + @Test + public void testCustomMessageConverter() throws Exception { + assertTrue(customMessageConverters.size() == 2); + assertThat(customMessageConverters, hasItem(isA(FooConverter.class))); + assertThat(customMessageConverters, hasItem(isA(BarConverter.class))); + testSource.output().send(MessageBuilder.withPayload(new Foo("hi")).build()); + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null)) + .messageCollector().forChannel(testSource.output()).poll(); + assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString(), + equalTo("application/x-java-object;type=org.springframework.cloud.stream.config.CustomMessageConverterTests$Bar")); + } + + @EnableBinding(Source.class) + @EnableAutoConfiguration + @PropertySource("classpath:/org/springframework/cloud/stream/config/custom/source-channel-configurers.properties") + @Configuration + public static class TestSource { + + @Bean + public AbstractFromMessageConverter fooConverter() { + return new FooConverter(); + } + + @Bean + public AbstractFromMessageConverter barConverter() { + return new BarConverter(); + } + } + + public static class FooConverter extends AbstractFromMessageConverter { + + public FooConverter() { + super(MimeType.valueOf("foo/test")); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] {Bar.class}; + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] {Foo.class}; + } + + @Override + public Object convertFromInternal(Message message, Class targetClass, Object conversionHint) { + Object result = null; + try { + if (message.getPayload() instanceof Foo) { + Foo fooPayload = (Foo) message.getPayload(); + result = new Bar(fooPayload.test); + } + } + catch (Exception e) { + logger.error(e.getMessage(), e); + return null; + } + return buildConvertedMessage(result, message.getHeaders(), + MessageConverterUtils.javaObjectMimeType(targetClass)); + } + } + + public static class BarConverter extends AbstractFromMessageConverter { + + public BarConverter() { + super(MimeType.valueOf("bar/test")); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] {Foo.class}; + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] {Bar.class}; + } + + @Override + public Object convertFromInternal(Message message, Class targetClass, Object conversionHint) { + Object result = null; + try { + if (message.getPayload() instanceof Bar) { + Bar barPayload = (Bar) message.getPayload(); + result = new Foo(barPayload.testing); + } + } + catch (Exception e) { + logger.error(e.getMessage(), e); + return null; + } + return buildConvertedMessage(result, message.getHeaders(), + MessageConverterUtils.javaObjectMimeType(targetClass)); + } + } + + public static class Foo { + + final String test; + + public Foo(String test) { + this.test = test; + } + + } + + public static class Bar { + + final String testing; + + public Bar(String testing) { + this.testing = testing; + } + } +} 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 afc525a66..c70090c8a 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 @@ -101,4 +101,3 @@ public class MessageChannelConfigurerTests { } } - diff --git a/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/custom/source-channel-configurers.properties b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/custom/source-channel-configurers.properties new file mode 100644 index 000000000..08e3f3224 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/custom/source-channel-configurers.properties @@ -0,0 +1,2 @@ +spring.cloud.stream.bindings.output.destination=configure1 +spring.cloud.stream.bindings.output.contentType=foo/test 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 ff3081489..3c204da67 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -15,6 +15,8 @@ */ package org.springframework.cloud.stream.binding; +import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.Set; @@ -41,10 +43,12 @@ import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.converter.MessageConverter; 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. * * @author Ilayaperumal Gopinathan */ @@ -56,8 +60,12 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea private final ChannelBindingServiceProperties channelBindingServiceProperties; - public MessageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties) { + private final Collection customMessageConverters; + + public MessageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties, + Collection customMessageConverters) { this.channelBindingServiceProperties = channelBindingServiceProperties; + this.customMessageConverters = customMessageConverters; } @Override @@ -69,6 +77,9 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea public void afterPropertiesSet() throws Exception { Assert.notNull(this.beanFactory, "Bean factory cannot be empty"); Set messageConverters = new HashSet<>(); + if (!CollectionUtils.isEmpty(customMessageConverters)) { + messageConverters.addAll(Collections.unmodifiableCollection(customMessageConverters)); + } messageConverters.add(new JsonToTupleMessageConverter()); messageConverters.add(new TupleToJsonMessageConverter()); messageConverters.add(new JsonToPojoMessageConverter()); @@ -97,9 +108,8 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea if (StringUtils.hasText(contentType)) { MimeType mimeType = MessageConverterUtils.getMimeType(contentType); MessageConverter messageConverter = this.messageConverterFactory.newInstance(mimeType); - Class dataType = MessageConverterUtils.getJavaTypeForContentType(mimeType, - Thread.currentThread().getContextClassLoader()); - messageChannel.setDatatypes(dataType); + Class[] supportedDataTypes = this.messageConverterFactory.supportedDataTypes(mimeType); + messageChannel.setDatatypes(supportedDataTypes); messageChannel.setMessageConverter(messageConverter); } } 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 75491ac82..04d4c807e 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 @@ -45,6 +45,7 @@ import org.springframework.cloud.stream.binding.MessageConverterConfigurer; import org.springframework.cloud.stream.binding.MessageHistoryTrackerConfigurer; import org.springframework.cloud.stream.binding.OutputBindingLifecycle; import org.springframework.cloud.stream.binding.SingleChannelBindable; +import org.springframework.cloud.stream.converter.AbstractFromMessageConverter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.DependsOn; @@ -77,6 +78,12 @@ public class ChannelBindingServiceConfiguration { @Autowired MessageBuilderFactory messageBuilderFactory; + /** + * User defined custom message converters + */ + @Autowired(required = false) + private List customMessageConverters; + @Bean // This conditional is intentionally not in an autoconfig (usually a bad idea) because // it is used to detect a ChannelBindingService in the parent context (which we know @@ -91,7 +98,7 @@ public class ChannelBindingServiceConfiguration { @Bean public MessageConverterConfigurer messageConverterConfigurer (ChannelBindingServiceProperties channelBindingServiceProperties) { - return new MessageConverterConfigurer(channelBindingServiceProperties); + return new MessageConverterConfigurer(channelBindingServiceProperties, customMessageConverters); } @Bean diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java index 64d3951ad..f966c6716 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -31,6 +31,7 @@ import org.springframework.util.MimeType; * A factory for creating an instance of {@link CompositeMessageConverter} for a given target MIME type * * @author David Turanski + * @author Ilayaperumal Gopinathan */ public class CompositeMessageConverterFactory { @@ -63,4 +64,28 @@ public class CompositeMessageConverterFactory { } return new CompositeMessageConverter(targetMimeTypeConverters); } + + public Class[] supportedDataTypes(MimeType targetMimeType) { + List> supportedDataTypes = new ArrayList<>(); + // Make sure to check if the target type is of explicit java object type. + if (MessageConverterUtils.X_JAVA_OBJECT.includes(targetMimeType)) { + supportedDataTypes.add(MessageConverterUtils.getJavaTypeForContentType(targetMimeType)); + } + else { + for (AbstractFromMessageConverter converter : converters) { + if (converter.supportsTargetMimeType(targetMimeType)) { + Class[] targetTypes = converter.supportedTargetTypes(); + if (targetTypes != null) { + Class[] dataTypes = converter.supportedTargetTypes(); + for (Class dataType : dataTypes) { + if (!supportedDataTypes.contains(dataType)) { + supportedDataTypes.add(dataType); + } + } + } + } + } + } + return supportedDataTypes.toArray(new Class[0]); + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java index 692ca40c3..ac5f5e93a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -16,11 +16,6 @@ package org.springframework.cloud.stream.converter; -import static org.springframework.util.MimeType.valueOf; -import static org.springframework.util.MimeTypeUtils.APPLICATION_JSON; -import static org.springframework.util.MimeTypeUtils.APPLICATION_OCTET_STREAM; - -import org.springframework.tuple.DefaultTuple; import org.springframework.tuple.Tuple; import org.springframework.util.ClassUtils; import org.springframework.util.MimeType; @@ -32,6 +27,7 @@ import org.springframework.util.StringUtils; * * @author David Turanski * @author Gary Russell + * @author Ilayaperumal Gopinathan */ public class MessageConverterUtils { @@ -51,42 +47,24 @@ public class MessageConverterUtils { public static final MimeType X_JAVA_SERIALIZED_OBJECT = MimeType.valueOf("application/x-java-serialized-object"); /** - * Map the contentType to a target class. + * Get the java Object type for the MimeType X_JAVA_OBJECT * - * @param contentType the content type - * @param classLoader the class loader used to resolve the class * @return the class for the content type */ - public static Class getJavaTypeForContentType(MimeType contentType, ClassLoader classLoader) { + public static Class getJavaTypeForContentType(MimeType contentType) { + Class javaType = Object.class; if (X_JAVA_OBJECT.includes(contentType)) { if (contentType.getParameter("type") != null) { try { - return ClassUtils.forName(contentType.getParameter("type"), classLoader); + javaType = ClassUtils.forName(contentType.getParameter("type"), + Thread.currentThread().getContextClassLoader()); } catch (Exception e) { throw new ConversionException(e.getMessage(), e); } } - else { - return Object.class; - } } - else if (APPLICATION_JSON.equals(contentType)) { - return String.class; - } - else if (valueOf("text/*").includes(contentType)) { - return String.class; - } - else if (X_SPRING_TUPLE.includes(contentType)) { - return DefaultTuple.class; - } - else if (APPLICATION_OCTET_STREAM.includes(contentType)) { - return byte[].class; - } - else if (X_JAVA_SERIALIZED_OBJECT.includes(contentType)) { - return byte[].class; - } - return null; + return javaType; } /**