You can try Spring Cloud Stream in less then 5 min even before you jump into any details and the following _three-step guide_ will help.
We'll create a simple Spring Cloud Stream application which receives messages coming from the messaging middleware of your choice (more on this later) and
logs received messages to the console. We'll call it _LoggingConsumer_. While not very practical it will certainly provide a good introduction to some of the main concepts
We'll create a simple Spring Cloud Stream application which receives messages coming from the messaging middleware of your choice (more on this later) and
logs received messages to the console. We'll call it _LoggingConsumer_. While not very practical it will certainly provide a good introduction to some of the main concepts
and abstractions, making it easier to digest the rest of this user guide.
So let's get started. . .
==== Step One - Create sample Application using Spring Initilaizer
Visit the https://start.spring.io[Spring Initializr]. This is where we'll generate our _LoggingConsumer_ application.
==== Step One - Create sample Application using Spring Initilaizer
Visit the https://start.spring.io[Spring Initializr]. This is where we'll generate our _LoggingConsumer_ application.
In the _Dependencies_ start typing 'stream' and _Cloud Stream_ option should pop up. Select it. Now start typing either 'kafka' or 'rabbit'. Basically this is where you are choosing
In the _Dependencies_ start typing 'stream' and _Cloud Stream_ option should pop up. Select it. Now start typing either 'kafka' or 'rabbit'. Basically this is where you are choosing
what messaging midleware this application will be bound to. Choose the one you have already installed and/or feel more comfortable with installing/running.
Also, as you can see from the Initilaizer screen there are few other options you can choose. For example, you can choose Gradle as your build tool instead of the default Maven.
With the _Dependencies_ selected the only other thing you have to identify is the application name - _logging-consumer_.
@@ -31,7 +31,7 @@ Here you simply import the project into your IDE of choice.
Please keep in mind that dependening on the IDE you may need to follow a specific import procedures. For example depending on how the project was generated (Maven or Gradle)
you may need to follow specific import procedure (e.g., in Eclipse/STS: `File -> Import -> Maven -> Existing Maven Project`).
Ones imported the project must have no errors of any kind and `src/main/java` should also contain `com.example.loggingconsumer.LoggingConsumerApplication`.
Ones imported the project must have no errors of any kind and `src/main/java` should also contain `com.example.loggingconsumer.LoggingConsumerApplication`.
Technically at this point you can just run the application's main class since it's already a valid _Spring Boot_ application, but it does not do anything, so let's add some code.
@@ -70,13 +70,13 @@ public class LoggingConsumerApplication {
As you can see from the above:
* We've enabled `Sink` binding (input-no-output) via `@EnableBinding(Sink.class)`. This will signal to the framework to initiate binding to the messaging middleware where
* We've enabled `Sink` binding (input-no-output) via `@EnableBinding(Sink.class)`. This will signal to the framework to initiate binding to the messaging middleware where
it will auto-create the destination (i.e., queue, topic) which will be bound to `Sink.INPUT` channel.
* We've added handler method to receive incoming Message as type `Person`. What this means is that here youcan already observe one of the core features of the framework where
it will attempt to automatically convert incoming message's payload to type `Person`.
This is it, we now have a fully functional Spring Cloud Stream application that does something. From here for simplicity we'll assume RabbitMQ was selected in _step one_.
Assuming you have RabbitMQ installed and running, start the application by simply running its `main` method.
Assuming you have RabbitMQ installed and running, start the application by simply running its `main` method.
You should see following output:
@@ -88,8 +88,8 @@ You should see following output:
. . .
--- [ main] c.e.l.LoggingConsumerApplication : Started LoggingConsumerApplication in 2.531 seconds (JVM running for 2.897)
Go to RabbitMQ management console or any other RabbitMQ client and simply send message to `input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg`
(NOTE: the `anonymous.CbMIwdkJSBO1ZoPDOtHtCg` part represents the group name and is generated and will be different in your environment. For something more
Go to RabbitMQ management console or any other RabbitMQ client and simply send message to `input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg`
(NOTE: the `anonymous.CbMIwdkJSBO1ZoPDOtHtCg` part represents the group name and is generated and will be different in your environment. For something more
predictable you can use explicit group name via `spring.cloud.stream.bindings.input.group=hello`).
The contents of the message should be JSON representation of `Person` class, so let's send this:
@@ -105,7 +105,7 @@ You can also build/package your application into a boot jar (i.e., `./mvnw clean
That is all!
== What's New in 2.0?
Spring Cloud Stream introduces quite a number of new features, enhancements and changes. The following sections outline most notable ones.
Spring Cloud Stream introduces quite a number of new features, enhancements and changes. The following sections outline most notable ones.
=== New Features and Components
@@ -122,14 +122,14 @@ Please refer to the appropriate section for more details
There are now new new Actuator binding controls to both visualize as well as control Bindings lifecycle. For more details please visit <<Binding visualization and control>>
==== Configurable RetryTemplate
Aside from providing properties to configure `RetryTemplate` we now allow you to provide your own effectively overriding the one provided by the framework. Simply configure
Aside from providing properties to configure `RetryTemplate` we now allow you to provide your own effectively overriding the one provided by the framework. Simply configure
it as a `@Bean` in your application.
=== Notable changes and enhancements
=== Notable changes and enhancements
==== Both Actuator and Web dependencies are now optional
This helps to slim down the footprint of the deployed application in the event neither of the functionality is required.
This helps to slim down the footprint of the deployed application in the event neither of the functionality is required.
It also allows one to swicth between the reactive and conventional web paradigms by adding one of the following dependencies manually:
* Introduction of `@StreamMessageConverter` annotation to provide custom `MessageConverters`.
* Introduction of the default _Content Type_ as `application/json` which needs to be taken into consideration when migrating 1.3
application and/or operating in the mixed mode (i.e., 1.3 producer -> 2.0 consumer).
* Messages with textual payloads and _contentType_ `text/...` or `.../json` are no longer converted to `Message<String>` for cases where argument type of the provided `MessageHandler`
can not be determnied (i.e., `public void handle(Message<?> message)` or `public void handle(Object payload)`). Further more, a strong argument type may not be enough
* Messages with textual payloads and _contentType_ `text/...` or `.../json` are no longer converted to `Message<String>` for cases where argument type of the provided `MessageHandler`
can not be determnied (i.e., `public void handle(Message<?> message)` or `public void handle(Object payload)`). Further more, a strong argument type may not be enough
to properly convert messages, so `contentType` header is may be used as supplement by some `MessageConverters`.
=== Notable Deprecations
==== Java serialization (Java native and Kryo)
* `JavaSerializationMessageConverter` and `KryoMessageConverter`. While these two converters remain for now, they will be moved out of the core packages and support in the future.
The main reason for this deprecation is to signal the issue _type-based language-specific_ serialization couuld cause in the distributed environments, where Producers and Consumers
may not only depend on different JVM versions or have different versions of supporting libraries (i.e., Kryo), but to also draw the attention to the fact that Consumers and Producers
may not only depend on different JVM versions or have different versions of supporting libraries (i.e., Kryo), but to also draw the attention to the fact that Consumers and Producers
may and in a lot of cases are non-Java based.
==== Deprecated classes and methods
Following is a quick summary of notable deprecations. See corresponding javadocs fort more details.
* `SharedChannelRegistry` in favor of `SharedBindingTargetRegistry`.
* `SharedChannelRegistry` in favor of `SharedBindingTargetRegistry`.
* `Bindings` - beans qualified by it are already uniquely identified by their type. For example, provided `Source`, `Processor` or custom bindings:
_PAUSED_ and _RESUMED_ are only effective if corresponding binder and its underlyig technology supports it, otherwise you'll see the warning message in the logs.
====
_PAUSED_ and _RESUMED_ are only effective if corresponding binder and its underlyig technology supports it, otherwise you'll see the warning message in the logs.
Currently only Kafka binder supports _PAUSED_ and _RESUMED_ state.
====
@@ -1557,26 +1557,26 @@ NOTE: If you need to support dynamic destinations with multiple binder types, us
=== Introduction
Data transformation is one of the core features of any message-driven microservice architecture. Given that in Spring Cloud Stream, such data
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:
_1. To convert the contents of the incoming message to match the signature of the application-provided handler._
_2. To convert the contents of the outgoing message to the wire format._
The wire format is typically `byte[]` (i.e., Kafka and Rabbit binders), but is governed by the binder implementation.
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]
====
As a supplement to the details to follow you may also want to read the following
As a supplement to the details to follow you may also want to read the following
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
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]
@@ -1586,35 +1586,35 @@ handler as an example. Also let’s assume that this is the only handler in the
public String handle(Person person) {..}
----
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.
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`.
Spring Cloud Stream provides three simple mechanisms to define `contentType` and they all come with precedence order:
_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
_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._
_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
_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._
_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
_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._
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.
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.
Another reason for making `application/json` the default stems from the interoperability requirements driven by distributed microservices architectures where producer and consumer not only
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.
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`
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
@@ -1622,24 +1622,24 @@ If there is an internal pipeline the `Message` is sent to the next handler going
==== Content type vs. argument type
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.
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).
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
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.
In other words:
[NOTE]
====
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.
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.
It is important to understand the contract of these methods and their usage specifically in the context of Spring Cloud Stream.
It is important to understand the contract of these methods and their usage specifically in the context of Spring Cloud Stream.
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.
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.
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`.
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]
====
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
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.
====
1. `ApplicationJsonMessageMarshallingConverter` - _variation of the `org.springframework.messaging.converter.MappingJackson2MessageConverter`. Supports conversion of the payload of the
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`.
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
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`)._
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`
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
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]
Reference in New Issue
Block a user
Blocking a user prevents them from interacting with repositories, such as opening or commenting on pull requests or issues. Learn more about blocking a user.
Gary Russell