376 lines
17 KiB
Plaintext
376 lines
17 KiB
Plaintext
[[message]]
|
|
=== Message
|
|
|
|
The Spring Integration `Message` is a generic container for data.
|
|
Any object can be provided as the payload, and each `Message` instance includes headers containing user-extensible properties as key-value pairs.
|
|
|
|
[[message-interface]]
|
|
==== The `Message` Interface
|
|
|
|
The following listing shows the definition of the `Message` interface:
|
|
|
|
====
|
|
[source,java]
|
|
----
|
|
public interface Message<T> {
|
|
|
|
T getPayload();
|
|
|
|
MessageHeaders getHeaders();
|
|
|
|
}
|
|
----
|
|
====
|
|
|
|
The `Message` interface is a core part of the API.
|
|
By encapsulating the data in a generic wrapper, the messaging system can pass it around without any knowledge of the data's type.
|
|
As an application evolves to support new types or when the types themselves are modified or extended, the messaging system is not affected.
|
|
On the other hand, when some component in the messaging system does require access to information about the `Message`, such metadata can typically be stored to and retrieved from the metadata in the message headers.
|
|
|
|
[[message-headers]]
|
|
==== Message Headers
|
|
|
|
Just as Spring Integration lets any `Object` be used as the payload of a `Message`, it also supports any `Object` types as header values.
|
|
In fact, the `MessageHeaders` class implements the `java.util.Map_ interface`, as the following class definition shows:
|
|
|
|
====
|
|
[source,java]
|
|
----
|
|
public final class MessageHeaders implements Map<String, Object>, Serializable {
|
|
...
|
|
}
|
|
----
|
|
====
|
|
|
|
NOTE: Even though the `MessageHeaders` class implements `Map`, it is effectively a read-only implementation.
|
|
Any attempt to `put` a value in the Map results in an `UnsupportedOperationException`.
|
|
The same applies for `remove` and `clear`.
|
|
Since messages may be passed to multiple consumers, the structure of the `Map` cannot be modified.
|
|
Likewise, the message's payload `Object` can not be `set` after the initial creation.
|
|
However, the mutability of the header values themselves (or the payload Object) is intentionally left as a decision for the framework user.
|
|
|
|
As an implementation of `Map`, the headers can be retrieved by calling `get(..)` with the name of the header.
|
|
Alternatively, you can provide the expected `Class` as an additional parameter.
|
|
Even better, when retrieving one of the pre-defined values, convenient getters are available.
|
|
The following example shows each of these three options:
|
|
|
|
====
|
|
[source,java]
|
|
----
|
|
Object someValue = message.getHeaders().get("someKey");
|
|
|
|
CustomerId customerId = message.getHeaders().get("customerId", CustomerId.class);
|
|
|
|
Long timestamp = message.getHeaders().getTimestamp();
|
|
----
|
|
====
|
|
|
|
The following table describes the pre-defined message headers:
|
|
|
|
.Pre-defined Message Headers
|
|
|
|
[cols="2l,2l,5", options="header"]
|
|
|===
|
|
| Header Name
|
|
| Header Type
|
|
| Usage
|
|
|
|
| MessageHeaders.ID
|
|
| java.util.UUID
|
|
| An identifier for this message instance.
|
|
Changes each time a message is mutated.
|
|
|
|
| MessageHeaders.
|
|
TIMESTAMP
|
|
| java.lang.Long
|
|
| The time the message was created.
|
|
Changes each time a message is mutated.
|
|
|
|
| MessageHeaders.
|
|
REPLY_CHANNEL
|
|
| java.lang.Object
|
|
(String or
|
|
MessageChannel)
|
|
| A channel to which a reply (if any) is sent when no explicit output channel is configured and there is no `ROUTING_SLIP` or the `ROUTING_SLIP` is exhausted.
|
|
If the value is a `String`, it must represent a bean name or have been generated by a `ChannelRegistry.`
|
|
|
|
| MessageHeaders.
|
|
ERROR_CHANNEL
|
|
| java.lang.Object
|
|
(String or
|
|
MessageChannel)
|
|
| A channel to which errors are sent.
|
|
If the value is a `String`, it must represent a bean name or have been generated by a `ChannelRegistry.`
|
|
|===
|
|
|
|
Many inbound and outbound adapter implementations also provide or expect certain headers, and you can configure additional user-defined headers.
|
|
Constants for these headers can be found in those modules where such headers exist -- for example.
|
|
`AmqpHeaders`, `JmsHeaders`, and so on.
|
|
|
|
[[message-header-accessor]]
|
|
===== `MessageHeaderAccessor` API
|
|
|
|
Starting with Spring Framework 4.0 and Spring Integration 4.0, the core messaging abstraction has been moved to the `spring-messaging` module, and the `MessageHeaderAccessor` API has been introduced to provide additional abstraction over messaging implementations.
|
|
All (core) Spring Integration-specific message headers constants are now declared in the `IntegrationMessageHeaderAccessor` class.
|
|
The following table describes the pre-defined message headers:
|
|
|
|
.Pre-defined Message Headers
|
|
[cols="5l,3l,5", options="header"]
|
|
|===
|
|
| Header Name
|
|
| Header Type
|
|
| Usage
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
CORRELATION_ID
|
|
| java.lang.Object
|
|
| Used to correlate two or more messages.
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
SEQUENCE_NUMBER
|
|
| java.lang.Integer
|
|
| Usually a sequence number with a group of messages with a `SEQUENCE_SIZE` but can also be used in a `<resequencer/>` to resequence an unbounded group of messages.
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
SEQUENCE_SIZE
|
|
| java.lang.Integer
|
|
| The number of messages within a group of correlated messages.
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
EXPIRATION_DATE
|
|
| java.lang.Long
|
|
| Indicates when a message is expired.
|
|
Not used by the framework directly but can be set with a header enricher and used in a `<filter/>` that is configured with an `UnexpiredMessageSelector`.
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
PRIORITY
|
|
| java.lang.Integer
|
|
| Message priority -- for example, within a `PriorityChannel`.
|
|
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
DUPLICATE_MESSAGE
|
|
| java.lang.Boolean
|
|
| True if a message was detected as a duplicate by an idempotent receiver interceptor.
|
|
See <<idempotent-receiver>>.
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
CLOSEABLE_RESOURCE
|
|
| java.io.Closeable
|
|
| This header is present if the message is associated with a `Closeable` that should be closed when message processing is complete.
|
|
An example is the `Session` associated with a streamed file transfer using FTP, SFTP, and so on.
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
DELIVERY_ATTEMPT
|
|
| java.lang.
|
|
AtomicInteger
|
|
| If a message-driven channel adapter supports the configuration of a `RetryTemplate`, this header contains the current delivery attempt.
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
ACKNOWLEDGMENT_CALLBACK
|
|
| o.s.i.support.
|
|
Acknowledgment
|
|
Callback
|
|
| If a message source supports it, a call back to accept, reject, or requeue a message.
|
|
See <<deferred-acks-message-source>>.
|
|
|===
|
|
|
|
Convenient typed getters for some of these headers are provided on the `IntegrationMessageHeaderAccessor` class, as the following example shows:
|
|
|
|
====
|
|
[source,java]
|
|
----
|
|
IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor(message);
|
|
int sequenceNumber = accessor.getSequenceNumber();
|
|
Object correlationId = accessor.getCorrelationId();
|
|
...
|
|
----
|
|
====
|
|
|
|
The following table describes headers that also appear in the `IntegrationMessageHeaderAccessor` but are generally not used by user code (that is, they are generally used by internal parts of Spring Integration -- their inclusion here is for completeness):
|
|
|
|
.Pre-defined Message Headers
|
|
[cols="5l,3l,5", options="header"]
|
|
|===
|
|
| Header Name
|
|
| Header Type
|
|
| Usage
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
SEQUENCE_DETAILS
|
|
| java.util.
|
|
List<List<Object>>
|
|
| A stack of correlation data used when nested correlation is needed (for example,
|
|
`splitter->...->splitter->...->aggregator->...->aggregator`).
|
|
|
|
| IntegrationMessageHeaderAccessor.
|
|
ROUTING_SLIP
|
|
| java.util.
|
|
Map<List<Object>, Integer>
|
|
| See <<routing-slip>>.
|
|
|===
|
|
|
|
[[message-id-generation]]
|
|
===== Message ID Generation
|
|
|
|
When a message transitions through an application, each time it is mutated (for example,
|
|
by a transformer) a new message ID is assigned.
|
|
The message ID is a `UUID`.
|
|
Beginning with Spring Integration 3.0, the default strategy used for IS generation is more efficient than the previous `java.util.UUID.randomUUID()` implementation.
|
|
It uses simple random numbers based on a secure random seed instead of creating a secure random number each time.
|
|
|
|
A different UUID generation strategy can be selected by declaring a bean that implements `org.springframework.util.IdGenerator` in the application context.
|
|
|
|
IMPORTANT: Only one UUID generation strategy can be used in a classloader.
|
|
This means that, if two or more application contexts run in the same classloader, they share the same strategy.
|
|
If one of the contexts changes the strategy, it is used by all contexts.
|
|
If two or more contexts in the same classloader declare a bean of type `org.springframework.util.IdGenerator`, they must all be an instance of the same class.
|
|
Otherwise, the context attempting to replace a custom strategy fails to initialize.
|
|
If the strategy is the same, but parameterized, the strategy in the first context to be initialized is used.
|
|
|
|
In addition to the default strategy, two additional `IdGenerators` are provided.
|
|
`org.springframework.util.JdkIdGenerator` uses the previous `UUID.randomUUID()` mechanism.
|
|
You can use `o.s.i.support.IdGenerators.SimpleIncrementingIdGenerator` when a UUID is not really needed and a simple incrementing value is sufficient.
|
|
|
|
[[read-only-headers]]
|
|
===== Read-only Headers
|
|
|
|
The `MessageHeaders.ID` and `MessageHeaders.TIMESTAMP` are read-only headers and cannot be overridden.
|
|
|
|
Since version 4.3.2, the `MessageBuilder` provides the `readOnlyHeaders(String... readOnlyHeaders)` API to customize a list of headers that should not be copied from an upstream `Message`.
|
|
Only the `MessageHeaders.ID` and `MessageHeaders.TIMESTAMP` are read only by default.
|
|
The global `spring.integration.readOnly.headers` property (see <<global-properties>>) is provided to customize `DefaultMessageBuilderFactory` for framework components.
|
|
This can be useful when you would like do not populate some out-of-the-box headers, such as `contentType` by the `ObjectToJsonTransformer` (see <<json-transformers>>).
|
|
|
|
When you try to build a new message using `MessageBuilder`, this kind of header is ignored and a particular `INFO` message is emitted to logs.
|
|
|
|
Starting with version 5.0, <<gateway,Messaging Gateway>>, <<header-enricher,Header Enricher>>, <<payload-enricher,Content Enricher>> and <<header-filter,Header Filter>> do not let you configure the `MessageHeaders.ID` and `MessageHeaders.TIMESTAMP` header names when `DefaultMessageBuilderFactory` is used, and they throw `BeanInitializationException`.
|
|
|
|
[[header-propagation]]
|
|
===== Header Propagation
|
|
|
|
When messages are processed (and modified) by message-producing endpoints (such as a <<service-activator,service activator>>), in general, inbound headers are propagated to the outbound message.
|
|
One exception to this is a <<transformer,transformer>>, when a complete message is returned to the framework.
|
|
In that case, the user code is responsible for the entire outbound message.
|
|
When a transformer just returns the payload, the inbound headers are propagated.
|
|
Also, a header is only propagated if it does not already exist in the outbound message, letting you change header values as needed.
|
|
|
|
Starting with version 4.3.10, you can configure message handlers (that modify messages and produce output) to suppress the propagation of specific headers.
|
|
To configure the header(s) you do not want to be copied, call the `setNotPropagatedHeaders()` or `addNotPropagatedHeaders()` methods on the `MessageProducingMessageHandler` abstract class.
|
|
|
|
You can also globally suppress propagation of specific message headers by setting the `readOnlyHeaders` property in `META-INF/spring.integration.properties` to a comma-delimited list of headers.
|
|
|
|
Starting with version 5.0, the `setNotPropagatedHeaders()` implementation on the `AbstractMessageProducingHandler` applies simple patterns (`xxx*`, `*xxx`, `*xxx*`, or `xxx*yyy`) to allow filtering headers with a common suffix or prefix.
|
|
See https://docs.spring.io/spring-integration/api/org/springframework/integration/util/PatternMatchUtils.html[`PatternMatchUtils` Javadoc] for more information.
|
|
When one of the patterns is `*` (asterisk), no headers are propagated.
|
|
All other patterns are ignored.
|
|
In that case, the service activator behaves the same way as a transformer and any required headers must be supplied in the `Message` returned from the service method.
|
|
The `notPropagatedHeaders()` option is available in the `ConsumerEndpointSpec` for the Java DSL
|
|
It is also available for XML configuration of the `<service-activator>` component as a `not-propagated-headers` attribute.
|
|
|
|
IMPORTANT: Header propagation suppression does not apply to those endpoints that do not modify the message, such as <<bridge,bridges>> and <<router,routers>>.
|
|
|
|
[[message-implementations]]
|
|
==== Message Implementations
|
|
|
|
The base implementation of the `Message` interface is `GenericMessage<T>`, and it provides two constructors, shown in the following listing:
|
|
|
|
====
|
|
[source,java]
|
|
----
|
|
new GenericMessage<T>(T payload);
|
|
|
|
new GenericMessage<T>(T payload, Map<String, Object> headers)
|
|
----
|
|
====
|
|
|
|
When a `Message` is created, a random unique ID is generated.
|
|
The constructor that accepts a `Map` of headers copies the provided headers to the newly created `Message`.
|
|
|
|
There is also a convenient implementation of `Message` designed to communicate error conditions.
|
|
This implementation takes a `Throwable` object as its payload, as the following example shows:
|
|
|
|
====
|
|
[source,java]
|
|
----
|
|
ErrorMessage message = new ErrorMessage(someThrowable);
|
|
|
|
Throwable t = message.getPayload();
|
|
----
|
|
====
|
|
|
|
Note that this implementation takes advantage of the fact that the `GenericMessage` base class is parameterized.
|
|
Therefore, as shown in both examples, no casting is necessary when retrieving the `Message` payload `Object`.
|
|
|
|
[[message-builder]]
|
|
==== The `MessageBuilder` Helper Class
|
|
|
|
You may notice that the `Message` interface defines retrieval methods for its payload and headers but provides no setters.
|
|
The reason for this is that a `Message` cannot be modified after its initial creation.
|
|
Therefore, when a `Message` instance is sent to multiple consumers (for example,
|
|
through a publish-subscribe Channel), if one of those consumers needs to send a reply with a different payload type, it must create a new `Message`.
|
|
As a result, the other consumers are not affected by those changes.
|
|
Keep in mind that multiple consumers may access the same payload instance or header value, and whether such an instance is itself immutable is a decision left to you.
|
|
In other words, the contract for `Message` instances is similar to that of an unmodifiable `Collection`, and the `MessageHeaders` map further exemplifies that.
|
|
Even though the `MessageHeaders` class implements `java.util.Map`, any attempt to invoke a `put` operation (or 'remove' or 'clear') on a `MessageHeaders` instance results in an `UnsupportedOperationException`.
|
|
|
|
Rather than requiring the creation and population of a Map to pass into the GenericMessage constructor, Spring Integration does provide a far more convenient way to construct Messages: `MessageBuilder`.
|
|
The `MessageBuilder` provides two factory methods for creating `Message` instances from either an existing `Message` or with a payload `Object`.
|
|
When building from an existing `Message`, the headers and payload of that `Message` are copied to the new `Message`, as the following example shows:
|
|
|
|
====
|
|
[source,java]
|
|
----
|
|
Message<String> message1 = MessageBuilder.withPayload("test")
|
|
.setHeader("foo", "bar")
|
|
.build();
|
|
|
|
Message<String> message2 = MessageBuilder.fromMessage(message1).build();
|
|
|
|
assertEquals("test", message2.getPayload());
|
|
assertEquals("bar", message2.getHeaders().get("foo"));
|
|
----
|
|
====
|
|
|
|
If you need to create a `Message` with a new payload but still want to copy the headers from an existing `Message`, you can use one of the 'copy' methods, as the following example shows:
|
|
|
|
[source,java]
|
|
----
|
|
Message<String> message3 = MessageBuilder.withPayload("test3")
|
|
.copyHeaders(message1.getHeaders())
|
|
.build();
|
|
|
|
Message<String> message4 = MessageBuilder.withPayload("test4")
|
|
.setHeader("foo", 123)
|
|
.copyHeadersIfAbsent(message1.getHeaders())
|
|
.build();
|
|
|
|
assertEquals("bar", message3.getHeaders().get("foo"));
|
|
assertEquals(123, message4.getHeaders().get("foo"));
|
|
----
|
|
|
|
Note that the `copyHeadersIfAbsent` method does not overwrite existing values.
|
|
Also, in the preceding example, you can see how to set any user-defined header with `setHeader`.
|
|
Finally, there are `set` methods available for the predefined headers as well as a non-destructive method for setting any header (`MessageHeaders` also defines constants for the pre-defined header names).
|
|
|
|
You can also use `MessageBuilder` to set the priority of messages, as the following example shows:
|
|
|
|
[source,java]
|
|
----
|
|
Message<Integer> importantMessage = MessageBuilder.withPayload(99)
|
|
.setPriority(5)
|
|
.build();
|
|
|
|
assertEquals(5, importantMessage.getHeaders().getPriority());
|
|
|
|
Message<Integer> lessImportantMessage = MessageBuilder.fromMessage(importantMessage)
|
|
.setHeaderIfAbsent(IntegrationMessageHeaderAccessor.PRIORITY, 2)
|
|
.build();
|
|
|
|
assertEquals(2, lessImportantMessage.getHeaders().getPriority());
|
|
|
|
----
|
|
|
|
The `priority` header is considered only when using a `PriorityChannel` (as described in the next chapter).
|
|
It is defined as a `java.lang.Integer`.
|