diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 0ff0aadbf..1fef8cd54 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -1537,7 +1537,7 @@ void configure(String channelName, MessageChannel channel, ProducerProperties pr The following is an example using the RabbitMQ binder: -[source, xml] +[source, java] ---- @Bean public NewBindingCallback dynamicConfigurer() { @@ -1553,196 +1553,167 @@ public NewBindingCallback dynamicConfigurer() { NOTE: If you need to support dynamic destinations with multiple binder types, use `Object` for the generic type and cast the `extended` argument as needed. [[contenttypemanagement]] -== Content Type and Transformation +== Content Type negotiation -To allow you to propagate information about the content type of produced messages, Spring Cloud Stream attaches, by default, a `contentType` header to outbound messages. -For middleware that does not directly support headers, Spring Cloud Stream provides its own mechanism of automatically wrapping outbound messages in an envelope of its own. -For middleware that does support headers, Spring Cloud Stream applications may receive messages with a given content type from non-Spring Cloud Stream applications. +=== Introduction -The content type resolution process have been redesigned for Spring Cloud Stream 2.0. +Data transformation is one of the core features of any message-driven microservice architecture. Given that in Spring Cloud Stream, such data +is represented as a Spring `Message`, such message may have to be transformed to a desired shape/size before reaching its destination. This is required for two reasons: -Please read the migrating from 1.3 section to understand the changes when interacting with applications using versions of the framework. +_1. To convert the contents of the incoming message to match the signature of the application-provided handler._ -The framework depends on a `contentType` to be present as a header in order to know how serialize/deserialize a payload. +_2. To convert the contents of the outgoing message to the wire format._ -Spring Cloud Stream allows you to declaratively configure type conversion for inputs and outputs using the `spring.cloud.stream.bindings..content-type` property of a binding. -Note that general type conversion may also be accomplished easily by using a transformer inside your application. +The wire format is typically `byte[]` (i.e., Kafka and Rabbit binders), but is governed by the binder implementation. + +In Spring Cloud Stream, message transformation is accomplished with a `org.springframework.messaging.converter.MessageConverter`. [NOTE] ==== -For both input and output channel, setting a contentType via a property or via annotation only triggers the `default` converter if a message header with value `contentType` is not present. -This is useful for cases where you just want to send a _POJO_ without sending any header information, or to consume messages that do not have a `contentType` header present. -The framework will always override any default settings with the value found on the message headers. +As a supplement to the details to follow you may also want to read the following +https://spring.io/blog/2018/02/26/spring-cloud-stream-2-0-content-type-negotiation-and-transformation[blog] ==== -[TIP] -==== -Although contentType became a required property, the framework will set a default value of `application/json` for all input/output channels if one is not -provided by the user. -==== +=== Mechanics -[[mime-types]] -=== MIME types -The `content-type` values are parsed as media types, e.g., `application/json` or `text/plain;charset=UTF-8`. - -MIME types are especially useful for indicating how to convert to String or byte[] content. -Spring Cloud Stream also uses MIME type format to represent Java types, using the general type `application/x-java-object` with a `type` parameter. -For example, `application/x-java-object;type=java.util.Map` or `application/x-java-object;type=com.bar.Foo` can be set as the `content-type` property of an input binding. -In addition, Spring Cloud Stream provides custom MIME types, notably, `application/x-spring-tuple` to specify a Tuple. - -[[mime-types-and-java-types]] - -=== Channel contentType and Message Headers - -You can configure a message channel content type using `spring.cloud.stream.bindings..content-type` property, or using the `@Input` and `@Output` annotations. -By doing so, even if you send a POJO with no `contentType` information, the framework will set the MessageHeader `contentType` to the specified value set for the channel. - -However, if you send a `Message` and sets the `contentType` manually, that takes precedence over the configured property value. -This is valid for both input and output channels. The `MessageHeader` will always take precedence over the default configured `contentType` for the channel. - -=== ContentType handling for output channels - -Starting with version 2.0, the framework will no longer try to infer a contentType based on the payload `T` of a `Message`. -It will instead use the contentType header (or the default provided by the framework) to configure the right `MessageConverter` to serialize the payload into `byte[]`. - -The `contentType` you set is a hint to activate the corresponding `MessageConverter`. The converter can then modify the contentType to augment the information, such as the case with `Kryo` and `Avro` conveters. - -For outbound messages, if your payload is of typ `byte[]`, the framework will skip the conversion logic, and just write those bytes to the wire. -In this case, if `contentType` of the message is absent, it will set the default value specified to channel. - - -[TIP] -==== -If you intend to bypass conversion, just make sure you set the appropriate `contentType` header, otherwise you could be sending some arbitrary binary data, and the framework may set the header as `application/json` (default). -==== - -The following snippet shows how you can bypass conversion and set the correct contentType header. +To better understand the mechanics and the necessity behind content-type negotiation let’s look at the very simple use case using the following message +handler as an example. Also let’s assume that this is the only handler in the application (no internal pipeline) for simplicity. [source, java] ---- -@Autowired -private Source source; - -public void sendImageData(File f) throws Exception { - byte[] data = Files.readAllBytes(f.toPath()); - MimeType mimeType = (f.getName().endsWith("gif")) ? MimeTypeUtils.IMAGE_GIF : MimeTypeUtils.IMAGE_JPEG; - source.output().send(MessageBuilder.withPayload(data) - .setHeader(MessageHeaders.CONTENT_TYPE, mimeType) - .build()); -} +@StreamListener(Processor.INPUT) +@SendTo(Processor.OUTPUT) +public String handle(Person person) {..} ---- -Regardless of contentType used, the result is always a `Message` with a header `contentType` set. This is what gets passed to the binder to be sent over the wire. +The above handler expects `Person` type as an argument and will produce `String` type as an output. In order for the framework to succeed in passing the incoming +`Message` as an argument to this handler it has to somehow transform the payload of the `Message` from the wire format to `Person` type. +In other words the framework must locate and apply the appropriate `MessageConverter`. To accomplish that the framework needs some instructions +from the user. One of these instructions is already provided by the signature of the handler method itself (`Person` type), so in theory, that should and in some +cases is enough, but for the majority of the use cases in order to select the appropriate `MessageConverter` the framework needs an additional piece of information. +That missing piece is `contentType`. -|=== -|`content-type` header | MessageConverter | `content-type` augmented |Supported types | Comments +Spring Cloud Stream provides three simple mechanisms to define `contentType` and they all come with precedence order: -|application/json -|CustomMappingJackson2MessageConverter -|application/json -| POJO, primitives and Strings that represent JSON data -| It's the default converter if none is specified. Note that if you send a raw String it will be quoted +_1. ***HEADER*** - the `contentType` can be communicated through the Message itself. By simply providing `contentType` header you are declaring the content type to use to locate and +apply the appropriate MessageConverter._ -|text/plain -|ObjectStringMessageConverter -|text/plain -|Invokes `toString()` of the object -| +_2. ***BINDING*** - the `contentType` can be set per destination binding via `spring.cloud.stream.bindings.input.content-type` property. NOTE: the segment `input` in the property name +corresponds to the actual name of the destination which is “input” in our case. This approach allows one to declare per-binding the content type to use to locate and +apply the appropriate MessageConverter._ -|application/x-spring-tuple -|TupleJsonMessageConverter -|application/x-spring-tuple -|org.springframework.tuple.Tuple -| +_3. ***DEFAULT*** - in the event `contentType` is not present in the Message header and/or binding, the default `application/json` content type will be used to +locate and apply the appropriate MessageConverter._ -|application/x-java-serialized-object -|JavaSerializationMessageConverter -|application/x-java-serialized-object -|Any Java type that implements `Serializable` -|This converter uses java native serialization. Receivers of this data must have the same class on the classpath. +As mentioned, the above also demonstrates the order of precedence in the event there is a tie. For example, header provided content type takes precedence over any other content type. +The same applies for content type set per binding which essentially allows one to override the default content type. But it also provides a sensible default which was determined from +the community feedback. -|application/x-java-object -|KryoMessageConverter -|application/x-java-object;type= -|Any Java type that can be serialized using Kryo -|Receivers of this data must have the same class on the classpath. +Another reason for making `application/json` the default stems from the interoperability requirements driven by distributed microservices architectures where producer and consumer not only +run in different JVMs, but can also run on different non-JVM platforms. -|application/avro -|AvroMessageConverter -|application/avro -|A Generic or SpecificRecord from Avro types, a POJO if reflection is used -|Avro needs an associated schema to write/read data. Please refer to the section on the docs on how to use it properly +Once the non-void handler method returns and unless the return value is already a `Message`, the new `Message` is constructed with return vlaue as the payload while inheriting +headers from the input `Message` less the ones defined/filtered by `SpringIntegrationProperties.messageHandlerNotPropagatedHeaders`. +By default, there is only one header set there - `contentType`. This means that the new `Message` will not have `contentType` header set, thus ensuring that the `contentType` +can evolve. You can always opt out to returning a `Message` from the handler method where you can inject any header you wish. -|=== +If there is an internal pipeline the `Message` is sent to the next handler going through the same process of conversion, or if there is no internal + pipeline or you’ve reached the end of it the `Message` is sent back to the output destination. -=== ContentType handling for input channels +==== Content type vs. argument type -For input channels, Spring Cloud Stream uses `@StreamListener` and `@ServiceActivator` content handling to support the conversion. -It does so by checking either the channel `content-type` set via `@Input(contentType="text/plain")` annotation or via `spring.cloud.stream.bindings..contentType` property, or the presense of a header `contentType`. +As it was mentioned, for the framework to select the appropriate MessageConverter it requires _argument type_ and optionally _content type_ information. +The logic for selecting the appropriate `MessageConverter` resides with the argument resolvers (`HandlerMethodArgumentResolvers`), right before the invocation of the user +defined handler method (that is when the actual argument type is known to the framework). +If argument type does NOT match the type of the current payload the framework delegates to the stack of the +pre-configured `MessageConverters` to see if any one of them can convert the payload. As you can see the `Object fromMessage(Message message, Class targetClass);` +operation of the MessageConverter takes `targetClass` as one of its arguments. The framework also ensures that the provided `Message` always contains `contentType` header +in the event one was not there already (injects the default one or the one set per binding). +That is the mechanism by which framework determines if message can be converted to a target type - `contentType` and argumenyt type. +If no appropriate `MessageConverter` is found the exception is thrown at which time you can add custom `MessageConverter` (more on this later). -The framework will check the contentType set for the Message, select the appropriate `MessageConverter` and apply conversion passing the argument as the target type. +But what if the payload type matches the target type declared by the handler method? In this cases there is obviously nothing to convert and the +payload will be passed unmodified. While this sounds pretty straight forward and logical, keep in mind handler methods that take `Message` and/or `Object` as an +argument. By doing so you are essentially forfeiting the conversion process by declaring the target type to be `Object` which is an `instanceof` everything in Java. -If the converter does not support the target type it will return `null`, if *all* configured converters return `null`, a `MessageConversionException` is thrown. - -Just like output channels, if your method payload argument is of type `Message`, `byte[]` or `Message` conversion is skipped and you get the raw bytes from the wire, plus the corresponding headers. - -[TIP] +In other words: +[NOTE] ==== -Remember, the MessageHeader always takes precedence over the annotation or property configuration. +Do NOT expect Message to be converted into some type based on the `contentType` only. Remember that the `contentType` is complimentary to the target type. +A hint if you wish which `MessageConverter` may or may not take into consideration. ==== -|=== -|`content-type` header | MessageConverter | Supported target type | Comments -|application/json -|CustomMappingJackson2MessageConverter -| POJO or String -| +==== Message Converters -|text/plain -|ObjectStringMessageConverter -|String -| +`MessageConverters` define two methods: -|application/x-spring-tuple -|TupleJsonMessageConverter -|org.springframework.tuple.Tuple -| +[source, java] +---- +Object fromMessage(Message message, Class targetClass); -|application/x-java-serialized-object -|JavaSerializationMessageConverter -|Any Java type that implements `Serializable` -| +Message toMessage(Object payload, @Nullable MessageHeaders headers); +---- -|application/x-java-object -|KryoMessageConverter -|Any Java type that can be serialized using Kryo -| +It is important to understand the contract of these methods and their usage specifically in the context of Spring Cloud Stream. -|application/avro -|AvroMessageConverter -|A Generic or SpecificRecord from Avro types, a POJO if reflection is used -|Avro needs an associated schema to write/read data. Please refer to the section on the docs on how to use it properly +The `fromMessage` method converts incoming `Message` to an argument type. The payload of the `Message` could be _any type_ and it's +up to the actual implementation of the `MessageConverter` to support multiple types. For example, some JSON converter may support the payload type as `byte[]` +and `String` etc. This is important when application contains an internal pipeline (i.e., _input -> handler1 -> handler2 ->. . . -> output_) and the output of +the upstream handler results in a `Message` which may not be in the initial wire format. -|=== +However. . . +The `toMessage` method has a more strict contract and must always convert `Message` to the wire format - `byte[]`. -=== Customizing message conversion +So for all intents and purposes (and especially when implementing your own converter) you might as well look at them as: -Besides the conversions that it supports out of the box, Spring Cloud Stream also supports registering your own message conversion implementations. -This allows you to send and receive data in a variety of custom formats, including binary, and associate them with specific `contentTypes`. +[source, java] +---- +Object fromMessage(Message message, Class targetClass); -Spring Cloud Stream registers all the beans of type `org.springframework.messaging.converter.MessageConverter` that are qualifeied using `@StreamConverter` annotation, as custom message converters along with the out of the box message converters. +Message toMessage(Object payload, @Nullable MessageHeaders headers); +---- + +=== Provided MessageConverters + +As it was mentioned earlier the framework already provides a stack of `MessageConverters` to handle most common use cases. Below is the ordered list of provided `MessageConverters`. [NOTE] ==== -The framework requires the `@StreamConverter` qualifier annotation to avoid picking up other converters that may be present on the `ApplicationContext` and could overlap with the default ones. +It is important to understand the importance of the order since the mechanism by which the framework locates the appropriate `MessageConverter` is by iterating through each and asking +if it can convert using the first one that can convert. ==== -If your message converter needs to work with a specific `content-type` and target class (for both input and output), then the message converter needs to extend `org.springframework.messaging.converter.AbstractMessageConverter`. -For conversion when using `@StreamListener`, a message converter that implements `org.springframework.messaging.converter.MessageConverter` would suffice. +1. `ApplicationJsonMessageMarshallingConverter` - _variation of the `org.springframework.messaging.converter.MappingJackson2MessageConverter`. Supports conversion of the payload of the +`Message` from `String` or `byte[]`._ +2. `TupleJsonMessageConverter` - _***[DEPRECATED]*** Supports conversion of the payload of the `Message` from `org.springframework.tuple.Tuple`._ +3. `ByteArrayMessageConverter` - _Supports conversion of the payload of the `Message` from `byte[]` to `byte[]` for cases when `contentType` is set to `application/octet-stream`. +Essentially a pass through and exists primarily for backward compatibility._ +4. `ObjectStringMessageConverter` - _Supports conversion of any type to a `String`, when contentType is `text/plain`. Invokes Object’s `toString()` method or if payload is +`byte[]` then new `String(byte[])`._ +5. `JavaSerializationMessageConverter` - _***[DEPRECATED]*** Supports conversion based on java serialization when `contentType` is `application/x-java-serialized-object`._ +6. `KryoMessageConverter` - _***[DEPRECATED]*** Supports conversion based on kryo serialization when `contentType` is `application/x-java-object`._ +7. `JsonUnmarshallingConverter` - _Similar to the `ApplicationJsonMessageMarshallingConverter`. Supports conversion of any type when `contentType` is `application/x-java-object`. +Expects the actual type information to be embedded in the `contentType` as an attribute (e.g., `application/x-java-object;type=foo.bar.Baz`)._ -Here is an example of creating a message converter bean (with the content-type `application/bar`) inside a Spring Cloud Stream application: +In the event no appropriate converter is found the framework will throw an exception at which point you should check your code and configfuration and ensure you didn't miss anything +(i.e., provide `contentType` via binding or header). However, most likely you are dealing with some uncommon case (custom `contentType` perhaps) and the current stack of provided `MessageConverters` +doesn't know how to convert. And if that's the case you can add custom `MessageConverter`. + +=== User defined Message Converters + +Spring Cloud Stream exposes a mechanism to define and register additional `MessageConverters`. All you need to do is implement `org.springframework.messaging.converter.MessageConverter`, +confiure it as `@Bean` and annotate it with `@StreamMessageConverter` and it will be added to the existing stack of `MessageConverters`. The `@StreamMessageConverter` qualifier annotation +is to avoid picking up other converters that may be present on the _Application Context_. + +[NOTE] +==== +It is important to undetrstand that custom `MessageConverters` are added to the head of the existing stack. +This allows custom `MessageConverters` to take precedence over the existing ones, thus supporting not only addition, but the override of the existing ones. +==== + +Here is an example of creating a message converter bean to support new content type `application/bar`: [source,java] ---- @@ -1784,42 +1755,6 @@ public class MyCustomMessageConverter extends AbstractMessageConverter { Spring Cloud Stream also provides support for Avro-based converters and schema evolution. See <> for details. -=== `@StreamListener` and Message Conversion - -The `@StreamListener` annotation provides a convenient way for converting incoming messages without the need to specify the content type of an input channel. -During the dispatching process to methods annotated with `@StreamListener`, a conversion will be applied automatically if the argument requires it. - -For example, let's consider a message with the String content `{"greeting":"Hello, world"}` and a `content-type` header of `application/json` is received on the input channel. -Let us consider the following application that receives it: - -[source,java] ----- -public class GreetingMessage { - - String greeting; - - public String getGreeting() { - return greeting; - } - - public void setGreeting(String greeting) { - this.greeting = greeting; - } -} - -@EnableBinding(Sink.class) -@EnableAutoConfiguration -public static class GreetingSink { - - @StreamListener(Sink.INPUT) - public void receive(Greeting greeting) { - // handle Greeting - } - } ----- - -The argument of the method will be populated automatically with the POJO containing the unmarshalled form of the JSON String. - [[schema-evolution]] == Schema evolution support 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 99368fe0f..0e14f3911 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 @@ -35,7 +35,8 @@ import org.springframework.messaging.converter.MessageConversionException; /** * Variation of {@link MappingJackson2MessageConverter} to support marshalling and - * unmarshalling of Messages's payload from 'byte[]' to and instance of a 'targetClass' and vice versa. + * unmarshalling of Messages's payload from 'String' or 'byte[]' to an instance of a 'targetClass' + * and and back to 'byte[]' * * * @author Oleg Zhurakousky diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleJsonMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleJsonMessageConverter.java index baaee9e94..97fe6d81b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleJsonMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleJsonMessageConverter.java @@ -37,7 +37,10 @@ import org.springframework.util.MimeTypeUtils; * @author Ilayaperumal Gopinathan * @author Marius Bogoevici * @author Vinicius Carvalho + * + * @deprecated as of 2.0. please use 'application/json' content type */ +@Deprecated public class TupleJsonMessageConverter extends AbstractMessageConverter { private final ObjectMapper objectMapper;