diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index e4857d2a7..dc58a1a30 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -23,6 +23,8 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; import org.springframework.validation.annotation.Validated; /** @@ -36,6 +38,8 @@ import org.springframework.validation.annotation.Validated; @JsonInclude(Include.NON_DEFAULT) @Validated public class BindingProperties { + + public static final MimeType DEFAULT_CONTENT_TYPE = MimeTypeUtils.APPLICATION_JSON; private static final String COMMA = ","; @@ -60,7 +64,7 @@ public class BindingProperties { * Specifies content-type that will be used by this binding in the event * it is not specified in Message headers. Default: 'application/json'. */ - private String contentType = "application/json"; + private String contentType = DEFAULT_CONTENT_TYPE.toString(); /** * The name of the binder to use for this binding in the event multiple binders available (e.g., 'rabbit'); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java index ca4c5bda3..99368fe0f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java @@ -47,6 +47,12 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC private final Map, JavaType> typeCache = new ConcurrentHashMap<>(); + ApplicationJsonMessageMarshallingConverter(@Nullable ObjectMapper objectMapper) { + if (objectMapper != null) { + this.setObjectMapper(objectMapper); + } + } + @Override protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) { if (payload instanceof byte[]) { 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 d1a9ca429..6af4acd2c 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 @@ -25,9 +25,11 @@ import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.messaging.converter.AbstractMessageConverter; import org.springframework.messaging.converter.ByteArrayMessageConverter; import org.springframework.messaging.converter.CompositeMessageConverter; +import org.springframework.messaging.converter.DefaultContentTypeResolver; import org.springframework.messaging.converter.MessageConverter; import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; @@ -66,20 +68,24 @@ public class CompositeMessageConverterFactory { this.converters = new ArrayList<>(); } initDefaultConverters(); + + DefaultContentTypeResolver resolver = new DefaultContentTypeResolver(); + resolver.setDefaultMimeType(BindingProperties.DEFAULT_CONTENT_TYPE); + this.converters.stream() + .filter(mc -> mc instanceof AbstractMessageConverter) + .forEach(mc -> ((AbstractMessageConverter)mc).setContentTypeResolver(resolver)); } @SuppressWarnings("deprecation") private void initDefaultConverters() { - ApplicationJsonMessageMarshallingConverter applicationJson = new ApplicationJsonMessageMarshallingConverter(); - applicationJson.setStrictContentTypeMatch(true); - if (this.objectMapper != null) { - applicationJson.setObjectMapper(this.objectMapper); - } - this.converters.add(applicationJson); - + ApplicationJsonMessageMarshallingConverter applicationJsonConverter = new ApplicationJsonMessageMarshallingConverter(this.objectMapper); + applicationJsonConverter.setStrictContentTypeMatch(true); + this.converters.add(applicationJsonConverter); this.converters.add(new TupleJsonMessageConverter(this.objectMapper)); this.converters.add(new ByteArrayMessageConverter()); this.converters.add(new ObjectStringMessageConverter()); + + // Deprecated converters this.converters.add(new JavaSerializationMessageConverter()); this.converters.add(new KryoMessageConverter(null,true)); this.converters.add(new JsonUnmarshallingConverter(this.objectMapper)); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java index 636e9f130..b475f45c3 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -31,7 +31,9 @@ import org.springframework.messaging.converter.AbstractMessageConverter; /** * @author Marius Bogoevici * @author Oleg Zhurakousky + * @deprecated as of 2.0. Will be removed in 2.1 */ +@Deprecated public class JavaSerializationMessageConverter extends AbstractMessageConverter { public JavaSerializationMessageConverter() { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonUnmarshallingConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonUnmarshallingConverter.java index 61ffe45bb..451688c4e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonUnmarshallingConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonUnmarshallingConverter.java @@ -30,7 +30,11 @@ import org.springframework.messaging.converter.MessageConversionException; * as input. * * @author Marius Bogoevici + * + * @deprecated as of 2.0. */ +// NOTE we need to revisit as to why do we need it in the first place, given that our first converter already handles JSON +@Deprecated public class JsonUnmarshallingConverter extends AbstractMessageConverter { private final ObjectMapper objectMapper; diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java index 44c8912c5..7f694233f 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java @@ -41,8 +41,10 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.DirectChannel; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.AbstractMessageConverter; import org.springframework.messaging.converter.MessageConversionException; @@ -69,6 +71,19 @@ import static org.junit.Assert.assertTrue; */ public class ContentTypeTckTests { + @Test + public void withInternalPipeline() { + ApplicationContext context = new SpringApplicationBuilder(InternalPipeLine.class) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false"); + InputDestination source = context.getBean(InputDestination.class); + OutputDestination target = context.getBean(OutputDestination.class); + String jsonPayload = "{\"name\":\"oleg\"}"; + source.send(new GenericMessage(jsonPayload.getBytes())); + Message outputMessage = target.receive(); + assertEquals("OLEG", new String(outputMessage.getPayload())); + } + @Test public void pojoToPojo() { ApplicationContext context = new SpringApplicationBuilder(PojoToPojoStreamListener.class) @@ -533,6 +548,27 @@ public class ContentTypeTckTests { } } + @EnableBinding(Processor.class) + @Import(TestChannelBinderConfiguration.class) + public static class InternalPipeLine { + @StreamListener(Processor.INPUT) + @SendTo("internalChannel") + public String handleA(Person value) { + return "{\"name\":\"" + value.getName().toUpperCase() + "\"}"; + } + + @Bean + public MessageChannel internalChannel() { + return new DirectChannel(); + } + + @StreamListener("internalChannel") + @SendTo(Processor.OUTPUT) + public String handleB(Person value) { + return value.toString(); + } + } + public static class Employee

{ private P person; private int id;