From 4d2ef3c4d0c7102a1dfdb8a1e1bb22064adb5f32 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Wed, 9 Mar 2016 01:45:00 -0500 Subject: [PATCH] Introduce StreamListener annotation Fixes #156 - Supports Spring Messaging infrastructure for argument mapping and type conversion, sharing the same configured converters as the ones configured on input/output channels; - Also: refactored AbstractFromMessageConverter to return only payload - conversion to a message enriched with the content type of the channel is done by a separate wrapping converter; --- .../stream/config/BindingListenerTests.java | 380 ++++++++++++++++++ .../config/CustomMessageConverterTests.java | 29 +- .../source-channel-configurers.properties | 2 +- .../stream/annotation/StreamListener.java | 63 +++ .../cloud/stream/binder/AbstractBinder.java | 2 +- .../binding/MessageConverterConfigurer.java | 131 +++--- ...amListenerAnnotationBeanPostProcessor.java | 204 ++++++++++ .../ChannelBindingServiceConfiguration.java | 43 +- .../AbstractFromMessageConverter.java | 35 +- .../ByteArrayToStringMessageConverter.java | 1 - .../CompositeMessageConverterFactory.java | 40 +- .../JavaToSerializedMessageConverter.java | 4 +- .../converter/JsonToPojoMessageConverter.java | 8 +- .../JsonToTupleMessageConverter.java | 3 +- .../converter/PojoToJsonMessageConverter.java | 10 +- .../PojoToStringMessageConverter.java | 2 +- .../SerializedToJavaMessageConverter.java | 3 +- .../TupleToJsonMessageConverter.java | 8 +- .../BinderAwareChannelResolverTests.java | 3 +- .../binding/ChannelBindingServiceTests.java | 3 +- 20 files changed, 840 insertions(+), 134 deletions(-) create mode 100644 spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/BindingListenerTests.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamListener.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/BindingListenerTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/BindingListenerTests.java new file mode 100644 index 000000000..9fb58a933 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/BindingListenerTests.java @@ -0,0 +1,380 @@ +/* + * Copyright 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. + * 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.Matchers.equalTo; +import static org.hamcrest.Matchers.hasProperty; +import static org.hamcrest.Matchers.hasSize; +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.not; +import static org.hamcrest.Matchers.nullValue; +import static org.hamcrest.Matchers.startsWith; +import static org.hamcrest.collection.IsMapContaining.hasEntry; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.cloud.stream.test.binder.MessageCollector; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.handler.annotation.Header; +import org.springframework.messaging.handler.annotation.Headers; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.handler.annotation.SendTo; + +/** + * @author Marius Bogoevici + */ +public class BindingListenerTests { + + @Test + public void testContentTypeConversion() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestSink.class); + @SuppressWarnings("unchecked") + TestSink testSink = context.getBean(TestSink.class); + Sink sink = context.getBean(Sink.class); + String id = UUID.randomUUID().toString(); + sink.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").build()); + assertTrue(testSink.latch.await(10, TimeUnit.SECONDS)); + assertThat(testSink.receivedArguments, hasSize(1)); + assertThat(testSink.receivedArguments.get(0), hasProperty("bar", equalTo("barbar" + id))); + context.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void testAnnotatedArguments() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithAnnotatedArguments.class); + + TestPojoWithAnnotatedArguments testPojoWithAnnotatedArguments = context.getBean(TestPojoWithAnnotatedArguments.class); + Sink sink = context.getBean(Sink.class); + String id = UUID.randomUUID().toString(); + sink.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").setHeader("testHeader", "testValue").build()); + assertThat(testPojoWithAnnotatedArguments.receivedArguments, hasSize(3)); + assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(0), instanceOf(FooPojo.class)); + assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(0), hasProperty("bar", equalTo("barbar" + id))); + assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(1), instanceOf(Map.class)); + assertThat((Map) testPojoWithAnnotatedArguments.receivedArguments.get(1), + hasEntry(MessageHeaders.CONTENT_TYPE, "application/json")); + assertThat((Map) testPojoWithAnnotatedArguments.receivedArguments.get(1), + hasEntry(equalTo("testHeader"), equalTo("testValue"))); + assertThat((String) testPojoWithAnnotatedArguments.receivedArguments.get(2), equalTo("application/json")); + context.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void testReturn() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestStringProcessor.class); + MessageCollector collector = context.getBean(MessageCollector.class); + Processor processor = context.getBean(Processor.class); + String id = UUID.randomUUID().toString(); + processor.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").build()); + Message message = (Message) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS); + TestStringProcessor testStringProcessor = context.getBean(TestStringProcessor.class); + assertThat(testStringProcessor.receivedPojos, hasSize(1)); + assertThat(testStringProcessor.receivedPojos.get(0), hasProperty("bar", equalTo("barbar" + id))); + assertThat(message, not(nullValue(Message.class))); + assertThat(message.getPayload(), equalTo("barbar" + id)); + context.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void testReturnConversion() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithMimeType.class, + "--spring.cloud.stream.bindings.output.contentType=application/json"); + MessageCollector collector = context.getBean(MessageCollector.class); + Processor processor = context.getBean(Processor.class); + String id = UUID.randomUUID().toString(); + processor.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").build()); + TestPojoWithMimeType testPojoWithMimeType = context.getBean(TestPojoWithMimeType.class); + assertThat(testPojoWithMimeType.receivedPojos, hasSize(1)); + assertThat(testPojoWithMimeType.receivedPojos.get(0), hasProperty("bar", equalTo("barbar" + id))); + Message message = (Message) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS); + assertThat(message, not(nullValue(Message.class))); + assertThat(message.getPayload(), equalTo("{\"qux\":\"barbar" + id + "\"}")); + assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, String.class), equalTo("application/json")); + context.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void testReturnNoConversion() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithMimeType.class); + MessageCollector collector = context.getBean(MessageCollector.class); + Processor processor = context.getBean(Processor.class); + String id = UUID.randomUUID().toString(); + processor.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").build()); + TestPojoWithMimeType testPojoWithMimeType = context.getBean(TestPojoWithMimeType.class); + assertThat(testPojoWithMimeType.receivedPojos, hasSize(1)); + assertThat(testPojoWithMimeType.receivedPojos.get(0), hasProperty("bar", equalTo("barbar" + id))); + Message message = (Message) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS); + assertThat(message, not(nullValue(Message.class))); + assertThat(message.getPayload().getQux(), equalTo("barbar" + id)); + context.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void testReturnMessage() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithMessageReturn.class); + MessageCollector collector = context.getBean(MessageCollector.class); + Processor processor = context.getBean(Processor.class); + String id = UUID.randomUUID().toString(); + processor.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").build()); + TestPojoWithMessageReturn testPojoWithMessageReturn = context.getBean(TestPojoWithMessageReturn.class); + assertThat(testPojoWithMessageReturn.receivedPojos, hasSize(1)); + assertThat(testPojoWithMessageReturn.receivedPojos.get(0), hasProperty("bar", equalTo("barbar" + id))); + Message message = (Message) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS); + assertThat(message, not(nullValue(Message.class))); + assertThat(message.getPayload().getQux(), equalTo("barbar" + id)); + context.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void testMessageArgument() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithMessageArgument.class); + MessageCollector collector = context.getBean(MessageCollector.class); + Processor processor = context.getBean(Processor.class); + String id = UUID.randomUUID().toString(); + processor.input().send(MessageBuilder.withPayload("barbar" + id) + .setHeader("contentType", "text/plain").build()); + TestPojoWithMessageArgument testPojoWithMessageArgument = context.getBean(TestPojoWithMessageArgument.class); + assertThat(testPojoWithMessageArgument.receivedMessages, hasSize(1)); + assertThat(testPojoWithMessageArgument.receivedMessages.get(0).getPayload(), equalTo("barbar" + id)); + Message message = (Message) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS); + assertThat(message, not(nullValue(Message.class))); + assertThat(message.getPayload().getQux(), equalTo("barbar" + id)); + context.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void testDuplicateMapping() throws Exception { + try { + ConfigurableApplicationContext context = SpringApplication.run(TestDuplicateMapping.class); + fail("Exception expected on duplicate mapping"); + } + catch (BeanCreationException e) { + assertThat(e.getCause().getMessage(), startsWith("Duplicate @StreamListener mapping")); + } + } + + + @Test + @SuppressWarnings("unchecked") + public void testHandlerBean() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestHandlerBean.class, + "--spring.cloud.stream.bindings.output.contentType=application/json"); + MessageCollector collector = context.getBean(MessageCollector.class); + Processor processor = context.getBean(Processor.class); + String id = UUID.randomUUID().toString(); + processor.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").build()); + HandlerBean handlerBean = context.getBean(HandlerBean.class); + assertThat(handlerBean.receivedPojos, hasSize(1)); + assertThat(handlerBean.receivedPojos.get(0), hasProperty("bar", equalTo("barbar" + id))); + Message message = (Message) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS); + assertThat(message, not(nullValue(Message.class))); + assertThat(message.getPayload(), equalTo("{\"qux\":\"barbar" + id + "\"}")); + assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, String.class), equalTo("application/json")); + context.close(); + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + public static class TestSink { + + List receivedArguments = new ArrayList<>(); + + CountDownLatch latch = new CountDownLatch(1); + + + @StreamListener(Sink.INPUT) + public void receive(FooPojo fooPojo) { + receivedArguments.add(fooPojo); + latch.countDown(); + } + } + + @EnableBinding(Processor.class) + @EnableAutoConfiguration + public static class TestStringProcessor { + + List receivedPojos = new ArrayList<>(); + + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public String receive(FooPojo fooPojo) { + receivedPojos.add(fooPojo); + return fooPojo.getBar(); + } + } + + @EnableBinding(Processor.class) + @EnableAutoConfiguration + public static class TestPojoWithMimeType { + + List receivedPojos = new ArrayList<>(); + + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public BazPojo receive(FooPojo fooPojo) { + receivedPojos.add(fooPojo); + BazPojo bazPojo = new BazPojo(); + bazPojo.setQux(fooPojo.getBar()); + return bazPojo; + } + } + + @EnableBinding(Processor.class) + @EnableAutoConfiguration + public static class TestPojoWithAnnotatedArguments { + + List receivedArguments = new ArrayList<>(); + + @StreamListener(Processor.INPUT) + public void receive(@Payload FooPojo fooPojo, @Headers Map headers, + @Header(MessageHeaders.CONTENT_TYPE) String contentType) { + receivedArguments.add(fooPojo); + receivedArguments.add(headers); + receivedArguments.add(contentType); + } + } + + @EnableBinding(Processor.class) + @EnableAutoConfiguration + public static class TestPojoWithMessageReturn { + + List receivedPojos = new ArrayList<>(); + + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public Message receive(FooPojo fooPojo) { + receivedPojos.add(fooPojo); + BazPojo bazPojo = new BazPojo(); + bazPojo.setQux(fooPojo.getBar()); + return MessageBuilder.withPayload(bazPojo).setHeader("foo", "bar").build(); + } + } + + @EnableBinding(Processor.class) + @EnableAutoConfiguration + public static class TestPojoWithMessageArgument { + + List> receivedMessages = new ArrayList<>(); + + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public BazPojo receive(Message fooMessage) { + receivedMessages.add(fooMessage); + BazPojo bazPojo = new BazPojo(); + bazPojo.setQux(fooMessage.getPayload()); + return bazPojo; + } + } + + @EnableBinding(Processor.class) + @EnableAutoConfiguration + public static class TestDuplicateMapping { + + @StreamListener(Processor.INPUT) + public void receive(Message fooMessage) { + } + + @StreamListener(Processor.INPUT) + public void receive2(Message fooMessage) { + } + } + + @EnableBinding(Processor.class) + @EnableAutoConfiguration + public static class TestHandlerBean { + + @Bean + public HandlerBean handlerBean() { + return new HandlerBean(); + } + } + + public static class HandlerBean { + + List receivedPojos = new ArrayList<>(); + + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public BazPojo receive(FooPojo fooMessage) { + receivedPojos.add(fooMessage); + BazPojo bazPojo = new BazPojo(); + bazPojo.setQux(fooMessage.getBar()); + return bazPojo; + } + } + + public static class FooPojo { + + private String bar; + + public String getBar() { + return bar; + } + + public void setBar(String bar) { + this.bar = bar; + } + } + + public static class BazPojo { + + private String qux; + + public String getQux() { + return qux; + } + + public void setQux(String qux) { + this.qux = qux; + } + } +} 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 index f7293ad17..51827a04c 100644 --- 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 @@ -21,7 +21,6 @@ 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; @@ -52,15 +51,15 @@ public class CustomMessageConverterTests { @Test public void testCustomMessageConverter() throws Exception { assertTrue(customMessageConverters.size() == 2); - assertThat(customMessageConverters, hasItem(isA(FooConverter.class))); - assertThat(customMessageConverters, hasItem(isA(BarConverter.class))); + assertThat(customMessageConverters, hasItem(isA(FooToBarConverter.class))); + assertThat(customMessageConverters, hasItem(isA(BarToFooConverter.class))); testSource.output().send(MessageBuilder.withPayload(new Foo("hi")).build()); @SuppressWarnings("unchecked") Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null)) .messageCollector().forChannel(testSource.output()).poll(1, TimeUnit.SECONDS); Assert.assertThat(received, notNullValue()); assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString(), - equalTo("application/x-java-object;type=org.springframework.cloud.stream.config.CustomMessageConverterTests$Bar")); + equalTo("test/bar")); } @EnableBinding(Source.class) @@ -71,19 +70,19 @@ public class CustomMessageConverterTests { @Bean public AbstractFromMessageConverter fooConverter() { - return new FooConverter(); + return new FooToBarConverter(); } @Bean public AbstractFromMessageConverter barConverter() { - return new BarConverter(); + return new BarToFooConverter(); } } - public static class FooConverter extends AbstractFromMessageConverter { + public static class FooToBarConverter extends AbstractFromMessageConverter { - public FooConverter() { - super(MimeType.valueOf("foo/test")); + public FooToBarConverter() { + super(MimeType.valueOf("test/bar")); } @Override @@ -109,15 +108,14 @@ public class CustomMessageConverterTests { logger.error(e.getMessage(), e); return null; } - return buildConvertedMessage(result, message.getHeaders(), - MessageConverterUtils.javaObjectMimeType(targetClass)); + return result; } } - public static class BarConverter extends AbstractFromMessageConverter { + public static class BarToFooConverter extends AbstractFromMessageConverter { - public BarConverter() { - super(MimeType.valueOf("bar/test")); + public BarToFooConverter() { + super(MimeType.valueOf("test/foo")); } @Override @@ -143,8 +141,7 @@ public class CustomMessageConverterTests { logger.error(e.getMessage(), e); return null; } - return buildConvertedMessage(result, message.getHeaders(), - MessageConverterUtils.javaObjectMimeType(targetClass)); + return result; } } 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 index 08e3f3224..052071a95 100644 --- 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 @@ -1,2 +1,2 @@ spring.cloud.stream.bindings.output.destination=configure1 -spring.cloud.stream.bindings.output.contentType=foo/test +spring.cloud.stream.bindings.output.contentType=test/bar diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamListener.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamListener.java new file mode 100644 index 000000000..cd55a11e0 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamListener.java @@ -0,0 +1,63 @@ +/* + * Copyright 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. + * 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.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +import org.springframework.messaging.handler.annotation.MessageMapping; + +/** + * Annotation that marks a method to be a listener to an input component declared through {@link EnableBinding} + * (e.g. a channel). + * + * Annotated methods are allowed to have flexible signatures, as described by {@link MessageMapping}. + * + * If the method returns a {@link org.springframework.messaging.Message}, the result will be automatically sent + * to a channel, as follows: + *
    + *
  • A result of the type {@link org.springframework.messaging.Message} will be sent as-is
    • + *
  • All other results will become the payload of a {@link org.springframework.messaging.Message}
    • + *
+ * + * The target channel of the return message is determined by consulting in the following order: + *
    + *
  • The {@link org.springframework.messaging.MessageHeaders} of the resulting message.
    • + *
  • The value set on the {@link org.springframework.messaging.handler.annotation.SendTo} annotation, if present
    • + *
+ * + * @see {@link MessageMapping} + * @see {@link EnableBinding} + * @see {@link org.springframework.messaging.handler.annotation.SendTo} + * @author Marius Bogoevici + */ +@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) +@Retention(RetentionPolicy.RUNTIME) +@MessageMapping +@Documented +public @interface StreamListener { + + /** + * The name of the bound component (e.g. channel) that the method subscribes to. + * + */ + String value() default ""; + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java index d85430112..23f2b6042 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java @@ -313,7 +313,7 @@ public abstract class AbstractBinder mimeTypesCache = new ConcurrentHashMap<>(); 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 2374e64f5..1c4c358b7 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 @@ -15,10 +15,8 @@ */ package org.springframework.cloud.stream.binding; -import java.util.Collection; + import java.util.Collections; -import java.util.HashSet; -import java.util.Set; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -27,57 +25,46 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; -import org.springframework.cloud.stream.converter.AbstractFromMessageConverter; -import org.springframework.cloud.stream.converter.ByteArrayToStringMessageConverter; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -import org.springframework.cloud.stream.converter.JavaToSerializedMessageConverter; -import org.springframework.cloud.stream.converter.JsonToPojoMessageConverter; -import org.springframework.cloud.stream.converter.JsonToTupleMessageConverter; import org.springframework.cloud.stream.converter.MessageConverterUtils; -import org.springframework.cloud.stream.converter.PojoToJsonMessageConverter; -import org.springframework.cloud.stream.converter.PojoToStringMessageConverter; -import org.springframework.cloud.stream.converter.SerializedToJavaMessageConverter; -import org.springframework.cloud.stream.converter.StringToByteArrayMessageConverter; -import org.springframework.cloud.stream.converter.TupleToJsonMessageConverter; import org.springframework.integration.channel.AbstractMessageChannel; 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.converter.SmartMessageConverter; 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 data types and message converters based on {@link - * BindingProperties#contentType} - * {@link BindingProperties}. This also adds a {@link org.springframework.messaging.support.ChannelInterceptor} to + * BindingProperties#contentType}. 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. + * binding property of the channel. + @ * @author Ilayaperumal Gopinathan + * @author Marius Bogoevici */ public class MessageConverterConfigurer implements MessageChannelConfigurer, BeanFactoryAware, InitializingBean { + private final MessageBuilderFactory messageBuilderFactory; + private ConfigurableListableBeanFactory beanFactory; - private CompositeMessageConverterFactory messageConverterFactory; + private final CompositeMessageConverterFactory compositeMessageConverterFactory; private final ChannelBindingServiceProperties channelBindingServiceProperties; - private final Collection customMessageConverters; - - private final MessageBuilderFactory messageBuilderFactory; - public MessageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties, - Collection customMessageConverters, - MessageBuilderFactory messageBuilderFactory) { - this.channelBindingServiceProperties = channelBindingServiceProperties; - this.customMessageConverters = customMessageConverters; + MessageBuilderFactory messageBuilderFactory, + CompositeMessageConverterFactory compositeMessageConverterFactory) { + Assert.notNull(compositeMessageConverterFactory, "The message converter factory cannot be null"); this.messageBuilderFactory = messageBuilderFactory; + this.channelBindingServiceProperties = channelBindingServiceProperties; + this.compositeMessageConverterFactory = compositeMessageConverterFactory; } @Override @@ -88,20 +75,6 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea @Override 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()); - messageConverters.add(new PojoToJsonMessageConverter()); - messageConverters.add(new ByteArrayToStringMessageConverter()); - messageConverters.add(new StringToByteArrayMessageConverter()); - messageConverters.add(new PojoToStringMessageConverter()); - messageConverters.add(new JavaToSerializedMessageConverter()); - messageConverters.add(new SerializedToJavaMessageConverter()); - this.messageConverterFactory = new CompositeMessageConverterFactory(messageConverters); } /** @@ -115,14 +88,13 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea AbstractMessageChannel messageChannel = (AbstractMessageChannel) channel; BindingProperties bindingProperties = this.channelBindingServiceProperties.getBindingProperties(channelName); final String contentType = bindingProperties.getContentType(); - if (bindingProperties != null && StringUtils.hasText(contentType)) { + if (StringUtils.hasText(contentType)) { MimeType mimeType = MessageConverterUtils.getMimeType(contentType); - MessageConverter messageConverter = this.messageConverterFactory.newInstance(mimeType); - Class[] supportedDataTypes = this.messageConverterFactory.supportedDataTypes(mimeType); + SmartMessageConverter messageConverter = this.compositeMessageConverterFactory.getMessageConverterForType(mimeType); + Class[] supportedDataTypes = this.compositeMessageConverterFactory.supportedDataTypes(mimeType); messageChannel.setDatatypes(supportedDataTypes); - messageChannel.setMessageConverter(messageConverter); + messageChannel.setMessageConverter(new MessageWrappingMessageConverter(messageConverter, mimeType)); messageChannel.addInterceptor(new ChannelInterceptorAdapter() { - @Override public Message preSend(Message message, MessageChannel messageChannel) { Object contentTypeFromMessage = message.getHeaders().get(MessageHeaders.CONTENT_TYPE); @@ -137,4 +109,69 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea }); } } + + /** + * A {@link SmartMessageConverter} that delegates to another {@link SmartMessageConverter} for conversion. + * + * Will wrap the returning result of the conversion into a {@link Message} if it is not a {@link Message} + * instance already. + */ + class MessageWrappingMessageConverter implements SmartMessageConverter { + + private final MimeType contentType; + + private final SmartMessageConverter delegate; + + public MessageWrappingMessageConverter(SmartMessageConverter delegate, MimeType contentType) { + Assert.notNull(delegate, "Delegate converter cannot be null"); + Assert.notNull(contentType, "Content type cannot be null"); + this.delegate = delegate; + this.contentType = contentType; + } + + @Override + public Object fromMessage(Message message, Class targetClass) { + Object converted = delegate.fromMessage(message, targetClass); + if (converted instanceof Message) { + return converted; + } + else { + return build(converted, message.getHeaders()); + } + } + + @Override + public Object fromMessage(Message message, Class targetClass, Object conversionHint) { + Object converted = delegate.fromMessage(message, targetClass, conversionHint); + if (converted == null || converted instanceof Message) { + return converted; + } + else { + return build(converted, message.getHeaders()); + } + } + + @Override + public Message toMessage(Object payload, MessageHeaders headers) { + return delegate.toMessage(payload, headers); + } + + @Override + public Message toMessage(Object payload, MessageHeaders headers, Object conversionHint) { + return delegate.toMessage(payload, headers, conversionHint); + } + + /** + * Convenience method to construct a converted message + * @param payload the converted payload + * @param headers the existing message headers + * @return the converted message + */ + protected final Object build(Object payload, MessageHeaders headers) { + MimeType messageContentType = MessageConverterUtils.X_JAVA_OBJECT.equals(contentType) ? + MessageConverterUtils.javaObjectMimeType(payload.getClass()) : contentType; + return messageBuilderFactory.withPayload(payload).copyHeaders(headers) + .copyHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, messageContentType.toString())).build(); + } + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java new file mode 100644 index 000000000..c4498236f --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java @@ -0,0 +1,204 @@ +/* + * Copyright 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. + * 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.binding; + +import java.lang.reflect.Method; +import java.util.HashMap; +import java.util.Map; + +import org.springframework.aop.framework.Advised; +import org.springframework.aop.support.AopUtils; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.core.DestinationResolver; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; +import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.StringUtils; + +/** + * {@link BeanPostProcessor} that handles {@link StreamListener} annotations found on bean methods. + * + * @author Marius Bogoevici + */ +public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProcessor, ApplicationContextAware, SmartInitializingSingleton { + + private final DestinationResolver binderAwareChannelResolver; + + private final MessageHandlerMethodFactory messageHandlerMethodFactory; + + private final Map mappedBindings = new HashMap<>(); + + private ConfigurableApplicationContext applicationContext; + + public StreamListenerAnnotationBeanPostProcessor(DestinationResolver binderAwareChannelResolver, MessageHandlerMethodFactory messageHandlerMethodFactory) { + Assert.notNull(binderAwareChannelResolver, "Destination resolver cannot be null"); + Assert.notNull(messageHandlerMethodFactory, "Message handler method factory cannot be null"); + this.binderAwareChannelResolver = binderAwareChannelResolver; + this.messageHandlerMethodFactory = messageHandlerMethodFactory; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = (ConfigurableApplicationContext) applicationContext; + } + + @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + @Override + public Object postProcessAfterInitialization(final Object bean, String beanName) throws BeansException { + Class targetClass = AopUtils.isAopProxy(bean) ? AopUtils.getTargetClass(bean) : bean.getClass(); + ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() { + @Override + public void doWith(final Method method) throws IllegalArgumentException, IllegalAccessException { + StreamListener streamListener = AnnotationUtils.findAnnotation(method, StreamListener.class); + if (streamListener != null) { + Method targetMethod = checkProxy(method, bean); + Assert.hasText(streamListener.value(), "The binding name cannot be null"); + final InvocableHandlerMethod invocableHandlerMethod = messageHandlerMethodFactory.createInvocableHandlerMethod(bean, targetMethod); + if (!StringUtils.hasText(streamListener.value())) { + throw new BeanInitializationException("A bound component name must be specified"); + } + if (mappedBindings.containsKey(streamListener.value())) { + throw new BeanInitializationException("Duplicate @" + StreamListener.class.getSimpleName() + + " mapping for '" + streamListener.value() + "' on " + invocableHandlerMethod.getShortLogMessage() + + " already existing for " + mappedBindings.get(streamListener.value()).getShortLogMessage()); + } + mappedBindings.put(streamListener.value(), invocableHandlerMethod); + // TODO: support pollable channels https://github.com/spring-cloud/spring-cloud-stream/issues/436 + SubscribableChannel channel = applicationContext.getBean(streamListener.value(), + SubscribableChannel.class); + final String defaultOutputChannel = extractDefaultOutput(method); + if (invocableHandlerMethod.isVoid()) { + Assert.isTrue(StringUtils.isEmpty(defaultOutputChannel), "An output channel cannot be specified for a method that " + + "does not return a value"); + } + else { + Assert.isTrue(!StringUtils.isEmpty(defaultOutputChannel), "An output channel must be specified for a method that " + + "can return a value"); + } + StreamListenerMessageHandler handler = new StreamListenerMessageHandler(invocableHandlerMethod); + handler.setApplicationContext(applicationContext); + handler.setChannelResolver(binderAwareChannelResolver); + if (!StringUtils.isEmpty(defaultOutputChannel)) { + handler.setOutputChannelName(defaultOutputChannel); + } + handler.afterPropertiesSet(); + channel.subscribe(handler); + } + } + }); + return bean; + } + + @Override + public void afterSingletonsInstantiated() { + // Dump the mappings after the context has been created, ensuring that beans can be processed correctly + // again. + mappedBindings.clear(); + } + + private String extractDefaultOutput(Method method) { + SendTo sendTo = AnnotationUtils.findAnnotation(method, SendTo.class); + if (sendTo != null) { + Assert.isTrue(!ObjectUtils.isEmpty(sendTo.value()), "At least one output must be specified"); + Assert.isTrue(sendTo.value().length == 1, "Multiple destinations cannot be specified"); + Assert.hasText(sendTo.value()[0], "An empty destination cannot be specified"); + return sendTo.value()[0]; + } + return null; + } + + private Method checkProxy(Method methodArg, Object bean) { + Method method = methodArg; + if (AopUtils.isJdkDynamicProxy(bean)) { + try { + // Found a @StreamListener method on the target class for this JDK proxy -> + // is it also present on the proxy itself? + method = bean.getClass().getMethod(method.getName(), method.getParameterTypes()); + Class[] proxiedInterfaces = ((Advised) bean).getProxiedInterfaces(); + for (Class iface : proxiedInterfaces) { + try { + method = iface.getMethod(method.getName(), method.getParameterTypes()); + break; + } + catch (NoSuchMethodException noMethod) { + } + } + } + catch (SecurityException ex) { + ReflectionUtils.handleReflectionException(ex); + } + catch (NoSuchMethodException ex) { + throw new IllegalStateException(String.format( + "@StreamListener method '%s' found on bean target class '%s', " + + "but not found in any interface(s) for bean JDK proxy. Either " + + "pull the method up to an interface or switch to subclass (CGLIB) " + + "proxies by setting proxy-target-class/proxyTargetClass " + + "attribute to 'true'", method.getName(), method.getDeclaringClass().getSimpleName()), ex); + } + } + return method; + } + + private class StreamListenerMessageHandler extends AbstractReplyProducingMessageHandler { + + private final InvocableHandlerMethod invocableHandlerMethod; + + public StreamListenerMessageHandler(InvocableHandlerMethod invocableHandlerMethod) { + this.invocableHandlerMethod = invocableHandlerMethod; + } + + @Override + protected boolean shouldCopyRequestHeaders() { + return false; + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + try { + return invocableHandlerMethod.invoke(requestMessage); + } + catch (Exception e) { + if (e instanceof MessagingException) { + throw (MessagingException) e; + } + else { + throw new MessagingException(requestMessage, "Exception thrown while invoking " + invocableHandlerMethod.getShortLogMessage(), e); + } + } + } + } +} 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 65fa4983d..bf5c24328 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 @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.config; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -34,6 +35,7 @@ import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binding.BindableChannelFactory; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor; +import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor; import org.springframework.cloud.stream.binding.ChannelBindingService; import org.springframework.cloud.stream.binding.CompositeMessageChannelConfigurer; import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener; @@ -46,9 +48,11 @@ 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.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.DependsOn; +import org.springframework.context.annotation.Lazy; import org.springframework.expression.PropertyAccessor; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.config.IntegrationEvaluationContextFactoryBean; @@ -58,7 +62,9 @@ import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.messaging.core.DestinationResolver; +import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; import org.springframework.tuple.spel.TuplePropertyAccessor; +import org.springframework.util.CollectionUtils; /** * Configuration class that provides necessary beans for {@link MessageChannel} binding. @@ -76,7 +82,7 @@ public class ChannelBindingServiceConfiguration { private static final String ERROR_CHANNEL_NAME = "error"; @Autowired - MessageBuilderFactory messageBuilderFactory; + private MessageBuilderFactory messageBuilderFactory; /** * User defined custom message converters @@ -96,15 +102,15 @@ public class ChannelBindingServiceConfiguration { } @Bean - public MessageConverterConfigurer messageConverterConfigurer - (ChannelBindingServiceProperties channelBindingServiceProperties) { - return new MessageConverterConfigurer(channelBindingServiceProperties, customMessageConverters, - messageBuilderFactory); + public MessageConverterConfigurer messageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties, + MessageBuilderFactory messageBuilderFactory, + CompositeMessageConverterFactory compositeMessageConverterFactory) { + return new MessageConverterConfigurer(channelBindingServiceProperties, messageBuilderFactory, compositeMessageConverterFactory); } @Bean - public BindableChannelFactory channelFactory(ChannelBindingServiceProperties channelBindingServiceProperties) { - return new DefaultBindableChannelFactory(compositeMessageChannelConfigurer(channelBindingServiceProperties)); + public BindableChannelFactory channelFactory(CompositeMessageChannelConfigurer compositeMessageChannelConfigurer) { + return new DefaultBindableChannelFactory(compositeMessageChannelConfigurer); } @Bean @@ -115,10 +121,10 @@ public class ChannelBindingServiceConfiguration { @Bean public CompositeMessageChannelConfigurer compositeMessageChannelConfigurer - (ChannelBindingServiceProperties channelBindingServiceProperties) { + (MessageConverterConfigurer messageConverterConfigurer, MessageHistoryTrackerConfigurer messageHistoryTrackerConfigurer) { List configurerList = new ArrayList<>(); - configurerList.add(messageConverterConfigurer(channelBindingServiceProperties)); - configurerList.add((messageHistoryTrackerConfigurer(channelBindingServiceProperties))); + configurerList.add(messageConverterConfigurer); + configurerList.add((messageHistoryTrackerConfigurer)); return new CompositeMessageChannelConfigurer(configurerList); } @@ -161,6 +167,15 @@ public class ChannelBindingServiceConfiguration { return new DynamicDestinationsBindable(); } + @Bean + public CompositeMessageConverterFactory compositeMessageConverterFactory() { + List messageConverters = new ArrayList<>(); + if (!CollectionUtils.isEmpty(customMessageConverters)) { + messageConverters.addAll(Collections.unmodifiableCollection(customMessageConverters)); + } + return new CompositeMessageConverterFactory(messageConverters); + } + // IMPORTANT: Nested class to avoid instantiating all of the above early @Configuration protected static class PostProcessorConfiguration { @@ -221,4 +236,12 @@ public class ChannelBindingServiceConfiguration { }; } } + + @Bean + public static StreamListenerAnnotationBeanPostProcessor bindToAnnotationBeanPostProcessor(@Lazy BinderAwareChannelResolver binderAwareChannelResolver, @Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) { + DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory(); + messageHandlerMethodFactory.setMessageConverter(compositeMessageConverterFactory.getMessageConverterForAllRegistered()); + messageHandlerMethodFactory.afterPropertiesSet(); + return new StreamListenerAnnotationBeanPostProcessor(binderAwareChannelResolver, messageHandlerMethodFactory); + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/AbstractFromMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/AbstractFromMessageConverter.java index 0c9cbdcb3..78066c75a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/AbstractFromMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/AbstractFromMessageConverter.java @@ -24,7 +24,6 @@ import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.AbstractMessageConverter; @@ -41,6 +40,7 @@ import org.springframework.util.MimeType; * * @author David Turanski * @author Ilayaperumal Gopinathan + * @author Marius Bogoevici */ public abstract class AbstractFromMessageConverter extends AbstractMessageConverter { @@ -63,8 +63,7 @@ public abstract class AbstractFromMessageConverter extends AbstractMessageConver /** * Creates a converter that handles one or more content-type message headers - * - * @param supportedSourceMimeTypes list of {@link MimeType} that may present in content-type header + * @param supportedSourceMimeTypes list of {@link MimeType} that may present in content-type header * @param targetMimeType the required target type */ protected AbstractFromMessageConverter(Collection supportedSourceMimeTypes, MimeType targetMimeType) { @@ -75,21 +74,19 @@ public abstract class AbstractFromMessageConverter extends AbstractMessageConver /** * Creates a converter that handles one or more content-type message headers and one or more target MIME types - * - * @param supportedSourceMimeTypes a list of supported content types + * @param supportedSourceMimeTypes a list of supported content types * @param targetMimeTypes a list of supported target types */ protected AbstractFromMessageConverter(Collection supportedSourceMimeTypes, - Collection targetMimeTypes) { + Collection targetMimeTypes) { super(supportedSourceMimeTypes); Assert.notNull(targetMimeTypes, "'targetMimeTypes' cannot be null"); - this.targetMimeTypes = new ArrayList(targetMimeTypes); + this.targetMimeTypes = new ArrayList<>(targetMimeTypes); } /** * Creates a converter that requires a specific content-type message header - * - * @param supportedSourceMimeType {@link MimeType} that must be present in content-type header + * @param supportedSourceMimeType {@link MimeType} that must be present in content-type header * @param targetMimeType the required target type */ protected AbstractFromMessageConverter(MimeType supportedSourceMimeType, MimeType targetMimeType) { @@ -98,8 +95,7 @@ public abstract class AbstractFromMessageConverter extends AbstractMessageConver /** * Creates a converter that requires a specific content-type message header and supports multiple target MIME types. - * - * @param supportedSourceMimeType {@link MimeType} that must be present in content-type header + * @param supportedSourceMimeType {@link MimeType} that must be present in content-type header * @param targetMimeTypes a list of supported target types */ protected AbstractFromMessageConverter(MimeType supportedSourceMimeType, Collection targetMimeTypes) { @@ -148,8 +144,7 @@ public abstract class AbstractFromMessageConverter extends AbstractMessageConver public boolean supportsTargetMimeType(MimeType mimeType) { for (MimeType targetMimeType : targetMimeTypes) { - if (mimeType.getType().equals(targetMimeType.getType()) && mimeType.getSubtype().equals( - targetMimeType.getSubtype())) { + if (targetMimeType.includes(mimeType)) { return true; } } @@ -172,18 +167,4 @@ public abstract class AbstractFromMessageConverter extends AbstractMessageConver throw new UnsupportedOperationException("'convertTo' not supported"); } - /** - * Convenience method to construct a converted message - * - * @param payload the converted payload - * @param headers the existing message headers - * @param contentType the value of the content-type header - * @return the converted message - */ - protected final Message buildConvertedMessage(Object payload, MessageHeaders headers, MimeType contentType) { - return MessageBuilder.withPayload(payload).copyHeaders(headers) - .copyHeaders( - Collections.singletonMap(MessageHeaders.CONTENT_TYPE, - contentType)).build(); - } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ByteArrayToStringMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ByteArrayToStringMessageConverter.java index a152492a7..de2d35e70 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ByteArrayToStringMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ByteArrayToStringMessageConverter.java @@ -81,6 +81,5 @@ public class ByteArrayToStringMessageConverter extends AbstractFromMessageConver } } return converted; - } } 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 3e57038a0..1af335690 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 @@ -18,14 +18,13 @@ package org.springframework.cloud.stream.converter; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; import org.springframework.messaging.converter.CompositeMessageConverter; 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.ObjectUtils; @@ -41,12 +40,35 @@ public class CompositeMessageConverterFactory { private final List converters; + public CompositeMessageConverterFactory() { + this(Collections.emptyList()); + } + /** - * @param converters a list of {@link AbstractFromMessageConverter} + * @param customConverters a list of {@link AbstractFromMessageConverter} */ - public CompositeMessageConverterFactory(Collection converters) { - Assert.notNull(converters, "'converters' cannot be null"); - this.converters = new ArrayList<>(converters); + public CompositeMessageConverterFactory(List customConverters) { + if (!CollectionUtils.isEmpty(customConverters)) { + this.converters = new ArrayList<>(customConverters); + } + else { + this.converters = new ArrayList<>(); + } + initDefaultConverters(); + } + + + + private void initDefaultConverters() { + this.converters.add(new JsonToTupleMessageConverter()); + this.converters.add(new TupleToJsonMessageConverter()); + this.converters.add(new JsonToPojoMessageConverter()); + this.converters.add(new PojoToJsonMessageConverter()); + this.converters.add(new ByteArrayToStringMessageConverter()); + this.converters.add(new StringToByteArrayMessageConverter()); + this.converters.add(new PojoToStringMessageConverter()); + this.converters.add(new JavaToSerializedMessageConverter()); + this.converters.add(new SerializedToJavaMessageConverter()); } /** @@ -54,7 +76,7 @@ public class CompositeMessageConverterFactory { * @param targetMimeType the target MIME type * @return a converter for the target MIME type */ - public CompositeMessageConverter newInstance(MimeType targetMimeType) { + public CompositeMessageConverter getMessageConverterForType(MimeType targetMimeType) { List targetMimeTypeConverters = new ArrayList(); for (AbstractFromMessageConverter converter : converters) { if (converter.supportsTargetMimeType(targetMimeType)) { @@ -68,6 +90,10 @@ public class CompositeMessageConverterFactory { return new CompositeMessageConverter(targetMimeTypeConverters); } + public CompositeMessageConverter getMessageConverterForAllRegistered() { + return new CompositeMessageConverter(new ArrayList(converters)); + } + public Class[] supportedDataTypes(MimeType targetMimeType) { Set> supportedDataTypes = new HashSet<>(); // Make sure to check if the target type is of explicit java object type. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaToSerializedMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaToSerializedMessageConverter.java index d2be552c9..d58f2939c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaToSerializedMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaToSerializedMessageConverter.java @@ -55,9 +55,7 @@ public class JavaToSerializedMessageConverter extends AbstractFromMessageConvert logger.error(e.getMessage(), e); return null; } - - return buildConvertedMessage(bos.toByteArray(), message.getHeaders(), - MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT); + return bos.toByteArray(); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java index 129336c1d..0cb408337 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java @@ -16,12 +16,11 @@ package org.springframework.cloud.stream.converter; -import org.springframework.messaging.Message; -import org.springframework.util.MimeTypeUtils; - import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; +import org.springframework.messaging.Message; +import org.springframework.util.MimeTypeUtils; /** * @@ -63,7 +62,6 @@ public class JsonToPojoMessageConverter extends AbstractFromMessageConverter { logger.error(e.getMessage(), e); return null; } - return buildConvertedMessage(result, message.getHeaders(), - MessageConverterUtils.javaObjectMimeType(targetClass)); + return result; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToTupleMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToTupleMessageConverter.java index 2bf07610f..e656e0f1b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToTupleMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToTupleMessageConverter.java @@ -64,7 +64,6 @@ public class JsonToTupleMessageConverter extends AbstractFromMessageConverter { else { source = (String) message.getPayload(); } - Tuple t = TupleBuilder.fromString(source); - return buildConvertedMessage(t, message.getHeaders(), MessageConverterUtils.javaObjectMimeType(t.getClass())); + return TupleBuilder.fromString(source); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.java index a92b931ea..bf53088fb 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.java @@ -16,14 +16,14 @@ package org.springframework.cloud.stream.converter; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.messaging.Message; -import org.springframework.util.MimeTypeUtils; - import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.messaging.Message; +import org.springframework.util.MimeTypeUtils; + /** * A {@link org.springframework.messaging.converter.MessageConverter} @@ -74,6 +74,6 @@ public class PojoToJsonMessageConverter extends AbstractFromMessageConverter { logger.error(e.getMessage(), e); return null; } - return buildConvertedMessage(result, message.getHeaders(), MimeTypeUtils.APPLICATION_JSON); + return result; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java index c2661c262..1341a8977 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java @@ -55,7 +55,7 @@ public class PojoToStringMessageConverter extends AbstractFromMessageConverter { else { payloadString = message.getPayload().toString(); } - return buildConvertedMessage(payloadString, message.getHeaders(), MimeTypeUtils.TEXT_PLAIN); + return payloadString; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SerializedToJavaMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SerializedToJavaMessageConverter.java index 5bb3d557d..f92d6a062 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SerializedToJavaMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SerializedToJavaMessageConverter.java @@ -57,7 +57,6 @@ public class SerializedToJavaMessageConverter extends AbstractFromMessageConvert return null; } - return buildConvertedMessage(result, message.getHeaders(), - MessageConverterUtils.javaObjectMimeType(targetClass)); + return result; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java index aea185817..9412cc853 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java @@ -18,14 +18,14 @@ package org.springframework.cloud.stream.converter; import java.io.IOException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; + import org.springframework.beans.factory.annotation.Value; import org.springframework.messaging.Message; import org.springframework.tuple.Tuple; import org.springframework.util.MimeTypeUtils; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.SerializationFeature; - /** * A {@link org.springframework.messaging.converter.MessageConverter} @@ -75,7 +75,7 @@ public class TupleToJsonMessageConverter extends AbstractFromMessageConverter { else { json = t.toString(); } - return buildConvertedMessage(json, message.getHeaders(), MimeTypeUtils.APPLICATION_JSON); + return json; } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java index 3e81f02f9..d906fe08f 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java @@ -50,6 +50,7 @@ import org.springframework.cloud.stream.binding.DynamicDestinationsBindable; import org.springframework.cloud.stream.binding.MessageConverterConfigurer; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.support.StaticApplicationContext; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.support.DefaultMessageBuilderFactory; @@ -96,7 +97,7 @@ public class BinderAwareChannelResolverTests { bindings.put("foo", bindingProperties); this.channelBindingServiceProperties.setBindings(bindings); MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer( - this.channelBindingServiceProperties, null, new DefaultMessageBuilderFactory()); + this.channelBindingServiceProperties, new DefaultMessageBuilderFactory(), new CompositeMessageConverterFactory()); messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class)); messageConverterConfigurer.afterPropertiesSet(); this.bindableChannelFactory = new DefaultBindableChannelFactory(messageConverterConfigurer); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java index 506c8784e..dfe182a23 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java @@ -56,6 +56,7 @@ import org.springframework.cloud.stream.binder.DefaultBinderFactory; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.cloud.stream.utils.MockBinderConfiguration; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.support.DefaultMessageBuilderFactory; @@ -202,7 +203,7 @@ public class ChannelBindingServiceTests { when(binder.bindProducer( matches("bar"), any(DirectChannel.class), any(ProducerProperties.class))).thenReturn(mockBinding); BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binderFactory, properties, dynamicDestinationsBindable, - new DefaultBindableChannelFactory(new MessageConverterConfigurer(properties, null, new DefaultMessageBuilderFactory()))); + new DefaultBindableChannelFactory(new MessageConverterConfigurer(properties, new DefaultMessageBuilderFactory(), new CompositeMessageConverterFactory()))); ConfigurableListableBeanFactory beanFactory = mock(ConfigurableListableBeanFactory.class); when(beanFactory.getBean("mock:bar", MessageChannel.class)) .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class));