@@ -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<RabbitProducerProperties> dynamicConfigurer() {
|
||||
@@ -1553,196 +1553,167 @@ public NewBindingCallback<RabbitProducerProperties> 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.<channelName>.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.<channelName>.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<T>` 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<T>`.
|
||||
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<byte[]>` 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=<Class being serialized>
|
||||
|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.<channel>.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[]>`, `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<byte[]> 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 <<schema-evolution,the specific section>> 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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user