SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Introduce StreamListener annotation

Browse Source 
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;
This commit is contained in:
Marius Bogoevici
2016-03-09 01:45:00 -05:00
committed by Mark Pollack
parent 0aeef58495
commit 4d2ef3c4d0
20 changed files with 840 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

380
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/BindingListenerTests.java Normal file
View File

@@ -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<String, String>) testPojoWithAnnotatedArguments.receivedArguments.get(1),
hasEntry(MessageHeaders.CONTENT_TYPE, "application/json"));
assertThat((Map<String, String>) 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<String> message = (Message<String>) 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<String> message = (Message<String>) 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<BazPojo> message = (Message<BazPojo>) 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<BazPojo> message = (Message<BazPojo>) 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<BazPojo> message = (Message<BazPojo>) 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<String> message = (Message<String>) 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<FooPojo> 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<FooPojo> 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<FooPojo> 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<Object> receivedArguments = new ArrayList<>();
@StreamListener(Processor.INPUT)
public void receive(@Payload FooPojo fooPojo, @Headers Map<String, Object> 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<FooPojo> 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<Message<String>> receivedMessages = new ArrayList<>();
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public BazPojo receive(Message<String> 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<String> fooMessage) {
}
@StreamListener(Processor.INPUT)
public void receive2(Message<String> fooMessage) {
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestHandlerBean {
@Bean
public HandlerBean handlerBean() {
return new HandlerBean();
}
}
public static class HandlerBean {
List<FooPojo> 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;
}
}
}

29
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomMessageConverterTests.java
View File

@@ -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<String> received = (Message<String>) ((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;
}
}

2
spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/custom/source-channel-configurers.properties
View File

@@ -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

63
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamListener.java Normal file
View File

@@ -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:
* <ul>
* <li>A result of the type {@link org.springframework.messaging.Message} will be sent as-is</li>
* <li>All other results will become the payload of a {@link org.springframework.messaging.Message}</li>
* </ul>
*
* The target channel of the return message is determined by consulting in the following order:
* <ul>
* <li>The {@link org.springframework.messaging.MessageHeaders} of the resulting message.</li>
* <li>The value set on the {@link org.springframework.messaging.handler.annotation.SendTo} annotation, if present</li>
* </ul>
*
* @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 "";
}

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java
View File

@@ -313,7 +313,7 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
* @author David Turanski
* @author Ilayaperumal Gopinathan
*/
abstract static class JavaClassMimeTypeConversion {
public abstract static class JavaClassMimeTypeConversion {
private static ConcurrentMap<String, MimeType> mimeTypesCache = new ConcurrentHashMap<>();

131
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -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<AbstractFromMessageConverter> customMessageConverters;
private final MessageBuilderFactory messageBuilderFactory;
public MessageConverterConfigurer(ChannelBindingServiceProperties channelBindingServiceProperties,
Collection<AbstractFromMessageConverter> 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<AbstractFromMessageConverter> 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();
}
}
}

204
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java Normal file
View File

@@ -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<MessageChannel> binderAwareChannelResolver;
private final MessageHandlerMethodFactory messageHandlerMethodFactory;
private final Map<String, InvocableHandlerMethod> mappedBindings = new HashMap<>();
private ConfigurableApplicationContext applicationContext;
public StreamListenerAnnotationBeanPostProcessor(DestinationResolver<MessageChannel> 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);
}
}
}
}
}

43
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java
View File

@@ -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<MessageChannelConfigurer> 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<AbstractFromMessageConverter> 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);
}
}

35
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/AbstractFromMessageConverter.java
View File

@@ -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<MimeType> 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<MimeType> supportedSourceMimeTypes,
Collection<MimeType> targetMimeTypes) {
Collection<MimeType> targetMimeTypes) {
super(supportedSourceMimeTypes);
Assert.notNull(targetMimeTypes, "'targetMimeTypes' cannot be null");
this.targetMimeTypes = new ArrayList<MimeType>(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<MimeType> 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();
}
}

1
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ByteArrayToStringMessageConverter.java
View File

@@ -81,6 +81,5 @@ public class ByteArrayToStringMessageConverter extends AbstractFromMessageConver
}
}
return converted;
}
}

40
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java
View File

@@ -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<AbstractFromMessageConverter> converters;
public CompositeMessageConverterFactory() {
this(Collections.<AbstractFromMessageConverter>emptyList());
}
/**
* @param converters a list of {@link AbstractFromMessageConverter}
* @param customConverters a list of {@link AbstractFromMessageConverter}
*/
public CompositeMessageConverterFactory(Collection<AbstractFromMessageConverter> converters) {
Assert.notNull(converters, "'converters' cannot be null");
this.converters = new ArrayList<>(converters);
public CompositeMessageConverterFactory(List<? extends AbstractFromMessageConverter> 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<MessageConverter> targetMimeTypeConverters = new ArrayList<MessageConverter>();
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<MessageConverter>(converters));
}
public Class<?>[] supportedDataTypes(MimeType targetMimeType) {
Set<Class<?>> supportedDataTypes = new HashSet<>();
// Make sure to check if the target type is of explicit java object type.

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaToSerializedMessageConverter.java
View File

@@ -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();
}
}

8
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java
View File

@@ -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;
}
}

3
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToTupleMessageConverter.java
View File

@@ -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);
}
}

10
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.java
View File

@@ -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;
}
}

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java
View File

@@ -55,7 +55,7 @@ public class PojoToStringMessageConverter extends AbstractFromMessageConverter {
else {
payloadString = message.getPayload().toString();
}
return buildConvertedMessage(payloadString, message.getHeaders(), MimeTypeUtils.TEXT_PLAIN);
return payloadString;
}
}

3
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SerializedToJavaMessageConverter.java
View File

@@ -57,7 +57,6 @@ public class SerializedToJavaMessageConverter extends AbstractFromMessageConvert
return null;
}
return buildConvertedMessage(result, message.getHeaders(),
MessageConverterUtils.javaObjectMimeType(targetClass));
return result;
}
}

8
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java
View File

@@ -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;
}
}

3
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java
View File

@@ -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);

3
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java
View File

@@ -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));