When the downstream flow initiated by that gateway completes, a Message
is returned to the gateway and continues its journey within the current chain.
8.7. Scatter-Gather
Starting with version 4.1, Spring Integration provides an implementation of the scatter-gather enterprise integration pattern. It is a compound endpoint for which the goal is to send a message to the recipients and aggregate the results. As noted in Enterprise Integration Patterns, it is a component for scenarios such as “best quote”, where we need to request information from several suppliers and decide which one provides us with the best term for the requested item.
Previously, the pattern could be configured by using discrete components. This enhancement brings more convenient configuration.
The ScatterGatherHandler
is a request-reply endpoint that combines a PublishSubscribeChannel
(or a RecipientListRouter
) and an AggregatingMessageHandler
.
The request message is sent to the scatter
channel, and the ScatterGatherHandler
waits for the reply that the aggregator sends to the outputChannel
.
8.7.1. Functionality
The Scatter-Gather
pattern suggests two scenarios: “auction” and “distribution”.
In both cases, the aggregation
function is the same and provides all the options available for the AggregatingMessageHandler
.
(Actually, the ScatterGatherHandler
requires only an AggregatingMessageHandler
as a constructor argument.)
See Aggregator for more information.
Auction
The auction Scatter-Gather
variant uses “publish-subscribe” logic for the request message, where the “scatter” channel is a PublishSubscribeChannel
with apply-sequence="true"
.
However, this channel can be any MessageChannel
implementation (as is the case with the request-channel
in the ContentEnricher
— see Content Enricher).
However, in this case, you should create your own custom correlationStrategy
for the aggregation
function.
Distribution
The distribution Scatter-Gather
variant is based on the RecipientListRouter
(see RecipientListRouter
) with all available options for the RecipientListRouter
.
This is the second ScatterGatherHandler
constructor argument.
If you want to rely on only the default correlationStrategy
for the recipient-list-router
and the aggregator
, you should specify apply-sequence="true"
.
Otherwise, you should supply a custom correlationStrategy
for the aggregator
.
Unlike the PublishSubscribeChannel
variant (the auction variant), having a recipient-list-router
selector
option lets filter target suppliers based on the message.
With apply-sequence="true"
, the default sequenceSize
is supplied, and the aggregator
can release the group correctly.
The distribution option is mutually exclusive with the auction option.
For both the auction and the distribution variants, the request (scatter) message is enriched with the gatherResultChannel
header to wait for a reply message from the aggregator
.
By default, all suppliers should send their result to the replyChannel
header (usually by omitting the output-channel
from the ultimate endpoint).
However, the gatherChannel
option is also provided, letting suppliers send their reply to that channel for the aggregation.
8.7.2. Configuring a Scatter-Gather Endpoint
The following example shows Java configuration for the bean definition for Scatter-Gather
:
@Bean
public MessageHandler distributor() {
RecipientListRouter router = new RecipientListRouter();
router.setApplySequence(true);
router.setChannels(Arrays.asList(distributionChannel1(), distributionChannel2(),
distributionChannel3()));
return router;
}
@Bean
public MessageHandler gatherer() {
return new AggregatingMessageHandler(
new ExpressionEvaluatingMessageGroupProcessor("^[payload gt 5] ?: -1D"),
new SimpleMessageStore(),
new HeaderAttributeCorrelationStrategy(
IntegrationMessageHeaderAccessor.CORRELATION_ID),
new ExpressionEvaluatingReleaseStrategy("size() == 2"));
}
@Bean
@ServiceActivator(inputChannel = "distributionChannel")
public MessageHandler scatterGatherDistribution() {
ScatterGatherHandler handler = new ScatterGatherHandler(distributor(), gatherer());
handler.setOutputChannel(output());
return handler;
}
In the preceding example, we configure the RecipientListRouter
distributor
bean with applySequence="true"
and the list of recipient channels.
The next bean is for an AggregatingMessageHandler
.
Finally, we inject both those beans into the ScatterGatherHandler
bean definition and mark it as a @ServiceActivator
to wire the scatter-gather component into the integration flow.
The following example shows how to configure the <scatter-gather>
endpoint by using the XML namespace:
<scatter-gather
id="" (1)
auto-startup="" (2)
input-channel="" (3)
output-channel="" (4)
scatter-channel="" (5)
gather-channel="" (6)
order="" (7)
phase="" (8)
send-timeout="" (9)
gather-timeout="" (10)
requires-reply="" > (11)
<scatterer/> (12)
<gatherer/> (13)
</scatter-gather>
| 1 | The id of the endpoint.
The ScatterGatherHandler bean is registered with an alias of id + '.handler' .
The RecipientListRouter bean is registered with an alias of id + '.scatterer' .
The AggregatingMessageHandler`bean is registered with an alias of `id + '.gatherer' .
Optional.
(The BeanFactory generates a default id value.) |
| 2 | Lifecycle attribute signaling whether the endpoint should be started during application context initialization.
In addition, the ScatterGatherHandler also implements Lifecycle and starts and stops gatherEndpoint , which is created internally if a gather-channel is provided.
Optional.
(The default is true .) |
| 3 | The channel on which to receive request messages to handle them in the ScatterGatherHandler .
Required. |
| 4 | The channel to which the ScatterGatherHandler sends the aggregation results.
Optional.
(Incoming messages can specify a reply channel themselves in the replyChannel message header). |
| 5 | The channel to which to send the scatter message for the auction scenario.
Optional.
Mutually exclusive with the <scatterer> sub-element. |
| 6 | The channel on which to receive replies from each supplier for the aggregation.
It is used as the replyChannel header in the scatter message.
Optional.
By default, the FixedSubscriberChannel is created. |
| 7 | The order of this component when more than one handler is subscribed to the same DirectChannel (use for load balancing purposes).
Optional. |
| 8 | Specifies the phase in which the endpoint should be started and stopped.
The startup order proceeds from lowest to highest, and the shutdown order is from highest to lowest.
By default, this value is Integer.MAX_VALUE , meaning that this container starts as late as possible and stops as soon as possible.
Optional. |
| 9 | The timeout interval to wait when sending a reply Message to the output-channel .
By default, the send blocks for one second.
It applies only if the output channel has some 'sending' limitations — for example, a QueueChannel with a fixed 'capacity' that is full.
In this case, a MessageDeliveryException is thrown.
The send-timeout is ignored for AbstractSubscribableChannel implementations.
For group-timeout(-expression) , the MessageDeliveryException from the scheduled expire task leads this task to be rescheduled.
Optional. |
| 10 | Lets you specify how long the scatter-gather waits for the reply message before returning.
By default, it waits indefinitely.
'null' is returned if the reply times out.
Optional.
It defaults to -1 , meaning to wait indefinitely. |
| 11 | Specifies whether the scatter-gather must return a non-null value.
This value is true by default.
Consequently, a ReplyRequiredException is thrown when the underlying aggregator returns a null value after gather-timeout .
Note, if null is a possibility, the gather-timeout should be specified to avoid an indefinite wait. |
| 12 | The <recipient-list-router> options.
Optional.
Mutually exclusive with scatter-channel attribute. |
| 13 | The <aggregator> options.
Required. |
8.7.3.
In some cases, it is better to just catch and ignore downstream exceptions if the ReleaseStrategy
allows the process to finish with fewer replies than requests.
In other cases something like a “compensation message” should be considered for returning from sub-flow, when an error happens.
Every async sub-flow should be configured with a errorChannel
header for the proper error message sending from the MessagePublishingErrorHandler
.
Otherwise, an error will be sent to the global errorChannel
with the common error handling logic.
See Error Handling for more information about async error processing.
Synchronous flows may use an ExpressionEvaluatingRequestHandlerAdvice
for ignoring the exception or returning a compensation message.
When an exception is thrown from one of the sub-flows to the ScatterGatherHandler
, it is just re-thrown to upstream.
This way all other sub-flows will work for nothing and their replies are going to be ignored in the ScatterGatherHandler
.
This might be an expected behavior sometimes, but in most cases it would be better to handle the error in the particular sub-flow without impacting all others and the expectations in the gatherer.
Starting with version 5.1.3, the ScatterGatherHandler
is supplied with the errorChannelName
option.
It is populated to the errorChannel
header of the scatter message and is used in the when async error happens or can be used in the regular synchronous sub-flow for directly sending an error message.
The sample configuration below demonstrates async error handling by returning a compensation message:
@Bean
public IntegrationFlow scatterGatherAndExecutorChannelSubFlow(TaskExecutor taskExecutor) {
return f -> f
.scatterGather(
scatterer -> scatterer
.applySequence(true)
.recipientFlow(f1 -> f1.transform(p -> "Sub-flow#1"))
.recipientFlow(f2 -> f2
.channel(c -> c.executor(taskExecutor))
.transform(p -> {
throw new RuntimeException("Sub-flow#2");
})),
null,
s -> s.errorChannel("scatterGatherErrorChannel"));
}
@ServiceActivator(inputChannel = "scatterGatherErrorChannel")
public Message<?> processAsyncScatterError(MessagingException payload) {
return MessageBuilder.withPayload(payload.getCause().getCause())
.copyHeaders(payload.getFailedMessage().getHeaders())
.build();
}
To produce a proper reply, we have to copy headers (including replyChannel
and errorChannel
) from the failedMessage
of the MessagingException
that has been sent to the scatterGatherErrorChannel
by the MessagePublishingErrorHandler
.
This way the target exception is returned to the gatherer of the ScatterGatherHandler
for reply messages group completion.
Such an exception payload
can be filtered out in the MessageGroupProcessor
of the gatherer or processed other way downstream, after the scatter-gather endpoint.
Before sending scattering results to the gatherer, ScatterGatherHandler reinstates the request message headers, including reply and error channels if any.
This way errors from the AggregatingMessageHandler are going to be propagated to the caller, even if an async hand off is applied in scatter recipient subflows.
In this case a reasonable, finite gatherTimeout must be configured for the ScatterGatherHandler .
Otherwise it is going to be blocked waiting for a reply from the gatherer forever, by default.
|
8.8. Thread Barrier
Sometimes, we need to suspend a message flow thread until some other asynchronous event occurs. For example, consider an HTTP request that publishes a message to RabbitMQ. We might wish to not reply to the user until the RabbitMQ broker has issued an acknowledgment that the message was received.
In version 4.2, Spring Integration introduced the <barrier/>
component for this purpose.
The underlying MessageHandler
is the BarrierMessageHandler
.
This class also implements
MessageTriggerAction
, in which a message passed to the trigger()
method releases a corresponding thread in the
handleRequestMessage()
method (if present).
The suspended thread and trigger thread are correlated by invoking a CorrelationStrategy
on the messages.
When a message is sent to the input-channel
, the thread is suspended for up to timeout
milliseconds, waiting for
a corresponding trigger message.
The default correlation strategy uses the IntegrationMessageHeaderAccessor.CORRELATION_ID
header.
When a trigger message arrives with the same correlation, the thread is released.
The message sent to the output-channel
after release is constructed by using a MessageGroupProcessor
.
By default, the message is a Collection<?>
of the two payloads, and the headers are merged by using a
DefaultAggregatingMessageGroupProcessor
.
If the trigger() method is invoked first (or after the main thread times out), it is suspended for up to timeout waiting for the suspending message to arrive.
If you do not want to suspend the trigger thread, consider handing off to a TaskExecutor instead so that its thread is suspended instead.
|
The requires-reply
property determines the action to take if the suspended thread times out before the trigger message arrives.
By default, it is false
, which means the endpoint returns null
, the flow ends, and the thread returns to the
caller.
When true
, a ReplyRequiredException
is thrown.
You can call the trigger()
method programmatically (obtain the bean reference by using the name, barrier.handler
— where barrier
is the bean name of the barrier endpoint).
Alternatively, you can configure an <outbound-channel-adapter/>
to trigger the release.
| Only one thread can be suspended with the same correlation. The same correlation can be used multiple times but only once concurrently. An exception is thrown if a second thread arrives with the same correlation. |
The following example shows how to use a custom header for correlation:
<int:barrier id="barrier1" input-channel="in" output-channel="out"
correlation-strategy-expression="headers['myHeader']"
output-processor="myOutputProcessor"
discard-channel="lateTriggerChannel"
timeout="10000">
</int:barrier>
<int:outbound-channel-adapter channel="release" ref="barrier1.handler" method="trigger" />
Depending on which one has a message arrive first, either the thread sending a message to in
or the thread sending a message to release
waits for up to ten seconds until the other message arrives.
When the message is released, the out
channel is sent a message that combines the result of invoking the custom MessageGroupProcessor
bean, named myOutputProcessor
.
If the main thread times out and a trigger arrives later, you can configure a discard channel to which the late trigger is sent.
The following example shows the Java configuration to do so:
@Configuration
@EnableIntegration
public class Config {
@ServiceActivator(inputChannel="in")
@Bean
public BarrierMessageHandler barrier() {
BarrierMessageHandler barrier = new BarrierMessageHandler(10000);
barrier.setOutputChannel(out());
barrier.setDiscardChannel(lateTriggers());
return barrier;
}
@ServiceActivator (inputChannel="release")
@Bean
public MessageHandler releaser() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
barrier().trigger(message);
}
};
}
}
For an example of this component, see the barrier sample application.
9. Message Transformation
9.1. Transformer
Message transformers play a very important role in enabling the loose-coupling of message producers and message consumers.
Rather than requiring every message-producing component to know what type is expected by the next consumer, you can add transformers between those components.
Generic transformers, such as one that converts a String
to an XML Document, are also highly reusable.
For some systems, it may be best to provide a canonical data model, but Spring Integration’s general philosophy is not to require any particular format. Rather, for maximum flexibility, Spring Integration aims to provide the simplest possible model for extension. As with the other endpoint types, the use of declarative configuration in XML or Java annotations enables simple POJOs to be adapted for the role of message transformers. The rest of this chapter describes these configuration options.
| For the sake of maximizing flexibility, Spring does not require XML-based message payloads. Nevertheless, the framework does provide some convenient transformers for dealing with XML-based payloads if that is indeed the right choice for your application. For more information on those transformers, see XML Support - Dealing with XML Payloads. |
9.1.1. Configuring a Transformer with XML
The <transformer>
element is used to create a message-transforming endpoint.
In addition to input-channel
and output-channel
attributes, it requires a ` attribute`.
The ref
may either point to an object that contains the @Transformer
annotation on a single method (see Configuring a Transformer with Annotations), or it may be combined with an explicit method name value provided in the method
attribute.
<int:transformer id="testTransformer" ref="testTransformerBean" input-channel="inChannel"
method="transform" output-channel="outChannel"/>
<beans:bean id="testTransformerBean" class="org.foo.TestTransformer" />
Using a ref
attribute is generally recommended if the custom transformer handler implementation can be reused in other <transformer>
definitions.
However, if the custom transformer handler implementation should be scoped to a single definition of the <transformer>
, you can define an inner bean definition, as the following example shows:
<int:transformer id="testTransformer" input-channel="inChannel" method="transform"
output-channel="outChannel">
<beans:bean class="org.foo.TestTransformer"/>
</transformer>
Using both the ref attribute and an inner handler definition in the same <transformer> configuration is not allowed, as it creates an ambiguous condition and results in an exception being thrown.
|
If the ref attribute references a bean that extends AbstractMessageProducingHandler (such as transformers provided by the framework itself), the configuration is optimized by injecting the output channel into the handler directly.
In this case, each ref must be to a separate bean instance (or a prototype -scoped bean) or use the inner <bean/> configuration type.
If you inadvertently reference the same message handler from multiple beans, you get a configuration exception.
|
When using a POJO, the method that is used for transformation may expect either the Message
type or the payload type of inbound messages.
It may also accept message header values either individually or as a full map by using the @Header
and @Headers
parameter annotations, respectively.
The return value of the method can be any type.
If the return value is itself a Message
, that is passed along to the transformer’s output channel.
As of Spring Integration 2.0, a message transformer’s transformation method can no longer return null
.
Returning null
results in an exception, because a message transformer should always be expected to transform each source message into a valid target message.
In other words, a message transformer should not be used as a message filter, because there is a dedicated <filter>
option for that.
However, if you do need this type of behavior (where a component might return null
and that should not be considered an error), you could use a service activator.
Its requires-reply
value is false
by default, but that can be set to true
in order to have exceptions thrown for null
return values, as with the transformer.
9.1.2. Transformers and Spring Expression Language (SpEL)
Like routers, aggregators, and other components, as of Spring Integration 2.0, transformers can also benefit from SpEL support whenever transformation logic is relatively simple. The following example shows how to use a SpEL expression:
<int:transformer input-channel="inChannel"
output-channel="outChannel"
expression="payload.toUpperCase() + '- [' + T(java.lang.System).currentTimeMillis() + ']'"/>
The preceding example transforms the payload without writing a custom transformer.
Our payload (assumed to be a String
) is upper-cased, concatenated with the current timestamp, and has some formatting applied.
9.1.3. Common Transformers
Spring Integration provides a few transformer implementations.
Object-to-String Transformer
Because it is fairly common to use the toString()
representation of an Object
, Spring Integration provides an ObjectToStringTransformer
whose output is a Message
with a String payload
.
That String
is the result of invoking the toString()
operation on the inbound Message’s payload.
The following example shows how to declare an instance of the object-to-string transformer:
<int:object-to-string-transformer input-channel="in" output-channel="out"/>
A potential use for this transformer would be sending some arbitrary object to the 'outbound-channel-adapter' in the file
namespace.
Whereas that channel adapter only supports String
, byte-array, or java.io.File
payloads by default, adding this transformer immediately before the adapter handles the necessary conversion.
That works fine as long as the result of the toString()
call is what you want to be written to the file.
Otherwise, you can provide a custom POJO-based transformer by using the generic 'transformer' element shown previously.
| When debugging, this transformer is not typically necessary, since the 'logging-channel-adapter' is capable of logging the message payload. See Wire Tap for more detail. |
|
The object-to-string transformer is very simple.
It invokes
For more sophistication (such as selection of the charset dynamically, at runtime), you can use a SpEL expression-based transformer instead, as the following example shows:
|
If you need to serialize an Object
to a byte array or deserialize a byte array back into an Object
, Spring Integration provides symmetrical serialization transformers.
These use standard Java serialization by default, but you can provide an implementation of Spring 3.0’s serializer or seserializer strategies by using the 'serializer' and 'deserializer' attributes, respectively.
The following example shows to use Spring’s serializer and deserializer:
<int:payload-serializing-transformer input-channel="objectsIn" output-channel="bytesOut"/>
<int:payload-deserializing-transformer input-channel="bytesIn" output-channel="objectsOut"
white-list="com.mycom.*,com.yourcom.*"/>
When deserializing data from untrusted sources, you should consider adding a white-list of package and class patterns.
By default, all classes are deserialized.
|
Object
-to-Map
and Map
-to-Object
Transformers
Spring Integration also provides Object
-to-Map
and Map
-to-Object
transformers, which use the JSON to serialize and de-serialize the object graphs.
The object hierarchy is introspected to the most primitive types (String
, int
, and so on).
The path to this type is described with SpEL, which becomes the key
in the transformed Map
.
The primitive type becomes the value.
Consider the following example:
public class Parent{
private Child child;
private String name;
// setters and getters are omitted
}
public class Child{
private String name;
private List<String> nickNames;
// setters and getters are omitted
}
The two classes in the preceding example are transformed to the following Map
:
{person.name=George, person.child.name=Jenna, person.child.nickNames[0]=Jen ...}
The JSON-based Map
lets you describe the object structure without sharing the actual types, which lets you restore and rebuild the object graph into a differently typed object graph, as long as you maintain the structure.
For example, the preceding structure could be restored back to the following object graph by using the Map
-to-Object
transformer:
public class Father {
private Kid child;
private String name;
// setters and getters are omitted
}
public class Kid {
private String name;
private List<String> nickNames;
// setters and getters are omitted
}
If you need to create a “structured” map, you can provide the 'flatten' attribute.
The default is 'true'.
If you set it to 'false', the structure is a Map
of Map
objects.
Consider the following example:
public class Parent {
private Child child;
private String name;
// setters and getters are omitted
}
public class Child {
private String name;
private List<String> nickNames;
// setters and getters are omitted
}
The two classes in the preceding example are transformed to the following Map
:
{name=George, child={name=Jenna, nickNames=[Bimbo, ...]}}
To configure these transformers, Spring Integration provides namespace support for Object-to-Map, as the following example shows:
<int:object-to-map-transformer input-channel="directInput" output-channel="output"/>
You can also set the flatten
attribute to false, as follows:
<int:object-to-map-transformer input-channel="directInput" output-channel="output" flatten="false"/>
Spring Integration provides namespace support for Map-to-Object, as the following example shows:
<int:map-to-object-transformer input-channel="input"
output-channel="output"
type="org.something.Person"/>
Alterately, you could use a ref
attribute and a prototype-scoped bean, as the following example shows:
<int:map-to-object-transformer input-channel="inputA"
output-channel="outputA"
ref="person"/>
<bean id="person" class="org.something.Person" scope="prototype"/>
The 'ref' and 'type' attributes are mutually exclusive.
Also, if you use the 'ref' attribute, you must point to a 'prototype' scoped bean.
Otherwise, a BeanCreationException is thrown.
|
Starting with version 5.0, you can supply the ObjectToMapTransformer
with a customized JsonObjectMapper
— for when you need special formats for dates or nulls for empty collections (and other uses).
See JSON Transformers for more information about JsonObjectMapper
implementations.
Stream Transformer
The StreamTransformer
transforms InputStream
payloads to a byte[]
( or a String
if a charset
is provided).
The following example shows how to use the stream-tansformer
element in XML:
<int:stream-transformer input-channel="directInput" output-channel="output"/> <!-- byte[] -->
<int:stream-transformer id="withCharset" charset="UTF-8"
input-channel="charsetChannel" output-channel="output"/> <!-- String -->
The following example shows how to use the StreamTransformer
class and the @Transformer
annotation to configure a stream transformer in Java:
@Bean
@Transformer(inputChannel = "stream", outputChannel = "data")
public StreamTransformer streamToBytes() {
return new StreamTransformer(); // transforms to byte[]
}
@Bean
@Transformer(inputChannel = "stream", outputChannel = "data")
public StreamTransformer streamToString() {
return new StreamTransformer("UTF-8"); // transforms to String
}
JSON Transformers
Spring Integration provides Object-to-JSON and JSON-to-Object transformers. The following pair of examples show how to declare them in XML:
<int:object-to-json-transformer input-channel="objectMapperInput"/>
<int:json-to-object-transformer input-channel="objectMapperInput"
type="foo.MyDomainObject"/>
By default, the transformers in the preceding listing use a vanilla JsonObjectMapper
.
It is based on an implementation from the classpath.
You can provide your own custom JsonObjectMapper
implementation with appropriate options or based on a required library (such as GSON), as the following example shows:
<int:json-to-object-transformer input-channel="objectMapperInput"
type="something.MyDomainObject" object-mapper="customObjectMapper"/>
|
Beginning with version 3.0, the Note, |
|
If you have requirements to use both Jackson and Boon in the same application, keep in mind that, before version 3.0, the JSON transformers used only Jackson 1.x. From 4.1 on, the framework selects Jackson 2 by default, preferring it to the Boon implementation if both are on the classpath. Jackson 1.x is no longer supported by the framework internally. However, you can still use it within your code by including the necessary library. To avoid unexpected issues with JSON mapping features when you use annotations, you may need to apply annotations from both Jackson and Boon on domain classes, as the following example shows:
|
You may wish to consider using a FactoryBean
or a factory method to create the JsonObjectMapper
with the required characteristics.
The following example shows how to use such a factory:
public class ObjectMapperFactory {
public static Jackson2JsonObjectMapper getMapper() {
ObjectMapper mapper = new ObjectMapper();
mapper.configure(JsonParser.Feature.ALLOW_COMMENTS, true);
return new Jackson2JsonObjectMapper(mapper);
}
}
The following example shows how to do the same thing in XML
<bean id="customObjectMapper" class="something.ObjectMapperFactory"
factory-method="getMapper"/>
|
Beginning with version 2.2, the It you wish to set the |
Beginning with version 3.0, the ObjectToJsonTransformer
adds headers, reflecting the source type, to the message.
Similarly, the JsonToObjectTransformer
can use those type headers when converting the JSON to an object.
These headers are mapped in the AMQP adapters so that they are entirely compatible with the Spring-AMQP JsonMessageConverter
.
This enables the following flows to work without any special configuration:
-
…→amqp-outbound-adapter---→
-
---→amqp-inbound-adapter→json-to-object-transformer→…
Where the outbound adapter is configured with a
JsonMessageConverter
and the inbound adapter uses the defaultSimpleMessageConverter
. -
…→object-to-json-transformer→amqp-outbound-adapter---→
-
---→amqp-inbound-adapter→…
Where the outbound adapter is configured with a
SimpleMessageConverter
and the inbound adapter uses the defaultJsonMessageConverter
. -
…→object-to-json-transformer→amqp-outbound-adapter---→
-
---→amqp-inbound-adapter→json-to-object-transformer→
Where both adapters are configured with a
SimpleMessageConverter
.
When using the headers to determine the type, you should not provide a class attribute, because it takes precedence over the headers.
|
In addition to JSON Transformers, Spring Integration provides a built-in #jsonPath
SpEL function for use in expressions.
For more information see Spring Expression Language (SpEL).
Since version 3.0, Spring Integration also provides a built-in #xpath
SpEL function for use in expressions.
For more information see #xpath SpEL Function.
Beginning with version 4.0, the ObjectToJsonTransformer
supports the resultType
property, to specify the node JSON representation.
The result node tree representation depends on the implementation of the provided JsonObjectMapper
.
By default, the ObjectToJsonTransformer
uses a Jackson2JsonObjectMapper
and delegates the conversion of the object to the node tree to the ObjectMapper#valueToTree
method.
The node JSON representation provides efficiency for using the JsonPropertyAccessor
when the downstream message flow uses SpEL expressions with access to the properties of the JSON data.
See Property Accessors for more information.
When using Boon, the NODE
representation is a Map<String, Object>
Beginning with version 5.1, the resultType
can be configured as BYTES
to produce a message with the byte[]
payload for convenience when working with downstream handlers which operate with this data type.
9.1.4. Configuring a Transformer with Annotations
You can add the @Transformer
annotation to methods that expect either the Message
type or the message payload type.
The return value is handled in the exact same way as described earlier in the section describing the <transformer>
element.
The following example shows how to use the @Transformer
annotation to transform a String
into an Order
:
@Transformer
Order generateOrder(String productId) {
return new Order(productId);
}
Transformer methods can also accept the @Header
and @Headers
annotations, as documented in Annotation Support
.
The following examples shows how to use the @Header
annotation:
@Transformer
Order generateOrder(String productId, @Header("customerName") String customer) {
return new Order(productId, customer);
}
See also Advising Endpoints Using Annotations.
9.1.5. Header Filter
Sometimes, your transformation use case might be as simple as removing a few headers. For such a use case, Spring Integration provides a header filter that lets you specify certain header names that should be removed from the output message (for example, removing headers for security reasons or a value that was needed only temporarily). Basically, the header filter is the opposite of the header enricher. The latter is discussed in Header Enricher. The following example defines a header filter:
<int:header-filter input-channel="inputChannel"
output-channel="outputChannel" header-names="lastName, state"/>
As you can see, configuration of a header filter is quite simple.
It is a typical endpoint with input and output channels and a header-names
attribute.
That attribute accepts the names of the headers (delimited by commas if there are multiple) that need to be removed.
So, in the preceding example, the headers named 'lastName' and 'state' are not present on the outbound message.
9.1.6. Codec-Based Transformers
See Codec.
9.2. Content Enricher
At times, you may have a requirement to enhance a request with more information than was provided by the target system. The data enricher pattern describes various scenarios as well as the component (Enricher) that lets you address such requirements.
The Spring Integration Core
module includes two enrichers:
It also includes three adapter-specific header enrichers:
See the adapter-specific sections of this reference manual to learn more about those adapters.
For more information regarding expressions support, see Spring Expression Language (SpEL).
9.2.1. Header Enricher
If you need do nothing more than add headers to a message and the headers are not dynamically determined by the message content, referencing a custom implementation of a transformer may be overkill.
For that reason, Spring Integration provides support for the header enricher pattern.
It is exposed through the <header-enricher>
element.
The following example shows how to use it:
<int:header-enricher input-channel="in" output-channel="out">
<int:header name="foo" value="123"/>
<int:header name="bar" ref="someBean"/>
</int:header-enricher>
The header enricher also provides helpful sub-elements to set well known header names, as the following example shows:
<int:header-enricher input-channel="in" output-channel="out">
<int:error-channel ref="applicationErrorChannel"/>
<int:reply-channel ref="quoteReplyChannel"/>
<int:correlation-id value="123"/>
<int:priority value="HIGHEST"/>
<routing-slip value="channel1; routingSlipRoutingStrategy; request.headers[myRoutingSlipChannel]"/>
<int:header name="bar" ref="someBean"/>
</int:header-enricher>
The preceding configuration shows that, for well known headers (such as errorChannel
, correlationId
, priority
, replyChannel
, routing-slip
, and others), instead of using generic <header>
sub-elements where you would have to provide both header 'name' and 'value', you can use convenient sub-elements to set those values directly.
Starting with version 4.1, the header enricher provides a routing-slip
sub-element.
See Routing Slip for more information.
POJO Support
Often, a header value cannot be defined statically and has to be determined dynamically based on some content in the message.
That is why the header enricher lets you also specify a bean reference by using the ref
and method
attributes.
The specified method calculates the header value.
Consider the following configuration and a bean with a method that modifies a String
:
<int:header-enricher input-channel="in" output-channel="out">
<int:header name="something" method="computeValue" ref="myBean"/>
</int:header-enricher>
<bean id="myBean" class="thing1.thing2.MyBean"/>
public class MyBean {
public String computeValue(String payload){
return payload.toUpperCase() + "_US";
}
}
You can also configure your POJO as an inner bean, as the following example shows:
<int:header-enricher input-channel="inputChannel" output-channel="outputChannel">
<int:header name="some_header">
<bean class="org.MyEnricher"/>
</int:header>
</int:header-enricher>
You can similarly point to a Groovy script, as the following example shows:
<int:header-enricher input-channel="inputChannel" output-channel="outputChannel">
<int:header name="some_header">
<int-groovy:script location="org/SampleGroovyHeaderEnricher.groovy"/>
</int:header>
</int:header-enricher>
SpEL Support
In Spring Integration 2.0, we introduced the convenience of the Spring Expression Language (SpEL) to help configure many different components. The header enricher is one of them. Look again at the POJO example shown earlier. You can see that the computation logic to determine the header value is pretty simple. A natural question would be: "Is there an even simpler way to accomplish this?". That is where SpEL shows its true power. Consider the following example:
<int:header-enricher input-channel="in" output-channel="out">
<int:header name="foo" expression="payload.toUpperCase() + '_US'"/>
</int:header-enricher>
By using SpEL for such simple cases, you no longer have to provide a separate class and configure it in the application context.
All you need do is configured the expression
attribute with a valid SpEL expression.
The 'payload' and 'headers' variables are bound to the SpEL evaluation context, giving you full access to the incoming message.
Configuring a Header Enricher with Java Configuration
The following two examples show how to use Java Configuration for header enrichers:
@Bean
@Transformer(inputChannel = "enrichHeadersChannel", outputChannel = "emailChannel")
public HeaderEnricher enrichHeaders() {
Map<String, ? extends HeaderValueMessageProcessor<?>> headersToAdd =
Collections.singletonMap("emailUrl",
new StaticHeaderValueMessageProcessor<>(this.imapUrl));
HeaderEnricher enricher = new HeaderEnricher(headersToAdd);
return enricher;
}
@Bean
@Transformer(inputChannel="enrichHeadersChannel", outputChannel="emailChannel")
public HeaderEnricher enrichHeaders() {
Map<String, HeaderValueMessageProcessor<?>> headersToAdd = new HashMap<>();
headersToAdd.put("emailUrl", new StaticHeaderValueMessageProcessor<String>(this.imapUrl));
Expression expression = new SpelExpressionParser().parseExpression("payload.from[0].toString()");
headersToAdd.put("from",
new ExpressionEvaluatingHeaderValueMessageProcessor<>(expression, String.class));
HeaderEnricher enricher = new HeaderEnricher(headersToAdd);
return enricher;
}
The first example adds a single literal header. The second example adds two headers, a literal header and one based on a SpEL expression.
Configuring a Header Enricher with the Java DSL
The following example shows Java DSL Configuration for a header enricher:
@Bean
public IntegrationFlow enrichHeadersInFlow() {
return f -> f
...
.enrichHeaders(h -> h.header("emailUrl", this.emailUrl)
.headerExpression("from", "payload.from[0].toString()"))
.handle(...);
}
Header Channel Registry
Starting with Spring Integration 3.0, a new sub-element <int:header-channels-to-string/>
is available.
It has no attributes.
This new sub-element converts existing replyChannel
and errorChannel
headers (when they are a MessageChannel
) to a String
and stores the channels in a registry for later resolution, when it is time to send a reply or handle an error.
This is useful for cases where the headers might be lost — for example, when serializing a message into a message store or when transporting the message over JMS.
If the header does not already exist or it is not a MessageChannel
, no changes are made.
Using this functionality requires the presence of a HeaderChannelRegistry
bean.
By default, the framework creates a DefaultHeaderChannelRegistry
with the default expiry (60 seconds).
Channels are removed from the registry after this time.
To change this behavior, define a bean with an id
of integrationHeaderChannelRegistry
and configure the required default delay by using a constructor argument (in milliseconds).
Since version 4.1, you can set a property called removeOnGet
to true
on the <bean/>
definition, and the mapping entry is removed immediately on first use.
This might be useful in a high-volume environment and when the channel is only used once, rather than waiting for the reaper to remove it.
The HeaderChannelRegistry
has a size()
method to determine the current size of the registry.
The runReaper()
method cancels the current scheduled task and runs the reaper immediately.
The task is then scheduled to run again based on the current delay.
These methods can be invoked directly by getting a reference to the registry, or you can send a message with, for example, the following content to a control bus:
"@integrationHeaderChannelRegistry.runReaper()"
This sub-element is a convenience, and is the equivalent of specifying the following configuration:
<int:reply-channel
expression="@integrationHeaderChannelRegistry.channelToChannelName(headers.replyChannel)"
overwrite="true" />
<int:error-channel
expression="@integrationHeaderChannelRegistry.channelToChannelName(headers.errorChannel)"
overwrite="true" />
Starting with version 4.1, you can now override the registry’s configured reaper delay so that the the channel mapping is retained for at least the specified time, regardless of the reaper delay. The following example shows how to do so:
<int:header-enricher input-channel="inputTtl" output-channel="next">
<int:header-channels-to-string time-to-live-expression="120000" />
</int:header-enricher>
<int:header-enricher input-channel="inputCustomTtl" output-channel="next">
<int:header-channels-to-string
time-to-live-expression="headers['channelTTL'] ?: 120000" />
</int:header-enricher>
In the first case, the time to live for every header channel mapping will be two minutes. In the second case, the time to live is specified in the message header and uses an Elvis operator to use two minutes if there is no header.
9.2.2. Payload Enricher
In certain situations, the header enricher, as discussed earlier, may not be sufficient and payloads themselves may have to be enriched with additional information. For example, order messages that enter the Spring Integration messaging system have to look up the order’s customer based on the provided customer number and then enrich the original payload with that information.
Spring Integration 2.1 introduced the payload enricher.
The payload enricher defines an endpoint that passes a Message
to the exposed request channel and then expects a reply message.
The reply message then becomes the root object for evaluation of expressions to enrich the target payload.
The payload enricher provides full XML namespace support through the enricher
element.
In order to send request messages, the payload enricher has a request-channel
attribute that lets you dispatch messages to a request channel.
Basically, by defining the request channel, the payload enricher acts as a gateway, waiting for the message sent to the request channel to return. The enricher then augments the message’s payload with the data provided by the reply message.
When sending messages to the request channel, you also have the option to send only a subset of the original payload by using the request-payload-expression
attribute.
The enriching of payloads is configured through SpEL expressions, providing a maximum degree of flexibility.
Therefore, you can not only enrich payloads with direct values from the reply channel’s Message
, but you can use SpEL expressions to extract a subset from that message or to apply additional inline transformations, letting you further manipulate the data.
If you need only to enrich payloads with static values, you need not provide the request-channel
attribute.
| Enrichers are a variant of transformers. In many cases, you could use a payload enricher or a generic transformer implementation to add additional data to your message payloads. You should familiarize yourself with all transformation-capable components that are provided by Spring Integration and carefully select the implementation that semantically fits your business case best. |
Configuration
The following example shows all available configuration options for the payload enricher:
<int:enricher request-channel="" (1)
auto-startup="true" (2)
id="" (3)
order="" (4)
output-channel="" (5)
request-payload-expression="" (6)
reply-channel="" (7)
error-channel="" (8)
send-timeout="" (9)
should-clone-payload="false"> (10)
<int:poller></int:poller> (11)
<int:property name="" expression="" null-result-expression="'Could not determine the name'"/> (12)
<int:property name="" value="23" type="java.lang.Integer" null-result-expression="'0'"/>
<int:header name="" expression="" null-result-expression=""/> (13)
<int:header name="" value="" overwrite="" type="" null-result-expression=""/>
</int:enricher>
| 1 | Channel to which a message is sent to get the data to use for enrichment. Optional. |
| 2 | Lifecycle attribute signaling whether this component should be started during the application context startup. Defaults to true. Optional. |
| 3 | ID of the underlying bean definition, which is either an EventDrivenConsumer or a PollingConsumer .
Optional. |
| 4 | Specifies the order for invocation when this endpoint is connected as a subscriber to a channel. This is particularly relevant when that channel is using a “failover” dispatching strategy. It has no effect when this endpoint is itself a polling consumer for a channel with a queue. Optional. |
| 5 | Identifies the message channel where a message is sent after it is being processed by this endpoint. Optional. |
| 6 | By default, the original message’s payload is used as payload that is sent to the request-channel .
By specifying a SpEL expression as the value for the request-payload-expression attribute, you can use a subset of the original payload, a header value, or any other resolvable SpEL expression as the basis for the payload that is sent to the request-channel.
For the expression evaluation, the full message is available as the 'root object'.
For instance, the following SpEL expressions (among others) are possible:
|
| 7 | Channel where a reply message is expected. This is optional. Typically, the auto-generated temporary reply channel suffices. Optional. |
| 8 | The channel to which an ErrorMessage is sent if an Exception occurs downstream of the request-channel .
This enables you to return an alternative object to use for enrichment.
If it is not set, an Exception is thrown to the caller.
Optional. |
| 9 | Maximum amount of time in milliseconds to wait when sending a message to the channel, if the channel might block.
For example, a queue channel can block until space is available, if its maximum capacity has been reached.
Internally, the send timeout is set on the MessagingTemplate and ultimately applied when invoking the send operation on the MessageChannel .
By default, the send timeout is set to '-1', which can cause the send operation on the MessageChannel , depending on the implementation, to block indefinitely.
Optional. |
| 10 | Boolean value indicating whether any payload that implements Cloneable should be cloned prior to sending the message to the request channel for acquiring the enriching data.
The cloned version would be used as the target payload for the ultimate reply.
The default is false .
Optional. |
| 11 | Lets you configure a message poller if this endpoint is a polling consumer. Optional. |
| 12 | Each property sub-element provides the name of a property (through the mandatory name attribute).
That property should be settable on the target payload instance.
Exactly one of the value or expression attributes must be provided as well — the former for a literal value to set and the latter for a SpEL expression to be evaluated.
The root object of the evaluation context is the message that was returned from the flow initiated by this enricher — the input message if there is no request channel or the application context (using the<EMAIL_ADDRESS>SpEL syntax).
Starting with version 4.0, when specifying a value attribute, you can also specify an optional type attribute.
When the destination is a typed setter method, the framework coerces the value appropriately (as long as a PropertyEditor ) exists to handle the conversion.
If, however, the target payload is a Map , the entry is populated with the value without conversion.
The type attribute lets you, for example, convert a String containing a number to an Integer value in the target payload.
Starting with version 4.1, you can also specify an optional null-result-expression attribute.
When the enricher returns null, it is evaluated, and the output of the evaluation is returned instead. |
| 13 | Each header sub-element provides the name of a message header (through the mandatory name attribute).
Exactly one of the value or expression attributes must also be provided — the former for a literal value to set and the latter for a SpEL expression to be evaluated.
The root object of the evaluation context is the message that was returned from the flow initiated by this enricher — the input message if there is no request channel or the application context (using the<EMAIL_ADDRESS>SpEL syntax).
Note that, similarly to the <header-enricher> , the <enricher> element’s header element has type and overwrite attributes.
However, a key difference is that, with the <enricher> , the overwrite attribute is true by default, to be consistent with the <enricher> element’s <property> sub-element.
Starting with version 4.1, you can also specify an optional null-result-expression attribute.
When the enricher returns null, it is evaluated, and the output of the evaluation is returned instead. |
Examples
This section contains several examples of using a payload enricher in various situations.
| The code samples shown here are part of the Spring Integration Samples project. See Spring Integration Samples. |
In the following example, a User
object is passed as the payload of the Message
:
<int:enricher id="findUserEnricher"
input-channel="findUserEnricherChannel"
request-channel="findUserServiceChannel">
<int:property name="email" expression="payload.email"/>
<int:property name="password" expression="payload.password"/>
</int:enricher>
The User
has several properties, but only the username
is set initially.
The enricher’s request-channel
attribute is configured to pass the User
to the findUserServiceChannel
.
Through the implicitly set reply-channel
, a User
object is returned and, by using the property
sub-element, properties from the reply are extracted and used to enrich the original payload.
How Do I Pass Only a Subset of Data to the Request Channel?
When using a request-payload-expression
attribute, a single property of the payload instead of the full message can be passed on to the request channel.
In the following example, the username property is passed on to the request channel:
<int:enricher id="findUserByUsernameEnricher"
input-channel="findUserByUsernameEnricherChannel"
request-channel="findUserByUsernameServiceChannel"
request-payload-expression="payload.username">
<int:property name="email" expression="payload.email"/>
<int:property name="password" expression="payload.password"/>
</int:enricher>
Keep in mind that, although only the username is passed, the resulting message to the request channel contains the full set of MessageHeaders
.
How Can I Enrich Payloads that Consist of Collection Data?
In the following example, instead of a User
object, a Map
is passed in:
<int:enricher id="findUserWithMapEnricher"
input-channel="findUserWithMapEnricherChannel"
request-channel="findUserByUsernameServiceChannel"
request-payload-expression="payload.username">
<int:property name="user" expression="payload"/>
</int:enricher>
The Map
contains the username under the username
map key.
Only the username
is passed on to the request channel.
The reply contains a full User
object, which is ultimately added to the Map
under the user
key.
How Can I Enrich Payloads with Static Information without Using a Request Channel?
The following example does not use a request channel at all but solely enriches the message’s payload with static values:
<int:enricher id="userEnricher"
input-channel="input">
<int:property name="user.updateDate" expression="new java.util.Date()"/>
<int:property name="user.firstName" value="William"/>
<int:property name="user.lastName" value="Shakespeare"/>
<int:property name="user.age" value="42"/>
</int:enricher>
Note that the word, 'static', is used loosely here. You can still use SpEL expressions for setting those values.
9.3. Claim Check
In earlier sections, we covered several content enricher components that can help you deal with situations where a message is missing a piece of data. We also discussed content filtering, which lets you remove data items from a message. However, there are times when we want to hide data temporarily. For example, in a distributed system, we may receive a message with a very large payload. Some intermittent message processing steps may not need access to this payload and some may only need to access certain headers, so carrying the large message payload through each processing step may cause performance degradation, may produce a security risk, and may make debugging more difficult.
The store in library (or claim check) pattern describes a mechanism that lets you store data in a well known place while maintaining only a pointer (a claim check) to where that data is located. You can pass that pointer around as the payload of a new message, thereby letting any component within the message flow get the actual data as soon as it needs it. This approach is very similar to the certified mail process, where you get a claim check in your mailbox and then have to go to the post office to claim your actual package. It is also the same idea as baggage claim after a flight or in a hotel.
Spring Integration provides two types of claim check transformers:
-
Incoming Claim Check Transformer
-
Outgoing Claim Check Transformer
Convenient namespace-based mechanisms are available to configure them.
9.3.1. Incoming Claim Check Transformer
An incoming claim check transformer transforms an incoming message by storing it in the message store identified by its message-store
attribute.
The following example defines an incoming claim check transformer:
<int:claim-check-in id="checkin"
input-channel="checkinChannel"
message-store="testMessageStore"
output-channel="output"/>
In the preceding configuration, the message that is received on the input-channel
is persisted to the message store identified with the message-store
attribute and indexed with a generated ID.
That ID is the claim check for that message.
The claim check also becomes the payload of the new (transformed) message that is sent to the output-channel
.
Now, assume that at some point you do need access to the actual message. You can access the message store manually and get the contents of the message, or you can use the same approach (creating a transformer) except that now you transform the Claim Check to the actual message by using an outgoing claim check transformer.
The following listing provides an overview of all available parameters of an incoming claim check transformer:
<int:claim-check-in auto-startup="true" (1)
id="" (2)
input-channel="" (3)
message-store="messageStore" (4)
order="" (5)
output-channel="" (6)
send-timeout=""> (7)
<int:poller></int:poller> (8)
</int:claim-check-in>
| 1 | Lifecycle attribute signaling whether this component should be started during application context startup.
It defaults to true .
This attribute is not available inside a Chain element.
Optional. |
| 2 | ID identifying the underlying bean definition (MessageTransformingHandler ).
This attribute is not available inside a Chain element.
Optional. |
| 3 | The receiving message channel of this endpoint.
This attribute is not available inside a Chain element.
Optional. |
| 4 | Reference to the MessageStore to be used by this claim check transformer.
If not specified, the default reference is to a bean named messageStore .
Optional. |
| 5 | Specifies the order for invocation when this endpoint is connected as a subscriber to a channel.
This is particularly relevant when that channel uses a failover dispatching strategy.
It has no effect when this endpoint is itself a polling consumer for a channel with a queue.
This attribute is not available inside a Chain element.
Optional. |
| 6 | Identifies the message channel where the message is sent after being processed by this endpoint.
This attribute is not available inside a Chain element.
Optional. |
| 7 | Specifies the maximum amount of time (in milliseconds) to wait when sending a reply message to the output channel.
Defaults to -1 — blocking indefinitely.
This attribute is not available inside a Chain element.
Optional. |
| 8 | Defines a poller.
This element is not available inside a Chain element.
Optional. |
9.3.2. Outgoing Claim Check Transformer
An outgoing claim check transformer lets you transform a message with a claim check payload into a message with the original content as its payload.
<int:claim-check-out id="checkout"
input-channel="checkoutChannel"
message-store="testMessageStore"
output-channel="output"/>
In the preceding configuration, the message received on the input-channel
should have a claim check as its payload.
The outgoing claim check transformer transforms it into a message with the original payload by querying the message store for a message identified by the provided claim check.
It then sends the newly checked-out message to the output-channel
.
The following listing provides an overview of all available parameters of an outgoing claim check transformer:
<int:claim-check-out auto-startup="true" (1)
id="" (2)
input-channel="" (3)
message-store="messageStore" (4)
order="" (5)
output-channel="" (6)
remove-message="false" (7)
send-timeout=""> (8)
<int:poller></int:poller> (9)
</int:claim-check-out>
| 1 | Lifecycle attribute signaling whether this component should be started during application context startup.
It defaults to true .
This attribute is not available inside a Chain element.
Optional. |
| 2 | ID identifying the underlying bean definition (MessageTransformingHandler ).
This attribute is not available inside a Chain element.
Optional. |
| 3 | The receiving message channel of this endpoint.
This attribute is not available inside a Chain element.
Optional. |
| 4 | Reference to the MessageStore to be used by this claim check transformer.
If not specified, the default reference is to a bean named messageStore .
Optional. |
| 5 | Specifies the order for invocation when this endpoint is connected as a subscriber to a channel.
This is particularly relevant when that channel is using a failover dispatching strategy.
It has no effect when this endpoint is itself a polling consumer for a channel with a queue.
This attribute is not available inside a Chain element.
Optional. |
| 6 | Identifies the message channel where the message is sent after being processed by this endpoint.
This attribute is not available inside a Chain element.
Optional. |
| 7 | If set to true , the message is removed from the MessageStore by this transformer.
This setting is useful when Message can be “claimed” only once.
It defaults to false .
Optional. |
| 8 | Specifies the maximum amount of time (in milliseconds) to wait when sending a reply message to the output channel.
It defaults to -1 — blocking indefinitely.
This attribute is not available inside a Chain element.
Optional. |
| 9 | Defines a poller.
This element is not available inside a Chain element.
Optional. |
9.3.3. Claim Once
Sometimes, a particular message must be claimed only once.
As an analogy, consider process of handling airplane luggage.

Once the luggage has been claimed, it can not be claimed again without first checking it back in.
To accommodate such cases, we introduced a remove-message
boolean attribute on the claim-check-out
transformer.
This attribute is set to false
by default.
However, if set to true
, the claimed message is removed from the MessageStore
so that it cannot be claimed again.
This feature has an impact in terms of storage space, especially in the case of the in-memory Map
-based SimpleMessageStore
, where failing to remove messages could ultimately lead to an OutOfMemoryException
.
Therefore, if you do not expect multiple claims to be made, we recommend that you set the remove-message
attribute’s value to true
.
The following example show how to use the remove-message
attribute:
<int:claim-check-out id="checkout"
input-channel="checkoutChannel"
message-store="testMessageStore"
output-channel="output"
remove-message="true"/>
9.3.4. A Word on Message Store
Although we rarely care about the details of the claim checks (as long as they work), you should know that the current implementation of the actual claim check (the pointer) in Spring Integration uses a UUID to ensure uniqueness.
org.springframework.integration.store.MessageStore
is a strategy interface for storing and retrieving messages.
Spring Integration provides two convenient implementations of it:
-
SimpleMessageStore
: An in-memory,Map
-based implementation (the default, good for testing) -
JdbcMessageStore
: An implementation that uses a relational database over JDBC
9.4. Codec
Version 4.2 of Spring Integration introduced the Codec
abstraction.
Codecs encode and decode objects to and from byte[]
.
They offer an alternative to Java serialization.
One advantage is that, typically, objects need not implement Serializable
.
We provide one implementation that uses Kryo for serialization, but you can provide your own implementation for use in any of the following components:
-
EncodingPayloadTransformer
-
DecodingTransformer
-
CodecMessageConverter
9.4.1. EncodingPayloadTransformer
This transformer encodes the payload to a byte[]
by using the codec.
It does not affect message headers.
See the Javadoc for more information.
9.4.2. DecodingTransformer
This transformer decodes a byte[]
by using the codec.
It needs to be configured with the Class
to which the object should be decoded (or an expression that resolves to a Class
).
If the resulting object is a Message<?>
, inbound headers are not retained.
See the Javadoc for more information.
9.4.3. CodecMessageConverter
Certain endpoints (such as TCP and Redis) have no concept of message headers.
They support the use of a MessageConverter
, and the CodecMessageConverter
can be used to convert a message to or from a byte[]
for
transmission.
See the Javadoc for more information.
9.4.4. Kryo
Currently, this is the only implementation of Codec
, and it provides two kinds of Codec
:
-
PojoCodec
: Used in the transformers -
MessageCodec
: Used in theCodecMessageConverter
The framework provides several custom serializers:
-
FileSerializer
-
MessageHeadersSerializer
-
MutableMessageHeadersSerializer
The first can be used with the PojoCodec
by initializing it with the FileKryoRegistrar
.
The second and third are used with the MessageCodec
, which is initialized with the MessageKryoRegistrar
.
Customizing Kryo
By default, Kryo delegates unknown Java types to its FieldSerializer
.
Kryo also registers default serializers for each primitive type, along with String
, Collection
, and Map
.
FieldSerializer
uses reflection to navigate the object graph.
A more efficient approach is to implement a custom serializer that is aware of the object’s structure and can directly serialize selected primitive fields.
The following example shows such a serializer:
public class AddressSerializer extends Serializer<Address> {
@Override
public void write(Kryo kryo, Output output, Address address) {
output.writeString(address.getStreet());
output.writeString(address.getCity());
output.writeString(address.getCountry());
}
@Override
public Address read(Kryo kryo, Input input, Class<Address> type) {
return new Address(input.readString(), input.readString(), input.readString());
}
}
The Serializer
interface exposes Kryo
, Input
, and Output
, which provide complete control over which fields are included and other internal settings, as described in the Kryo documentation.
| When registering your custom serializer, you need a registration ID. The registration IDs are arbitrary. However, in our case, the IDs must be explicitly defined, because each Kryo instance across the distributed application must use the same IDs. Kryo recommends small positive integers and reserves a few ids (value < 10). Spring Integration currently defaults to using 40, 41, and 42 (for the file and message header serializers mentioned earlier). We recommend you start at 60, to allow for expansion in the framework. You can override these framework defaults by configuring the registrars mentioned earlier. |
Using a Custom Kryo Serializer
If you need custom serialization, see the Kryo documentation, because you need to use the native API to do the customization.
For an example, see the MessageCodec
implementation.
Implementing KryoSerializable
If you have write access to the domain object source code, you can implement KryoSerializable
as described here.
In this case, the class provides the serialization methods itself and no further configuration is required.
However benchmarks have shown this is not quite as efficient as registering a custom serializer explicitly.
The following example shows a custom Kryo serializer:
public class Address implements KryoSerializable {
...
@Override
public void write(Kryo kryo, Output output) {
output.writeString(this.street);
output.writeString(this.city);
output.writeString(this.country);
}
@Override
public void read(Kryo kryo, Input input) {
this.street = input.readString();
this.city = input.readString();
this.country = input.readString();
}
}
You can also use this technique to wrap a serialization library other than Kryo.
Using the @DefaultSerializer
Annotation
Kryo also provides a @DefaultSerializer
annotation, as described here.
@DefaultSerializer(SomeClassSerializer.class)
public class SomeClass {
// ...
}
If you have write access to the domain object, this may be a simpler way to specify a custom serializer. Note that this does not register the class with an ID, which may make the technique unhelpful for certain situations.
10. Messaging Endpoints
10.1. Message Endpoints
The first part of this chapter covers some background theory and reveals quite a bit about the underlying API that drives Spring Integration’s various messaging components. This information can be helpful if you want to really understand what goes on behind the scenes. However, if you want to get up and running with the simplified namespace-based configuration of the various elements, feel free to skip ahead to Endpoint Namespace Support for now.
As mentioned in the overview, message endpoints are responsible for connecting the various messaging components to channels. Over the next several chapters, we cover a number of different components that consume messages. Some of these are also capable of sending reply messages. Sending messages is quite straightforward. As shown earlier in Message Channels, you can send a message to a message channel. However, receiving is a bit more complicated. The main reason is that there are two types of consumers: polling consumers and event-driven consumers.
Of the two, event-driven consumers are much simpler.
Without any need to manage and schedule a separate poller thread, they are essentially listeners with a callback method.
When connecting to one of Spring Integration’s subscribable message channels, this simple option works great.
However, when connecting to a buffering, pollable message channel, some component has to schedule and manage the polling threads.
Spring Integration provides two different endpoint implementations to accommodate these two types of consumers.
Therefore, the consumers themselves need only implement the callback interface.
When polling is required, the endpoint acts as a container for the consumer instance.
The benefit is similar to that of using a container for hosting message-driven beans, but, since these consumers are Spring-managed objects running within an ApplicationContext
, it more closely resembles Spring’s own MessageListener
containers.
10.1.1. Message Handler
Spring Integration’s MessageHandler
interface is implemented by many of the components within the framework.
In other words, this is not part of the public API, and you would not typically implement MessageHandler
directly.
Nevertheless, it is used by a message consumer for actually handling the consumed messages, so being aware of this strategy interface does help in terms of understanding the overall role of a consumer.
The interface is defined as follows:
public interface MessageHandler {
void handleMessage(Message<?> message);
}
Despite its simplicity, this interface provides the foundation for most of the components (routers, transformers, splitters, aggregators, service activators, and others) covered in the following chapters. Those components each perform very different functionality with the messages they handle, but the requirements for actually receiving a message are the same, and the choice between polling and event-driven behavior is also the same. Spring Integration provides two endpoint implementations that host these callback-based handlers and let them be connected to message channels.
10.1.2. Event-driven Consumer
Because it is the simpler of the two, we cover the event-driven consumer endpoint first.
You may recall that the SubscribableChannel
interface provides a subscribe()
method and that the method accepts a MessageHandler
parameter (as shown in SubscribableChannel
).
The following listing shows the definition of the subscribe
method:
subscribableChannel.subscribe(messageHandler);
Since a handler that is subscribed to a channel does not have to actively poll that channel, this is an event-driven consumer, and the implementation provided by Spring Integration accepts a SubscribableChannel
and a MessageHandler
, as the following example shows:
SubscribableChannel channel = context.getBean("subscribableChannel", SubscribableChannel.class);
EventDrivenConsumer consumer = new EventDrivenConsumer(channel, exampleHandler);
10.1.3. Polling Consumer
Spring Integration also provides a PollingConsumer
, and it can be instantiated in the same way except that the channel must implement PollableChannel
, as the following example shows:
PollableChannel channel = context.getBean("pollableChannel", PollableChannel.class);
PollingConsumer consumer = new PollingConsumer(channel, exampleHandler);
| For more information regarding polling consumers, see Poller and Channel Adapter. |
There are many other configuration options for the polling consumer. For example, the trigger is a required property. The following example shows how to set the trigger:
PollingConsumer consumer = new PollingConsumer(channel, handler);
consumer.setTrigger(new IntervalTrigger(30, TimeUnit.SECONDS));
Spring Integration currently provides two implementations of the Trigger
interface: IntervalTrigger
and CronTrigger
.
The IntervalTrigger
is typically defined with a simple interval (in milliseconds) but also supports an initialDelay
property and a boolean fixedRate
property (the default is false
— that is, no fixed delay).
The following example sets both properties:
IntervalTrigger trigger = new IntervalTrigger(1000);
trigger.setInitialDelay(5000);
trigger.setFixedRate(true);
The result of the three settings in the preceding example is a trigger that waits five seconds and then triggers every second.
The CronTrigger
requires a valid cron expression.
See the Javadoc for details.
The following example sets a new CronTrigger
:
CronTrigger trigger = new CronTrigger("*/10 * * * * MON-FRI");
The result of the trigger defined in the previous example is a trigger that triggers every ten seconds, Monday through Friday.
In addition to the trigger, you can specify two other polling-related configuration properties: maxMessagesPerPoll
and receiveTimeout
.
The following example shows how to set these two properties:
PollingConsumer consumer = new PollingConsumer(channel, handler);
consumer.setMaxMessagesPerPoll(10);
consumer.setReceiveTimeout(5000);
The maxMessagesPerPoll
property specifies the maximum number of messages to receive within a given poll operation.
This means that the poller continues calling receive() without waiting, until either null
is returned or the maximum value is reached.
For example, if a poller has a ten-second interval trigger and a maxMessagesPerPoll
setting of 25
, and it is polling a channel that has 100 messages in its queue, all 100 messages can be retrieved within 40 seconds.
It grabs 25, waits ten seconds, grabs the next 25, and so on.
The receiveTimeout
property specifies the amount of time the poller should wait if no messages are available when it invokes the receive operation.
For example, consider two options that seem similar on the surface but are actually quite different: The first has an interval trigger of 5 seconds and a receive timeout of 50 milliseconds, while the second has an interval trigger of 50 milliseconds and a receive timeout of 5 seconds.
The first one may receive a message up to 4950 milliseconds later than it arrived on the channel (if that message arrived immediately after one of its poll calls returned).
On the other hand, the second configuration never misses a message by more than 50 milliseconds.
The difference is that the second option requires a thread to wait.
However, as a result, it can respond much more quickly to arriving messages.
This technique, known as “long polling”, can be used to emulate event-driven behavior on a polled source.
A polling consumer can also delegate to a Spring TaskExecutor
, as the following example shows:
PollingConsumer consumer = new PollingConsumer(channel, handler);
TaskExecutor taskExecutor = context.getBean("exampleExecutor", TaskExecutor.class);
consumer.setTaskExecutor(taskExecutor);
Furthermore, a PollingConsumer
has a property called adviceChain
.
This property lets you to specify a List
of AOP advices for handling additional cross cutting concerns including transactions.
These advices are applied around the doPoll()
method.
For more in-depth information, see the sections on AOP advice chains and transaction support under Endpoint Namespace Support.
The earlier examples show dependency lookups.
However, keep in mind that these consumers are most often configured as Spring bean definitions.
In fact, Spring Integration also provides a FactoryBean
called ConsumerEndpointFactoryBean
that creates the appropriate consumer type based on the type of channel.
Also, Spring Integration has full XML namespace support to even further hide those details.
The namespace-based configuration is in this guide featured as each component type is introduced.
Many of the MessageHandler implementations can generate reply messages.
As mentioned earlier, sending messages is trivial when compared to receiving messages.
Nevertheless, when and how many reply messages are sent depends on the handler type.
For example, an aggregator waits for a number of messages to arrive and is often configured as a downstream consumer for a splitter, which can generate multiple replies for each message it handles.
When using the namespace configuration, you do not strictly need to know all of the details.
However, it still might be worth knowing that several of these components share a common base class, the AbstractReplyProducingMessageHandler , and that it provides a setOutputChannel(..) method.
|
10.1.4. Endpoint Namespace Support
Throughout this reference manual, you can find specific configuration examples for endpoint elements, such as router, transformer, service-activator, and so on.
Most of these support an input-channel
attribute and many support an output-channel
attribute.
After being parsed, these endpoint elements produce an instance of either the PollingConsumer
or the EventDrivenConsumer
, depending on the type of the input-channel
that is referenced: PollableChannel
or SubscribableChannel
, respectively.
When the channel is pollable, the polling behavior is based on the endpoint element’s poller
sub-element and its attributes.
The following listing lists all of the available configuration options for a poller
:
<int:poller cron="" (1)
default="false" (2)
error-channel="" (3)
fixed-delay="" (4)
fixed-rate="" (5)
id="" (6)
max-messages-per-poll="" (7)
receive-timeout="" (8)
ref="" (9)
task-executor="" (10)
time-unit="MILLISECONDS" (11)
trigger=""> (12)
<int:advice-chain /> (13)
<int:transactional /> (14)
</int:poller>
| 1 | Provides the ability to configure pollers by using Cron expressions.
The underlying implementation uses an org.springframework.scheduling.support.CronTrigger .
If this attribute is set, none of the following attributes must be specified: fixed-delay , trigger , fixed-rate , and ref . |
| 2 | By setting this attribute to true , you can define exactly one global default poller.
An exception is raised if more than one default poller is defined in the application context.
Any endpoints connected to a PollableChannel (PollingConsumer ) or any SourcePollingChannelAdapter that does not have an explicitly configured poller then uses the global default poller.
It defaults to false .
Optional. |
| 3 | Identifies the channel to which error messages are sent if a failure occurs in this poller’s invocation.
To completely suppress exceptions, you can provide a reference to the nullChannel .
Optional. |
| 4 | The fixed delay trigger uses a PeriodicTrigger under the covers.
If you do not use the time-unit attribute, the specified value is represented in milliseconds.
If this attribute is set, none of the following attributes must be specified: fixed-rate , trigger , cron , and ref . |
| 5 | The fixed rate trigger uses a PeriodicTrigger under the covers.
If you do not use the time-unit attribute, the specified value is represented in milliseconds.
If this attribute is set, none of the following attributes must be specified: fixed-delay , trigger , cron , and ref . |
| 6 | The ID referring to the poller’s underlying bean-definition, which is of type org.springframework.integration.scheduling.PollerMetadata .
The id attribute is required for a top-level poller element, unless it is the default poller (default="true" ). |
| 7 | See Configuring An Inbound Channel Adapter for more information.
If not specified, the default value depends on the context.
If you use a PollingConsumer , this attribute defaults to -1 .
However, if you use a SourcePollingChannelAdapter , the max-messages-per-poll attribute defaults to 1 .
Optional. |
| 8 | Value is set on the underlying class PollerMetadata .
If not specified, it defaults to 1000 (milliseconds).
Optional. |
| 9 | Bean reference to another top-level poller.
The ref attribute must not be present on the top-level poller element.
However, if this attribute is set, none of the following attributes must be specified: fixed-rate , trigger , cron , and fixed-delay . |
| 10 | Provides the ability to reference a custom task executor. See TaskExecutor Support for further information. Optional. |
| 11 | This attribute specifies the java.util.concurrent.TimeUnit enum value on the underlying org.springframework.scheduling.support.PeriodicTrigger .
Therefore, this attribute can be used only in combination with the fixed-delay or fixed-rate attributes.
If combined with either cron or a trigger reference attribute, it causes a failure.
The minimal supported granularity for a PeriodicTrigger is milliseconds.
Therefore, the only available options are milliseconds and seconds.
If this value is not provided, any fixed-delay or fixed-rate value is interpreted as milliseconds.
Basically, this enum provides a convenience for seconds-based interval trigger values.
For hourly, daily, and monthly settings, we recommend using a cron trigger instead. |
| 12 | Reference to any Spring-configured bean that implements the org.springframework.scheduling.Trigger interface.
However, if this attribute is set, none of the following attributes must be specified: fixed-delay , fixed-rate , cron , and ref .
Optional. |
| 13 | Allows specifying extra AOP advices to handle additional cross-cutting concerns. See Transaction Support for further information. Optional. |
| 14 | Pollers can be made transactional. See AOP Advice chains for further information. Optional. |
Examples
A simple interval-based poller with a 1-second interval can be configured as follows:
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output">
<int:poller fixed-rate="1000"/>
</int:transformer>
As an alternative to using the fixed-rate
attribute, you can also use the fixed-delay
attribute.
For a poller based on a Cron expression, use the cron
attribute instead, as the following example shows:
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output">
<int:poller cron="*/10 * * * * MON-FRI"/>
</int:transformer>
If the input channel is a PollableChannel
, the poller configuration is required.
Specifically, as mentioned earlier, the trigger
is a required property of the PollingConsumer
class.
Therefore, if you omit the poller
sub-element for a polling consumer endpoint’s configuration, an exception may be thrown.
The exception may also be thrown if you attempt to configure a poller on the element that is connected to a non-pollable channel.
It is also possible to create top-level pollers, in which case only a ref
attribute is required, as the following example shows:
<int:poller id="weekdayPoller" cron="*/10 * * * * MON-FRI"/>
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output">
<int:poller ref="weekdayPoller"/>
</int:transformer>
The ref attribute is allowed only on the inner poller definitions.
Defining this attribute on a top-level poller results in a configuration exception being thrown during initialization of the application context.
|
Global Default Pollers
To simplify the configuration even further, you can define a global default poller.
A single top-level poller within an ApplicationContext
may have the default
attribute set to true
.
In that case, any endpoint with a PollableChannel
for its input channel, that is defined within the same ApplicationContext
, and has no explicitly configured poller
sub-element uses that default.
The following example shows such a poller and a transformer that uses it:
<int:poller id="defaultPoller" default="true" max-messages-per-poll="5" fixed-rate="3000"/>
<!-- No <poller/> sub-element is necessary, because there is a default -->
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output"/>
Transaction Support
Spring Integration also provides transaction support for the pollers so that each receive-and-forward operation can be performed as an atomic unit of work.
To configure transactions for a poller, add the <transactional/>
sub-element.
The following example shows the available attributes:
<int:poller fixed-delay="1000">
<int:transactional transaction-manager="txManager"
propagation="REQUIRED"
isolation="REPEATABLE_READ"
timeout="10000"
read-only="false"/>
</int:poller>
For more information, see Poller Transaction Support.
AOP Advice chains
Since Spring transaction support depends on the proxy mechanism with TransactionInterceptor
(AOP Advice) handling transactional behavior of the message flow initiated by the poller, you must sometimes provide extra advices to handle other cross cutting behavior associated with the poller.
For that, the poller
defines an advice-chain
element that lets you add more advices in a class that implements the MethodInterceptor
interface.
The following example shows how to define an advice-chain
for a poller
:
<int:service-activator id="advicedSa" input-channel="goodInputWithAdvice" ref="testBean"
method="good" output-channel="output">
<int:poller max-messages-per-poll="1" fixed-rate="10000">
<int:advice-chain>
<ref bean="adviceA" />
<beans:bean class="org.something.SampleAdvice" />
<ref bean="txAdvice" />
</int:advice-chain>
</int:poller>
</int:service-activator>
For more information on how to implement the MethodInterceptor
interface, see the AOP sections of the Spring Framework Reference Guide.
An advice chain can also be applied on a poller that does not have any transaction configuration, letting you enhance the behavior of the message flow initiated by the poller.
When using an advice chain, the <transactional/> child element cannot be specified.
Instead, declare a <tx:advice/> bean and add it to the <advice-chain/> .
See Poller Transaction Support for complete configuration details.
|
TaskExecutor Support
The polling threads may be executed by any instance of Spring’s TaskExecutor
abstraction.
This enables concurrency for an endpoint or group of endpoints.
As of Spring 3.0, the core Spring Framework has a task
namespace, and its <executor/>
element supports the creation of a simple thread pool executor.
That element accepts attributes for common concurrency settings, such as pool-size and queue-capacity.
Configuring a thread-pooling executor can make a substantial difference in how the endpoint performs under load.
These settings are available for each endpoint, since the performance of an endpoint is one of the major factors to consider (the other major factor being the expected volume on the channel to which the endpoint subscribes).
To enable concurrency for a polling endpoint that is configured with the XML namespace support, provide the task-executor
reference on its <poller/>
element and then provide one or more of the properties shown in the following example:
<int:poller task-executor="pool" fixed-rate="1000"/>
<task:executor id="pool"
pool-size="5-25"
queue-capacity="20"
keep-alive="120"/>
If you do not provide a task-executor, the consumer’s handler is invoked in the caller’s thread.
Note that the caller is usually the default TaskScheduler
(see Configuring the Task Scheduler).
You should also keep in mind that the task-executor
attribute can provide a reference to any implementation of Spring’s TaskExecutor
interface by specifying the bean name.
The executor
element shown earlier is provided for convenience.
As mentioned earlier in the background section for polling consumers, you can also configure a polling consumer in such a way as to emulate event-driven behavior.
With a long receive-timeout
and a short interval-trigger
, you can ensure a very timely reaction to arriving messages even on a polled message source.
Note that this applies only to sources that have a blocking wait call with a timeout.
For example, the file poller does not block.
Each receive() call returns immediately and either contains new files or not.
Therefore, even if a poller contains a long receive-timeout
, that value would never be used in such a scenario.
On the other hand, when using Spring Integration’s own queue-based channels, the timeout value does have a chance to participate.
The following example shows how a polling consumer can receive messages nearly instantaneously:
<int:service-activator input-channel="someQueueChannel"
output-channel="output">
<int:poller receive-timeout="30000" fixed-rate="10"/>
</int:service-activator>
Using this approach does not carry much overhead, since, internally, it is nothing more then a timed-wait thread, which does not require nearly as much CPU resource usage as (for example) a thrashing, infinite while loop.
10.1.5. Changing Polling Rate at Runtime
When configuring a poller with a fixed-delay
or a fixed-rate
attribute, the default implementation uses a PeriodicTrigger
instance.
The PeriodicTrigger
is part of the core Spring Framework.
It accepts the interval only as a constructor argument.
Therefore, it cannot be changed at runtime.
However, you can define your own implementation of the org.springframework.scheduling.Trigger
interface.
You could even use the PeriodicTrigger
as a starting point.
Then you can add a setter for the interval (period), or you can even embed your own throttling logic within the trigger itself.
The period
property is used with each call to nextExecutionTime
to schedule the next poll.
To use this custom trigger within pollers, declare the bean definition of the custom trigger in your application context and inject the dependency into your poller configuration by using the trigger
attribute, which references the custom trigger bean instance.
You can now obtain a reference to the trigger bean and change the polling interval between polls.
For an example, see the Spring Integration Samples project.
It contains a sample called dynamic-poller
, which uses a custom trigger and demonstrates the ability to change the polling interval at runtime.
The sample provides a custom trigger that implements the org.springframework.scheduling.Trigger
interface.
The sample’s trigger is based on Spring’s PeriodicTrigger
implementation.
However, the fields of the custom trigger are not final, and the properties have explicit getters and setters, letting you dynamically change the polling period at runtime.
It is important to note, though, that because the Trigger method is nextExecutionTime() , any changes to a dynamic trigger do not take effect until the next poll, based on the existing configuration.
It is not possible to force a trigger to fire before its currently configured next execution time.
|
10.1.6. Payload Type Conversion
Throughout this reference manual, you can also see specific configuration and implementation examples of various endpoints that accept a message or any arbitrary Object
as an input parameter.
In the case of an Object
, such a parameter is mapped to a message payload or part of the payload or header (when using the Spring Expression Language).
However, the type of input parameter of the endpoint method sometimes does not match the type of the payload or its part.
In this scenario, we need to perform type conversion.
Spring Integration provides a convenient way for registering type converters (by using the Spring ConversionService
) within its own instance of a conversion service bean named integrationConversionService
.
That bean is automatically created as soon as the first converter is defined by using the Spring Integration infrastructure.
To register a converter, you can implement org.springframework.core.convert.converter.Converter
, org.springframework.core.convert.converter.GenericConverter
, or org.springframework.core.convert.converter.ConverterFactory
.
The Converter
implementation is the simplest and converts from a single type to another.
For more sophistication, such as converting to a class hierarchy, you can implement a GenericConverter
and possibly a ConditionalConverter
.
These give you complete access to the from
and to
type descriptors, enabling complex conversions.
For example, if you have an abstract class called Something
that is the target of your conversion (parameter type, channel data type, and so on), you have two concrete implementations called Thing1
and Thing
, and you wish to convert to one or the other based on the input type, the GenericConverter
would be a good fit.
For more information, see the Javadoc for these interfaces:
When you have implemented your converter, you can register it with convenient namespace support, as the following example shows:
<int:converter ref="sampleConverter"/>
<bean id="sampleConverter" class="foo.bar.TestConverter"/>
Alternately, you can use an inner bean, as the following example shows:
<int:converter>
<bean class="o.s.i.config.xml.ConverterParserTests$TestConverter3"/>
</int:converter>
Starting with Spring Integration 4.0, you can use annotations to create the preceding configuration, as the following example shows:
@Component
@IntegrationConverter
public class TestConverter implements Converter<Boolean, Number> {
public Number convert(Boolean source) {
return source ? 1 : 0;
}
}
Alternately, you can use the @Configuration
annotation, as the following example shows:
@Configuration
@EnableIntegration
public class ContextConfiguration {
@Bean
@IntegrationConverter
public SerializingConverter serializingConverter() {
return new SerializingConverter();
}
}
|
When configuring an application context, the Spring Framework lets you add a In contrast, the However, if you do want to use the Spring
In this case, the converters provided by the |
10.1.7. Content Type Conversion
Starting with version 5.0, by default, the method invocation mechanism is based on the org.springframework.messaging.handler.invocation.InvocableHandlerMethod
infrastructure.
Its HandlerMethodArgumentResolver
implementations (such as PayloadArgumentResolver
and MessageMethodArgumentResolver
) can use the MessageConverter
abstraction to convert an incoming payload
to the target method argument type.
The conversion can be based on the contentType
message header.
For this purpose, Spring Integration provides the ConfigurableCompositeMessageConverter
, which delegates to a list of registered converters to be invoked until one of them returns a non-null result.
By default, this converter provides (in strict order):
-
MappingJackson2MessageConverter
if the Jackson processor is present on the classpath
See the Javadoc (linked in the preceding list) for more information about their purpose and appropriate contentType
values for conversion.
The ConfigurableCompositeMessageConverter
is used because it can be be supplied with any other MessageConverter
implementations, including or excluding the previously mentioned default converters.
It can also be registered as an appropriate bean in the application context, overriding the default converter, as the following example shows:
@Bean(name = IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)
public ConfigurableCompositeMessageConverter compositeMessageConverter() {
List<MessageConverter> converters =
Arrays.asList(new MarshallingMessageConverter(jaxb2Marshaller()),
new JavaSerializationMessageConverter());
return new ConfigurableCompositeMessageConverter(converters);
}
Those two new converters are registered in the composite before the defaults.
You can also not use a ConfigurableCompositeMessageConverter
but provide your own MessageConverter
by registering a bean with the name, integrationArgumentResolverMessageConverter
(by setting the IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME
property).
The MessageConverter -based (including contentType header) conversion is not available when using SpEL method invocation.
In this case, only the regular class-to-class conversion mentioned above in the Payload Type Conversion is available.
|
10.1.8. Asynchronous Polling
If you want the polling to be asynchronous, a poller can optionally specify a task-executor
attribute that points to an existing instance of any TaskExecutor
bean (Spring 3.0 provides a convenient namespace configuration through the task
namespace).
However, there are certain things you must understand when configuring a poller with a TaskExecutor
.
The problem is that there are two configurations in place, the poller and the TaskExecutor
.
They must be in tune with each other.
Otherwise, you might end up creating an artificial memory leak.
Consider the following configuration:
<int:channel id="publishChannel">
<int:queue />
</int:channel>
<int:service-activator input-channel="publishChannel" ref="myService">
<int:poller receive-timeout="5000" task-executor="taskExecutor" fixed-rate="50" />
</int:service-activator>
<task:executor id="taskExecutor" pool-size="20" />
The preceding configuration demonstrates an out-of-tune configuration.
By default, the task executor has an unbounded task queue. The poller keeps scheduling new tasks even though all the threads are blocked, waiting for either a new message to arrive or the timeout to expire. Given that there are 20 threads executing tasks with a five-second timeout, they aree executed at a rate of 4 per second. However, new tasks are being scheduled at a rate of 20 per second, so the internal queue in the task executor grows at a rate of 16 per second (while the process is idle), so we have a memory leak.
One of the ways to handle this is to set the queue-capacity
attribute of the task executor.
Even 0 is a reasonable value.
You can also manage it by specifying what to do with messages that can not be queued by setting the rejection-policy
attribute of the Task Executor (for example, to DISCARD
).
In other words, there are certain details you must understand when configuring TaskExecutor
.
See “Task Execution and Scheduling” in the Spring reference manual for more detail on the subject.
10.1.9. Endpoint Inner Beans
Many endpoints are composite beans.
This includes all consumers and all polled inbound channel adapters.
Consumers (polled or event-driven) delegate to a MessageHandler
.
Polled adapters obtain messages by delegating to a MessageSource
.
Often, it is useful to obtain a reference to the delegate bean, perhaps to change configuration at runtime or for testing.
These beans can be obtained from the ApplicationContext
with well known names.
MessageHandler
instances are registered with the application context with bean IDs similar to someConsumer.handler
(where 'consumer' is the value of the endpoint’s id
attribute).
MessageSource
instances are registered with bean IDs similar to somePolledAdapter.source
, where 'somePolledAdapter' is the ID of the adapter.
The preceding only applies to the framework component itself. You can instead use an inner bean definition, as the following example shows:
<int:service-activator id="exampleServiceActivator" input-channel="inChannel"
output-channel = "outChannel" method="foo">
<beans:bean class="org.foo.ExampleServiceActivator"/>
</int:service-activator>
The bean is treated like any inner bean declared and is not registered with the application context.
If you wish to access this bean in some other manner, declare it at the top level with an id
and use the ref
attribute instead.
See the Spring Documentation for more information.
10.2. Endpoint Roles
Starting with version 4.2, endpoints can be assigned to roles.
Roles let endpoints be started and stopped as a group.
This is particularly useful when using leadership election, where a set of endpoints can be started or stopped when leadership is granted or revoked, respectively.
For this purpose the framework registers a SmartLifecycleRoleController
bean in the application context with the name IntegrationContextUtils.INTEGRATION_LIFECYCLE_ROLE_CONTROLLER
.
Whenever it is necessary to control lifecycles, this bean can be injected or @Autowired
:
<bean class="com.some.project.SomeLifecycleControl">
<property name="roleController" ref="integrationLifecycleRoleController"/>
</bean>
You can assign endpoints to roles using XML, Java configuration, or programmatically. The following example shows how to configure endpoint roles with XML:
<int:inbound-channel-adapter id="ica" channel="someChannel" expression="'foo'" role="cluster"
auto-startup="false">
<int:poller fixed-rate="60000" />
</int:inbound-channel-adapter>
The following example shows how to configure endpoint roles for a bean created in Java:
@Bean
@ServiceActivator(inputChannel = "sendAsyncChannel", autoStartup="false")
@Role("cluster")
public MessageHandler sendAsyncHandler() {
return // some MessageHandler
}
The following example shows how to configure endpoint roles on a method in Java:
@Payload("#args[0].toLowerCase()")
@Role("cluster")
public String handle(String payload) {
return payload.toUpperCase();
}
The following example shows how to configure endpoint roles by using the SmartLifecycleRoleController
in Java:
@Autowired
private SmartLifecycleRoleController roleController;
...
this.roleController.addSmartLifeCycleToRole("cluster", someEndpoint);
...
The following example shows how to configure endpoint roles by using an IntegrationFlow
in Java:
IntegrationFlow flow -> flow
.handle(..., e -> e.role("cluster"));
Each of these adds the endpoint to the cluster
role.
Invoking roleController.startLifecyclesInRole("cluster")
and the corresponding stop…
method starts and stops the endpoints.
Any object that implements SmartLifecycle can be programmatically added — not just endpoints.
|
The SmartLifecycleRoleController
implements ApplicationListener<AbstractLeaderEvent>
and it automatically starts and stops its configured SmartLifecycle
objects when leadership is granted or revoked (when some bean publishes OnGrantedEvent
or OnRevokedEvent
, respectively).
When using leadership election to start and stop components, it is important to set the auto-startup XML attribute (autoStartup bean property) to false so that the application context does not start the components during context initialization.
|
Starting with version 4.3.8, the SmartLifecycleRoleController
provides several status methods:
public Collection<String> getRoles() (1)
public boolean allEndpointsRunning(String role) (2)
public boolean noEndpointsRunning(String role) (3)
public Map<String, Boolean> getEndpointsRunningStatus(String role) (4)
| 1 | Returns a list of the roles being managed. |
| 2 | Returns true if all endpoints in the role are running. |
| 3 | Returns true if none of the endpoints in the role are running. |
| 4 | Returns a map of component name : running status .
The component name is usually the bean name. |
10.3. Leadership Event Handling
Groups of endpoints can be started and stopped based on leadership being granted or revoked, respectively. This is useful in clustered scenarios where shared resources must be consumed by only a single instance. An example of this is a file inbound channel adapter that is polling a shared directory. (See Reading Files).
To participate in a leader election and be notified when elected leader, when leadership is revoked, or on failure to acquire the resources to become leader, an application creates a component in the application context called a “leader initiator”.
Normally, a leader initiator is a SmartLifecycle
, so it starts (optionally) when the context starts and then publishes notifications when leadership changes.
You can also receive failure notifications by setting the publishFailedEvents
to true
(starting with version 5.0), for cases when you want take a specific action if a failure occurs.
By convention, you should provide a Candidate
that receives the callbacks.
You can also revoke the leadership through a Context
object provided by the framework.
Your code can also listen for o.s.i.leader.event.AbstractLeaderEvent
instances (the super class of OnGrantedEvent
and OnRevokedEvent
) and respond accordingly (for instance, by using a SmartLifecycleRoleController
).
The events contain a reference to the Context
object.
The following listing shows the definition of the Context
interface:
public interface Context {
boolean isLeader();
void yield();
String getRole();
}
Starting with version 5.0.6, the context provides a reference to the candidate’s role.
Spring Integration provides a basic implementation of a leader initiator that is based on the LockRegistry
abstraction.
To use it, you need to create an instance as a bean, as the following example shows:
@Bean
public LockRegistryLeaderInitiator leaderInitiator(LockRegistry locks) {
return new LockRegistryLeaderInitiator(locks);
}
If the lock registry is implemented correctly, there is only ever at most one leader.
If the lock registry also provides locks that throw exceptions (ideally, InterruptedException
) when they expire or are broken, the duration of the leaderless periods can be as short as is allowed by the inherent latency in the lock implementation.
By default, the busyWaitMillis
property adds some additional latency to prevent CPU starvation in the (more usual) case that the locks are imperfect and you only know they expired when you try to obtain one again.
See Zookeeper Leadership Event Handling for more information about leadership election and events that use Zookeeper.
10.4. Messaging Gateways
A gateway hides the messaging API provided by Spring Integration. It lets your application’s business logic be unaware of the Spring Integration API. By using a generic Gateway, your code interacts with only a simple interface.
10.4.1. Enter the GatewayProxyFactoryBean
As mentioned earlier, it would be great to have no dependency on the Spring Integration API — including the gateway class.
For that reason, Spring Integration provides the GatewayProxyFactoryBean
, which generates a proxy for any interface and internally invokes the gateway methods shown below.
By using dependency injection, you can then expose the interface to your business methods.
The following example shows an interface that can be used to interact with Spring Integration:
package org.cafeteria;
public interface Cafe {
void placeOrder(Order order);
}
10.4.2. Gateway XML Namespace Support
Namespace support is also provided. It lets you configure an interface as a service, as the following example shows:
<int:gateway id="cafeService"
service-interface="org.cafeteria.Cafe"
default-request-channel="requestChannel"
default-reply-timeout="10000"
default-reply-channel="replyChannel"/>
With this configuration defined, the cafeService
can now be injected into other beans, and the code that invokes the methods on that proxied instance of the Cafe
interface has no awareness of the Spring Integration API.
The general approach is similar to that of Spring Remoting (RMI, HttpInvoker, and so on).
See the “Samples” Appendix for an example that uses the gateway
element (in the Cafe demo).
The defaults in the preceding configuration are applied to all methods on the gateway interface. If a reply timeout is not specified, the calling thread waits indefinitely for a reply. See Gateway Behavior When No response Arrives.
The defaults can be overridden for individual methods. See Gateway Configuration with Annotations and XML.
10.4.3. Setting the Default Reply Channel
Typically, you need not specify the default-reply-channel
, since a Gateway auto-creates a temporary, anonymous reply channel, where it listens for the reply.
However, some cases may prompt you to define a default-reply-channel
(or reply-channel
with adapter gateways, such as HTTP, JMS, and others).
For some background, we briefly discuss some of the inner workings of the gateway.
A gateway creates a temporary point-to-point reply channel.
It is anonymous and is added to the message headers with the name, replyChannel
.
When providing an explicit default-reply-channel
(reply-channel
with remote adapter gateways), you can point to a publish-subscribe channel, which is so named because you can add more than one subscriber to it.
Internally, Spring Integration creates a bridge between the temporary replyChannel
and the explicitly defined default-reply-channel
.
Suppose you want your reply to go not only to the gateway but also to some other consumer. In this case, you want two things:
-
A named channel to which you can subscribe
-
That channel to be a publish-subscribe-channel
The default strategy used by the gateway does not satisfy those needs, because the reply channel added to the header is anonymous and point-to-point.
This means that no other subscriber can get a handle to it and, even if it could, the channel has point-to-point behavior such that only one subscriber would get the message.
By defining a default-reply-channel
you can point to a channel of your choosing.
In this case, that is a publish-subscribe-channel
.
The gateway creates a bridge from it to the temporary, anonymous reply channel that is stored in the header.
You might also want to explicitly provide a reply channel for monitoring or auditing through an interceptor (for example, wiretap). To configure a channel interceptor, you need a named channel.
10.4.4. Gateway Configuration with Annotations and XML
Consider the following example, which expands on the previous Cafe
interface example by adding a @Gateway
annotation:
public interface Cafe {
@Gateway(requestChannel="orders")
void placeOrder(Order order);
}
The @Header
annotation lets you add values that are interpreted as message headers, as the following example shows:
public interface FileWriter {
@Gateway(requestChannel="filesOut")
void write(byte[] content, @Header(FileHeaders.FILENAME) String filename);
}
If you prefer the XML approach to configuring gateway methods, you can add method
elements to the gateway configuration, as the following example shows:
<int:gateway id="myGateway" service-interface="org.foo.bar.TestGateway"
default-request-channel="inputC">
<int:default-header name="calledMethod" expression="#gatewayMethod.name"/>
<int:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/>
<int:method name="echoUpperCase" request-channel="inputB"/>
<int:method name="echoViaDefault"/>
</int:gateway>
You can also use XML to provide individual headers for each method invocation.
This could be useful if the headers you want to set are static in nature and you do not want to embed them in the gateway’s method signature by using @Header
annotations.
For example, in the loan broker example, we want to influence how aggregation of the loan quotes is done, based on what type of request was initiated (single quote or all quotes).
Determining the type of the request by evaluating which gateway method was invoked, although possible, would violate the separation of concerns paradigm (the method is a Java artifact).
However, expressing your intention (meta information) in message headers is natural in a messaging architecture.
The following example shows how to add a different message header for each of two methods:
<int:gateway id="loanBrokerGateway"
service-interface="org.springframework.integration.loanbroker.LoanBrokerGateway">
<int:method name="getLoanQuote" request-channel="loanBrokerPreProcessingChannel">
<int:header name="RESPONSE_TYPE" value="BEST"/>
</int:method>
<int:method name="getAllLoanQuotes" request-channel="loanBrokerPreProcessingChannel">
<int:header name="RESPONSE_TYPE" value="ALL"/>
</int:method>
</int:gateway>
In the preceding example a different value is set for the 'RESPONSE_TYPE' header, based on the gateway’s method.
Expressions and “Global” Headers
The <header/>
element supports expression
as an alternative to value
.
The SpEL expression is evaluated to determine the value of the header.
There is no #root
object, but the following variables are available:
-
#args: An
Object[]
containing the method arguments -
#gatewayMethod: The object (derived from
java.reflect.Method
) that represents the method in theservice-interface
that was invoked. A header containing this variable can be used later in the flow (for example, for routing). For example, if you wish to route on the simple method name, you might add a header with the following expression:#gatewayMethod.name
.
The java.reflect.Method is not serializable.
A header with an expression of #gatewayMethod is lost if you later serialize the message.
Consequently, you may wish to use #gatewayMethod.name or #gatewayMethod.toString() in those cases.
The toString() method provides a String representation of the method, including parameter and return types.
|
Since version 3.0, <default-header/>
elements can be defined to add headers to all the messages produced by the gateway, regardless of the method invoked.
Specific headers defined for a method take precedence over default headers.
Specific headers defined for a method here override any @Header
annotations in the service interface.
However, default headers do NOT override any @Header
annotations in the service interface.
The gateway now also supports a default-payload-expression
, which is applied for all methods (unless overridden).
10.4.5. Mapping Method Arguments to a Message
Using the configuration techniques in the previous section allows control of how method arguments are mapped to message elements (payload and headers). When no explicit configuration is used, certain conventions are used to perform the mapping. In some cases, these conventions cannot determine which argument is the payload and which should be mapped to headers. Consider the following example:
public String send1(Object thing1, Map thing2);
public String send2(Map thing1, Map thing2);
In the first case, the convention is to map the first argument to the payload (as long as it is not a Map
) and the contents of the second argument become headers.
In the second case (or the first when the argument for parameter thing1
is a Map
), the framework cannot determine which argument should be the payload.
Consequently, mapping fails.
This can generally be resolved using a payload-expression
, a @Payload
annotation, or a @Headers
annotation.
Alternatively (and whenever the conventions break down), you can take the entire responsibility for mapping the method calls to messages.
To do so, implement an MethodArgsMessageMapper
and provide it to the <gateway/>
by using the mapper
attribute.
The mapper maps a MethodArgsHolder
, which is a simple class that wraps the java.reflect.Method
instance and an Object[]
containing the arguments.
When providing a custom mapper, the default-payload-expression
attribute and <default-header/>
elements are not allowed on the gateway.
Similarly, the payload-expression
attribute and <header/>
elements are not allowed on any <method/>
elements.
Mapping Method Arguments
The following examples show how method arguments can be mapped to the message and shows some examples of invalid configuration:
public interface MyGateway {
void payloadAndHeaderMapWithoutAnnotations(String s, Map<String, Object> map);
void payloadAndHeaderMapWithAnnotations(@Payload String s, @Headers Map<String, Object> map);
void headerValuesAndPayloadWithAnnotations(@Header("k1") String x, @Payload String s, @Header("k2") String y);
void mapOnly(Map<String, Object> map); // the payload is the map and no custom headers are added
void twoMapsAndOneAnnotatedWithPayload(@Payload Map<String, Object> payload, Map<String, Object> headers);
@Payload("#args[0] + #args[1] + '!'")
void payloadAnnotationAtMethodLevel(String a, String b);
@Payload("@someBean.exclaim(#args[0])")
void payloadAnnotationAtMethodLevelUsingBeanResolver(String s);
void payloadAnnotationWithExpression(@Payload("toUpperCase()") String s);
void payloadAnnotationWithExpressionUsingBeanResolver(@Payload("@someBean.sum(#this)") String s); // (1)
// invalid
void twoMapsWithoutAnnotations(Map<String, Object> m1, Map<String, Object> m2);
// invalid
void twoPayloads(@Payload String s1, @Payload String s2);
// invalid
void payloadAndHeaderAnnotationsOnSameParameter(@Payload @Header("x") String s);
// invalid
void payloadAndHeadersAnnotationsOnSameParameter(@Payload @Headers Map<String, Object> map);
}
| 1 | Note that, in this example, the SpEL variable, #this , refers to the argument — in this case, the value of s . |
The XML equivalent looks a little different, since there is no #this
context for the method argument.
However, expressions can refer to method arguments by using the #args
variable, as the following example shows:
<int:gateway id="myGateway" service-interface="org.something.MyGateway">
<int:method name="send1" payload-expression="#args[0] + 'thing2'"/>
<int:method name="send2" payload-expression="@someBean.sum(#args[0])"/>
<int:method name="send3" payload-expression="#method"/>
<int:method name="send4">
<int:header name="thing1" expression="#args[2].toUpperCase()"/>
</int:method>
</int:gateway>
10.4.6. @MessagingGateway
Annotation
Starting with version 4.0, gateway service interfaces can be marked with a @MessagingGateway
annotation instead of requiring the definition of a <gateway />
xml element for configuration.
The following pair of examples compares the two approaches for configuring the same gateway:
<int:gateway id="myGateway" service-interface="org.something.TestGateway"
default-request-channel="inputC">
<int:default-header name="calledMethod" expression="#gatewayMethod.name"/>
<int:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/>
<int:method name="echoUpperCase" request-channel="inputB">
<int:header name="thing1" value="thing2"/>
</int:method>
<int:method name="echoViaDefault"/>
</int:gateway>
@MessagingGateway(name = "myGateway", defaultRequestChannel = "inputC",
defaultHeaders = @GatewayHeader(name = "calledMethod",
expression="#gatewayMethod.name"))
public interface TestGateway {
@Gateway(requestChannel = "inputA", replyTimeout = 2, requestTimeout = 200)
String echo(String payload);
@Gateway(requestChannel = "inputB", headers = @GatewayHeader(name = "thing1", value="thing2"))
String echoUpperCase(String payload);
String echoViaDefault(String payload);
}
Similarly to the XML version, when Spring Integration discovers these annotations during a component scan, it creates the proxy implementation with its messaging infrastructure.
To perform this scan and register the BeanDefinition in the application context, add the @IntegrationComponentScan annotation to a @Configuration class.
The standard @ComponentScan infrastructure does not deal with interfaces.
Consequently, we introduced the custom @IntegrationComponentScan logic to fine the @MessagingGateway annotation on the interfaces and register GatewayProxyFactoryBean instances for them.
See also Annotation Support.
|
Along with the @MessagingGateway
annotation you can mark a service interface with the @Profile
annotation to avoid the bean creation, if such a profile is not active.
If you have no XML configuration, the @EnableIntegration annotation is required on at least one @Configuration class.
See Configuration and @EnableIntegration for more information.
|
10.4.7. Invoking No-Argument Methods
When invoking methods on a Gateway interface that do not have any arguments, the default behavior is to receive a Message
from a PollableChannel
.
Sometimes, however, you may want to trigger no-argument methods so that you can interact with other components downstream that do not require user-provided parameters, such as triggering no-argument SQL calls or stored procedures.
To achieve send-and-receive semantics, you must provide a payload.
To generate a payload, method parameters on the interface are not necessary.
You can either use the @Payload
annotation or the payload-expression
attribute in XML on the method
element.
The following list includes a few examples of what the payloads could be:
-
a literal string
-
#gatewayMethod.name
-
new java.util.Date()
-
@someBean.someMethod()'s return value
The following example shows how to use the @Payload
annotation:
public interface Cafe {
@Payload("new java.util.Date()")
List<Order> retrieveOpenOrders();
}
If a method has no argument and no return value but does contain a payload expression, it is treated as a send-only operation.
10.4.8. Error Handling
The gateway invocation can result in errors. By default, any error that occurs downstream is re-thrown “as is” upon the gateway’s method invocation. For example, consider the following simple flow:
gateway -> service-activator
If the service invoked by the service activator throws a MyException
(for example), the framework wraps it in a MessagingException
and attaches the message passed to the service activator in the failedMessage
property.
Consequently, any logging performed by the framework has full the context of the failure.
By default, when the exception is caught by the gateway, the MyException
is unwrapped and thrown to the caller.
You can configure a throws
clause on the gateway method declaration to match the particular exception type in the cause chain.
For example, if you want to catch a whole MessagingException
with all the messaging information of the reason of downstream error, you should have a gateway method similar to the following:
public interface MyGateway {
void performProcess() throws MessagingException;
}
Since we encourage POJO programming, you may not want to expose the caller to messaging infrastructure.
If your gateway method does not have a throws
clause, the gateway traverses the cause tree, looking for a RuntimeException
that is not a MessagingException
.
If none is found, the framework throws the MessagingException
.
If the MyException
in the preceding discussion has a cause of SomeOtherException
and your method throws SomeOtherException
, the gateway further unwraps that and throws it to the caller.
When a gateway is declared with no service-interface
, an internal framework interface RequestReplyExchanger
is used.
Consider the following example:
public interface RequestReplyExchanger {
Message<?> exchange(Message<?> request) throws MessagingException;
}
Before version 5.0, this exchange
method did not have a throws
clause and, as a result, the exception was unwrapped.
If you use this interface and want to restore the previous unwrap behavior, use a custom service-interface
instead or access the cause
of the MessagingException
yourself.
However, you may want to log the error rather than propagating it or you may want to treat an exception as a valid reply (by mapping it to a message that conforms to some "error message" contract that the caller understands).
To accomplish this, the gateway provides support for a message channel dedicated to the errors by including support for the error-channel
attribute.
In the following example, a 'transformer' creates a reply Message
from the Exception
:
<int:gateway id="sampleGateway"
default-request-channel="gatewayChannel"
service-interface="foo.bar.SimpleGateway"
error-channel="exceptionTransformationChannel"/>
<int:transformer input-channel="exceptionTransformationChannel"
ref="exceptionTransformer" method="createErrorResponse"/>
The exceptionTransformer
could be a simple POJO that knows how to create the expected error response objects.
That becomes the payload that is sent back to the caller.
You could do many more elaborate things in such an “error flow”, if necessary.
It might involve routers (including Spring Integration’s ErrorMessageExceptionTypeRouter
), filters, and so on.
Most of the time, a simple 'transformer' should be sufficient, however.
Alternatively, you might want to only log the exception (or send it somewhere asynchronously).
If you provide a one-way flow, nothing would be sent back to the caller.
If you want to completely suppress exceptions, you can provide a reference to the global nullChannel
(essentially a /dev/null
approach).
Finally, as mentioned above, if no error-channel
is defined, then the exceptions propagate as usual.
When you use the @MessagingGateway
annotation (see
), you can use use the @MessagingGateway
AnnotationerrorChannel
attribute.
Starting with version 5.0, when you use a gateway method with a void
return type (one-way flow), the error-channel
reference (if provided) is populated in the standard errorChannel
header of each sent message.
This feature allows a downstream asynchronous flow, based on the standard ExecutorChannel
configuration (or a QueueChannel
), to override a default global errorChannel
exceptions sending behavior.
Previously you had to manually specify an errorChannel
header with the @GatewayHeader
annotation or the <header>
element.
The error-channel
property was ignored for void
methods with an asynchronous flow.
Instead, error messages were sent to the default errorChannel
.
Exposing the messaging system through simple POJI Gateways provides benefits, but “hiding” the reality of the underlying messaging system does come at a price, so there are certain things you should consider.
We want our Java method to return as quickly as possible and not hang for an indefinite amount of time while the caller is waiting on it to return (whether void, a return value, or a thrown Exception).
When regular methods are used as a proxies in front of the messaging system, we have to take into account the potentially asynchronous nature of the underlying messaging.
This means that there might be a chance that a message that was initiated by a gateway could be dropped by a filter and never reach a component that is responsible for producing a reply.
Some service activator method might result in an exception, thus providing no reply (as we do not generate null messages).
In other words, multiple scenarios can cause a reply message to never come.
That is perfectly natural in messaging systems.
However, think about the implication on the gateway method. The gateway’s method input arguments were incorporated into a message and sent downstream.
The reply message would be converted to a return value of the gateway’s method.
So you might want to ensure that, for each gateway call, there is always a reply message.
Otherwise, your gateway method might never return and hang indefinitely.
One way to handle this situation is by using an asynchronous gateway (explained later in this section).
Another way of handling it is to explicitly set the reply-timeout attribute.
That way, the gateway does not hang any longer than the time specified by the reply-timeout and returns 'null' if that timeout does elapse.
Finally, you might want to consider setting downstream flags, such as 'requires-reply', on a service-activator or 'throw-exceptions-on-rejection' on a filter. These options are discussed in more detail in the final section of this chapter.
|
If the downstream flow returns an ErrorMessage , its payload (a Throwable ) is treated as a regular downstream error.
If there is an error-channel configured, it is sent to the error flow.
Otherwise the payload is thrown to the caller of the gateway.
Similarly, if the error flow on the error-channel returns an ErrorMessage , its payload is thrown to the caller.
The same applies to any message with a Throwable payload.
This can be useful in asynchronous situations when when you need to propagate an Exception directly to the caller.
To do so, you can either return an Exception (as the reply from some service) or throw it.
Generally, even with an asynchronous flow, the framework takes care of propagating an exception thrown by the downstream flow back to the gateway.
The TCP Client-Server Multiplex sample demonstrates both techniques to return the exception to the caller.
It emulates a socket IO error to the waiting thread by using an aggregator with group-timeout (see Aggregator and Group Timeout) and a MessagingTimeoutException reply on the discard flow.
|
10.4.9. Gateway Timeouts
Gateways have two timeout properties: requestTimeout
and replyTimeout
.
The request timeout applies only if the channel can block (for example, a bounded QueueChannel
that is full).
The replyTimeout
value is how long the gateway waits for a reply or returns null
.
It defaults to infinity.
The timeouts can be set as defaults for all methods on the gateway (defaultRequestTimeout
and defaultReplyTimeout
) or on the MessagingGateway
interface annotation.
Individual methods can override these defaults (in <method/>
child elements) or on the @Gateway
annotation.
Starting with version 5.0, the timeouts can be defined as expressions, as the following example shows:
@Gateway(payloadExpression = "#args[0]", requestChannel = "someChannel",
requestTimeoutExpression = "#args[1]", replyTimeoutExpression = "#args[2]")
String lateReply(String payload, long requestTimeout, long replyTimeout);
The evaluation context has a BeanResolver
(use @someBean
to reference other beans), and the #args
array variable is available.
When configuring with XML, the timeout attributes can be a long value or a SpEL expression, as the following example shows:
<method name="someMethod" request-channel="someRequestChannel"
payload-expression="#args[0]"
request-timeout="1000"
reply-timeout="#args[1]">
</method>