10.4.10. Asynchronous Gateway
As a pattern, the messaging gateway offers a nice way to hide messaging-specific code while still exposing the full capabilities of the messaging system.
As described earlier, the GatewayProxyFactoryBean
provides a convenient way to expose a proxy over a service-interface giving you POJO-based access to a messaging system (based on objects in your own domain, primitives/Strings, or other objects).
However, when a gateway is exposed through simple POJO methods that return values, it implies that, for each request message (generated when the method is invoked), there must be a reply message (generated when the method has returned).
Since messaging systems are naturally asynchronous, you may not always be able to guarantee the contract where “for each request, there will always be be a reply”. Spring Integration 2.0 introduced support for an asynchronous gateway, which offers a convenient way to initiate flows when you may not know if a reply is expected or how long it takes for replies to arrive.
To handle these types of scenarios, Spring Integration uses java.util.concurrent.Future
instances to support an asynchronous gateway.
From the XML configuration, nothing changes, and you still define asynchronous gateway the same way as you define a regular gateway, as the following example shows:
<int:gateway id="mathService"
service-interface="org.springframework.integration.sample.gateway.futures.MathServiceGateway"
default-request-channel="requestChannel"/>
However, the gateway interface (a service interface) is a little different, as follows:
public interface MathServiceGateway {
Future<Integer> multiplyByTwo(int i);
}
As the preceding example shows, the return type for the gateway method is a Future
.
When GatewayProxyFactoryBean
sees that the return type of the gateway method is a Future
, it immediately switches to the asynchronous mode by using an AsyncTaskExecutor
.
That is the extent of the differences.
The call to such a method always returns immediately with a Future
instance.
Then you can interact with the Future
at your own pace to get the result, cancel, and so on.
Also, as with any other use of Future
instances, calling get()
may reveal a timeout, an execution exception, and so on.
The following example shows how to use a Future
that returns from an asynchronous gateway:
MathServiceGateway mathService = ac.getBean("mathService", MathServiceGateway.class);
Future<Integer> result = mathService.multiplyByTwo(number);
// do something else here since the reply might take a moment
int finalResult = result.get(1000, TimeUnit.SECONDS);
For a more detailed example, see the async-gateway sample in the Spring Integration samples.
ListenableFuture
Starting with version 4.1, asynchronous gateway methods can also return ListenableFuture
(introduced in Spring Framework 4.0).
These return types let you provide a callback, which is invoked when the result is available (or an exception occurs).
When the gateway detects this return type and the task executor is an AsyncListenableTaskExecutor
, the executor’s submitListenable()
method is invoked.
The following example shows how to use a ListenableFuture
:
ListenableFuture<String> result = this.asyncGateway.async("something");
result.addCallback(new ListenableFutureCallback<String>() {
@Override
public void onSuccess(String result) {
...
}
@Override
public void onFailure(Throwable t) {
...
}
});
AsyncTaskExecutor
By default, the GatewayProxyFactoryBean
uses org.springframework.core.task.SimpleAsyncTaskExecutor
when submitting internal AsyncInvocationTask
instances for any gateway method whose return type is a Future
.
However, the async-executor
attribute in the <gateway/>
element’s configuration lets you provide a reference to any implementation of java.util.concurrent.Executor
available within the Spring application context.
The (default) SimpleAsyncTaskExecutor
supports both Future
and ListenableFuture
return types, returning FutureTask
or ListenableFutureTask
respectively.
See CompletableFuture
.
Even though there is a default executor, it is often useful to provide an external one so that you can identify its threads in logs (when using XML, the thread name is based on the executor’s bean name), as the following example shows:
@Bean
public AsyncTaskExecutor exec() {
SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
simpleAsyncTaskExecutor.setThreadNamePrefix("exec-");
return simpleAsyncTaskExecutor;
}
@MessagingGateway(asyncExecutor = "exec")
public interface ExecGateway {
@Gateway(requestChannel = "gatewayChannel")
Future<?> doAsync(String foo);
}
If you wish to return a different Future
implementation, you can provide a custom executor or disable the executor altogether and return the Future
in the reply message payload from the downstream flow.
To disable the executor, set it to null
in the GatewayProxyFactoryBean
(by using setAsyncTaskExecutor(null)
).
When configuring the gateway with XML, use async-executor=""
.
When configuring by using the @MessagingGateway
annotation, use code similar to the following:
@MessagingGateway(asyncExecutor = AnnotationConstants.NULL)
public interface NoExecGateway {
@Gateway(requestChannel = "gatewayChannel")
Future<?> doAsync(String foo);
}
If the return type is a specific concrete Future implementation or some other sub-interface that is not supported by the configured executor, the flow runs on the caller’s thread and the flow must return the required type in the reply message payload.
|
CompletableFuture
Starting with version 4.2, gateway methods can now return CompletableFuture<?>
.
There are two modes of operation when returning this type:
-
When an async executor is provided and the return type is exactly
CompletableFuture
(not a subclass), the framework runs the task on the executor and immediately returns aCompletableFuture
to the caller.CompletableFuture.supplyAsync(Supplier<U> supplier, Executor executor)
is used to create the future. -
When the async executor is explicitly set to
null
and the return type isCompletableFuture
or the return type is a subclass ofCompletableFuture
, the flow is invoked on the caller’s thread. In this scenario, the downstream flow is expected to return aCompletableFuture
of the appropriate type.
Usage Scenarios
In the following scenario, the caller thread returns immediately with a CompletableFuture<Invoice>
, which is completed when the downstream flow replies to the gateway (with an Invoice
object).
CompletableFuture<Invoice> order(Order order);
<int:gateway service-interface="something.Service" default-request-channel="orders" />
In the following scenario, the caller thread returns with a CompletableFuture<Invoice>
when the downstream flow provides it as the payload of the reply to the gateway.
Some other process must complete the future when the invoice is ready.
CompletableFuture<Invoice> order(Order order);
<int:gateway service-interface="foo.Service" default-request-channel="orders"
async-executor="" />
In the following scenario, the caller thread returns with a CompletableFuture<Invoice>
when the downstream flow provides it as the payload of the reply to the gateway.
Some other process must complete the future when the invoice is ready.
If DEBUG
logging is enabled, a log entry is emitted, indicating that the async executor cannot be used for this scenario.
MyCompletableFuture<Invoice> order(Order order);
<int:gateway service-interface="foo.Service" default-request-channel="orders" />
CompletableFuture
instances can be used to perform additional manipulation on the reply, as the following example shows:
CompletableFuture<String> process(String data);
...
CompletableFuture result = process("foo")
.thenApply(t -> t.toUpperCase());
...
String out = result.get(10, TimeUnit.SECONDS);
Reactor Mono
Starting with version 5.0, the GatewayProxyFactoryBean
allows the use of Project Reactor with gateway interface methods, using a Mono<T>
return type.
The internal AsyncInvocationTask
is wrapped in a Mono.fromCallable()
.
A Mono
can be used to retrieve the result later (similar to a Future<?>
), or you can consume from it with the dispatcher by invoking your Consumer
when the result is returned to the gateway.
The Mono is not immediately flushed by the framework.
Consequently, the underlying message flow is not started before the gateway method returns (as it is with a Future<?> Executor task).
The flow starts when the Mono is subscribed to.
Alternatively, the Mono (being a Composable ) might be a part of Reactor stream, when the subscribe() is related to the entire Flux .
The following example shows how to create a gateway with Project Reactor:
|
@MessagingGateway
public static interface TestGateway {
@Gateway(requestChannel = "promiseChannel")
Mono<Integer> multiply(Integer value);
}
...
@ServiceActivator(inputChannel = "promiseChannel")
public Integer multiply(Integer value) {
return value * 2;
}
...
Flux.just("1", "2", "3", "4", "5")
.map(Integer::parseInt)
.flatMap(this.testGateway::multiply)
.collectList()
.subscribe(integers -> ...);
Another example that uses Project Reactor is a simple callback scenario, as the following example shows:
Mono<Invoice> mono = service.process(myOrder);
mono.subscribe(invoice -> handleInvoice(invoice));
The calling thread continues, with handleInvoice()
being called when the flow completes.
Downstream Flows Returning an Asynchronous Type
As mentioned in the ListenableFuture
section above, if you wish some downstream component to return a message with an async payload (Future
, Mono
, and others), you must explicitly set the async executor to null
(or ""
when using XML configuration).
The flow is then invoked on the caller thread and the result can be retrieved later.
void
Return Type
Unlike the return types mentioned earlier, when the method return type is void
, the framework cannot implicitly determine that you wish the downstream flow to run asynchronously, with the caller thread returning immediately.
In this case, you must annotate the interface method with @Async
, as the following example shows:
@MessagingGateway
public interface MyGateway {
@Gateway(requestChannel = "sendAsyncChannel")
@Async
void sendAsync(String payload);
}
Unlike the Future<?>
return types, there is no way to inform the caller if some exception is thrown by the flow, unless some custom TaskExecutor
(such as an ErrorHandlingTaskExecutor
) is associated with the @Async
annotation.
10.4.11. Gateway Behavior When No response Arrives
As explained earlier, the gateway provides a convenient way of interacting with a messaging system through POJO method invocations. However, a typical method invocation, which is generally expected to always return (even with an Exception), might not always map one-to-one to message exchanges (for example, a reply message might not arrive — the equivalent to a method not returning).
The rest of this section covers various scenarios and how to make the gateway behave more predictably.
Certain attributes can be configured to make synchronous gateway behavior more predictable, but some of them might not always work as you might expect.
One of them is reply-timeout
(at the method level or default-reply-timeout
at the gateway level).
We examine the reply-timeout
attribute to see how it can and cannot influence the behavior of the synchronous gateway in various scenarios.
We examine a single-threaded scenario (all components downstream are connected through a direct channel) and multi-threaded scenarios (for example, somewhere downstream you may have a pollable or executor channel that breaks the single-thread boundary).
Long-running Process Downstream
- Sync Gateway, single-threaded
-
If a component downstream is still running (perhaps because of an infinite loop or a slow service), setting a
reply-timeout
has no effect, and the gateway method call does not return until the downstream service exits (by returning or throwing an exception). - Sync Gateway, multi-threaded
-
If a component downstream is still running (perhaps because of an infinite loop or a slow service) in a multi-threaded message flow, setting the
reply-timeout
has an effect by allowing gateway method invocation to return once the timeout has been reached, because theGatewayProxyFactoryBean
polls on the reply channel, waiting for a message until the timeout expires. However, if the timeout has been reached before the actual reply was produced, it could result in a 'null' return from the gateway method. You should understand that the reply message (if produced) is sent to a reply channel after the gateway method invocation might have returned, so you must be aware of that and design your flow with it in mind.
Downstream Component Returns 'null'
- Sync Gateway — single-threaded
-
If a component downstream returns 'null' and no
reply-timeout
has been configured, the gateway method call hangs indefinitely, unless areply-timeout
has been configured or therequires-reply
attribute has been set on the downstream component (for example, a service activator) that might return 'null'. In this case, an exception would be thrown and propagated to the gateway. - Sync Gateway — multi-threaded
-
The behavior is the same as the previous case.
Downstream Component Return Signature is 'void' While Gateway Method Signature Is Non-void
- Sync Gateway — single-threaded
-
If a component downstream returns 'void' and no
reply-timeout
has been configured, the gateway method call hangs indefinitely unless areply-timeout
has been configured. - Sync Gateway — multi-threaded
-
The behavior is the same as the previous case.
Downstream Component Results in Runtime Exception
- Sync Gateway — single-threaded
-
If a component downstream throws a runtime exception, the exception is propagated through an error message back to the gateway and re-thrown.
- Sync Gateway — multi-threaded
-
The behavior is the same as the previous case.
You should understand that, by default, reply-timeout is unbounded.
Consequently, if you do not explicitly set the reply-timeout , your gateway method invocation might hang indefinitely.
So, to make sure you analyze your flow and if there is even a remote possibility of one of these scenarios to occur, you should set the reply-timeout attribute to a "'safe'" value.
Even better, you can set the requires-reply attribute of the downstream component to 'true' to ensure a timely response, as produced by the throwing of an exception as soon as that downstream component returns null internally.
However you should also realize that there are some scenarios (see the first one) where reply-timeout does not help.
That means it is also important to analyze your message flow and decide when to use a synchronous gateway rather than an asynchrnous gateway.
As described earlier, the latter case is a matter of defining gateway methods that return Future instances.
Then you are guaranteed to receive that return value, and you have more granular control over the results of the invocation.
Also, when dealing with a router, you should remember that setting the resolution-required attribute to 'true' results in an exception thrown by the router if it can not resolve a particular channel.
Likewise, when dealing with a Filter, you can set the throw-exception-on-rejection attribute.
In both of these cases, the resulting flow behaves like it contain a service activator with the 'requires-reply' attribute.
In other words, it helps to ensure a timely response from the gateway method invocation.
|
reply-timeout is unbounded for <gateway/> elements (created by the GatewayProxyFactoryBean ).
Inbound gateways for external integration (WS, HTTP, and so on) share many characteristics and attributes with these gateways.
However, for those inbound gateways, the default reply-timeout is 1000 milliseconds (one second).
If a downstream asynchronous hand-off is made to another thread, you may need to increase this attribute to allow enough time for the flow to complete before the gateway times out.
|
| You should understand that the timer starts when the thread returns to the gateway — that is, when the flow completes or a message is handed off to another thread. At that time, the calling thread starts waiting for the reply. If the flow was completely synchronous, the reply is immediately available. For asynchronous flows, the thread waits for up to this time. |
See IntegrationFlow
as Gateway in the Java DSL chapter for options to define gateways through IntegrationFlows
.
10.5. Service Activator
The service activator is the endpoint type for connecting any Spring-managed object to an input channel so that it may play the role of a service.
If the service produces output, it may also be connected to an output channel.
Alternatively, an output-producing service may be located at the end of a processing pipeline or message flow, in which case the inbound message’s replyChannel
header can be used.
This is the default behavior if no output channel is defined.
As with most of the configuration options described here, the same behavior actually applies for most of the other components.
10.5.1. Configuring Service Activator
To create a service activator, use the 'service-activator' element with the 'input-channel' and 'ref' attributes, as the following example shows:
<int:service-activator input-channel="exampleChannel" ref="exampleHandler"/>
The preceding configuration selects all the methods from the exampleHandler
that meet one of the messaging requirements, which are as follows:
-
annotated with
@ServiceActivator
-
is
public
-
not return
void
ifrequiresReply == true
The target method for invocation at runtime is selected for each request message by their payload
type or as a fallback to the Message<?>
type if such a method is present on target class.
Starting with version 5.0, one service method can be marked with the @org.springframework.integration.annotation.Default
as a fallback for all non-matching cases.
This can be useful when using content-type conversion with the target method being invoked after conversion.
To delegate to an explicitly defined method of any object, you can add the method
attribute, as the following example shows:
<int:service-activator input-channel="exampleChannel" ref="somePojo" method="someMethod"/>
In either case, when the service method returns a non-null value, the endpoint tries to send the reply message to an appropriate reply channel.
To determine the reply channel, it first checks whether an output-channel
was provided in the endpoint configuration, as the following example shows:
<int:service-activator input-channel="exampleChannel" output-channel="replyChannel"
ref="somePojo" method="someMethod"/>
If the method returns a result and no output-channel
is defined, the framework then checks the request message’s replyChannel
header value.
If that value is available, it then checks its type.
If it is a MessageChannel
, the reply message is sent to that channel.
If it is a String
, the endpoint tries to resolve the channel name to a channel instance.
If the channel cannot be resolved, a DestinationResolutionException
is thrown.
It it can be resolved, the message is sent there.
If the request message does not have a replyChannel
header and the reply
object is a Message
, its replyChannel
header is consulted for a target destination.
This is the technique used for request-reply messaging in Spring Integration, and it is also an example of the return address pattern.
If your method returns a result and you want to discard it and end the flow, you should configure the output-channel
to send to a NullChannel
.
For convenience, the framework registers one with the name, nullChannel
.
See Special Channels for more information.
The service activator is one of those components that is not required to produce a reply message.
If your method returns null
or has a void
return type, the service activator exits after the method invocation, without any signals.
This behavior can be controlled by the AbstractReplyProducingMessageHandler.requiresReply
option, which is also exposed as requires-reply
when configuring with the XML namespace.
If the flag is set to true
and the method returns null, a ReplyRequiredException
is thrown.
The argument in the service method could be either a message or an arbitrary type.
If the latter, then it is assumed to be a message payload, which is extracted from the message and injected into the service method.
We generally recommend this approach, as it follows and promotes a POJO model when working with Spring Integration.
Arguments may also have @Header
or @Headers
annotations, as described in Annotation Support.
| The service method is not required to have any arguments, which means you can implement event-style service activators (where all you care about is an invocation of the service method) and not worry about the contents of the message. Think of it as a null JMS message. An example use case for such an implementation is a simple counter or monitor of messages deposited on the input channel. |
Starting with version 4.1, the framework correctly converts message properties (payload
and headers
) to the Java 8 Optional
POJO method parameters, as the following example shows:
public class MyBean {
public String computeValue(Optional<String> payload,
@Header(value="foo", required=false) String foo1,
@Header(value="foo") Optional<String> foo2) {
if (payload.isPresent()) {
String value = payload.get();
...
}
else {
...
}
}
}
We generally recommend using a ref
attribute if the custom service activator handler implementation can be reused in other <service-activator>
definitions.
However, if the custom service activator handler implementation is only used within a single definition of the <service-activator>
, you can provide an inner bean definition, as the following example shows:
<int:service-activator id="exampleServiceActivator" input-channel="inChannel"
output-channel = "outChannel" method="someMethod">
<beans:bean class="org.something.ExampleServiceActivator"/>
</int:service-activator>
Using both the ref attribute and an inner handler definition in the same <service-activator> 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 handlers 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.
|
Service Activators and the Spring Expression Language (SpEL)
Since Spring Integration 2.0, service activators can also benefit from SpEL.
For example, you can invoke any bean method without pointing to the bean in a ref
attribute or including it as an inner bean definition, as follows:
<int:service-activator input-channel="in" output-channel="out"
expression="@accountService.processAccount(payload, headers.accountId)"/>
<bean id="accountService" class="thing1.thing2.Account"/>
In the preceding configuration, instead of injecting 'accountService' by using a ref
or as an inner bean, we use SpEL’s @beanId
notation and invoke a method that takes a type compatible with the message payload.
We also pass a header value.
Any valid SpEL expression can be evaluated against any content in the message.
For simple scenarios, your service activators need not reference a bean if all logic can be encapsulated in such an expression, as the following example shows:
<int:service-activator input-channel="in" output-channel="out" expression="payload * 2"/>
In the preceding configuration, our service logic is to multiply the payload value by two. SpEL lets us handle it relatively easily.
See Service Activators and the .handle()
method in the Java DSL chapter for more information about configuring service activator.
10.5.2. Asynchronous Service Activator
The service activator is invoked by the calling thread.
This is an upstream thread if the input channel is a SubscribableChannel
or a poller thread for a PollableChannel
.
If the service returns a ListenableFuture<?>
, the default action is to send that as the payload of the message sent to the output (or reply) channel.
Starting with version 4.3, you can now set the async
attribute to true
(by using setAsync(true)
when using Java configuration).
If the service returns a ListenableFuture<?>
when this the async
attribute is set to true
, the calling thread is released immediately and the reply message is sent on the thread (from within your service) that completes the future.
This is particularly advantageous for long-running services that use a PollableChannel
, because the poller thread is released to perform other services within the framework.
If the service completes the future with an Exception
, normal error processing occurs.
An ErrorMessage
is sent to the errorChannel
message header, if present.
Otherwise, an ErrorMessage
is sent to the default errorChannel
(if available).
10.5.3. Service Activator and Method Return Type
The service method can return any type which becomes reply message payload.
In this case a new Message<?>
object is created and all the headers from a request message are copied.
This works the same way for most Spring Integration MessageHandler
implementations, when interaction is based on a POJO method invocation.
A complete Message<?>
object can also be returned from the method.
However keep in mind that, unlike transformers, for a Service Activator this message will be modified by copying the headers from the request message if they are not already present in the returned message.
So, if your method parameter is a Message<?>
and you copy some, but not all, existing headers in your service method, they will reappear in the reply message.
It is not a Service Activator responsibility to remove headers from a reply message and, pursuing the loosely-coupled principle, it is better to add a HeaderFilter
in the integration flow.
Alternatively, a Transformer can be used instead of a Service Activator but, in that case, when returning a full Message<?>
the method is completely responsible for the message, including copying request message headers (if needed).
You must ensure that important framework headers (e.g. replyChannel
, errorChannel
), if present, have to be preserved.
10.6. Delayer
A delayer is a simple endpoint that lets a message flow be delayed by a certain interval.
When a message is delayed, the original sender does not block.
Instead, the delayed messages are scheduled with an instance of org.springframework.scheduling.TaskScheduler
to be sent to the output channel after the delay has passed.
This approach is scalable even for rather long delays, since it does not result in a large number of blocked sender threads.
On the contrary, in the typical case, a thread pool is used for the actual execution of releasing the messages.
This section contains several examples of configuring a delayer.
10.6.1. Configuring a Delayer
The <delayer>
element is used to delay the message flow between two message channels.
As with the other endpoints, you can provide the 'input-channel' and 'output-channel' attributes, but the delayer also has 'default-delay' and 'expression' attributes (and the 'expression' element) that determine the number of milliseconds by which each message should be delayed.
The following example delays all messages by three seconds:
<int:delayer id="delayer" input-channel="input"
default-delay="3000" output-channel="output"/>
If you need to determine the delay for each message, you can also provide the SpEL expression by using the 'expression' attribute, as the following expression shows:
<int:delayer id="delayer" input-channel="input" output-channel="output"
default-delay="3000" expression="headers['delay']"/>
In the preceding example, the three-second delay applies only when the expression evaluates to null for a given inbound message.
If you want to apply a delay only to messages that have a valid result of the expression evaluation, you can use a 'default-delay' of 0
(the default).
For any message that has a delay of 0
(or less), the message is sent immediately, on the calling thread.
The following example shows the Java configuration equivalent of the preceding example:
@ServiceActivator(inputChannel = "input")
@Bean
public DelayHandler delayer() {
DelayHandler handler = new DelayHandler("delayer.messageGroupId");
handler.setDefaultDelay(3_000L);
handler.setDelayExpressionString("headers['delay']");
handler.setOutputChannelName("output");
return handler;
}
The following example shows the Java DSL equivalent of the preceding example:
@Bean
public IntegrationFlow flow() {
return IntegrationFlows.from("input")
.delay("delayer.messageGroupId", d -> d
.defaultDelay(3_000L)
.delayExpression("headers['delay']"))
.channel("output")
.get();
}
The XML parser uses a message group ID of <beanName>.messageGroupId .
|
The delay handler supports expression evaluation results that represent an interval in milliseconds (any Object whose toString() method produces a value that can be parsed into a Long ) as well as java.util.Date instances representing an absolute time.
In the first case, the milliseconds are counted from the current time (for example
a value of 5000 would delay the message for at least five seconds from the time it is received by the delayer).
With a Date instance, the message is not released until the time represented by that Date object.
A value that equates to a non-positive delay or a Date in the past results in no delay.
Instead, it is sent directly to the output channel on the original sender’s thread.
If the expression evaluation result is not a Date and can not be parsed as a Long , the default delay (if any — the default is 0 ) is applied.
|
The expression evaluation may throw an evaluation exception for various reasons, including an invalid expression or other conditions.
By default, such exceptions are ignored (though logged at the DEBUG level) and the delayer falls back to the default delay (if any).
You can modify this behavior by setting the ignore-expression-failures attribute.
By default, this attribute is set to true and the delayer behavior is as described earlier.
However, if you wish to not ignore expression evaluation exceptions and throw them to the delayer’s caller, set the ignore-expression-failures attribute to false .
|
|
In the preceding example, the delay expression is specified as
Consequently, if there is a possibility of the header being omitted and you want to fall back to the default delay, it is generally more efficient (and recommended) to use the indexer syntax instead of dot property accessor syntax, because detecting the null is faster than catching an exception. |
The delayer delegates to an instance of Spring’s TaskScheduler
abstraction.
The default scheduler used by the delayer is the ThreadPoolTaskScheduler
instance provided by Spring Integration on startup.
See Configuring the Task Scheduler.
If you want to delegate to a different scheduler, you can provide a reference through the delayer element’s 'scheduler' attribute, as the following example shows:
<int:delayer id="delayer" input-channel="input" output-channel="output"
expression="headers.delay"
scheduler="exampleTaskScheduler"/>
<task:scheduler id="exampleTaskScheduler" pool-size="3"/>
If you configure an external ThreadPoolTaskScheduler , you can set waitForTasksToCompleteOnShutdown = true on this property.
It allows successful completion of 'delay' tasks that are already in the execution state (releasing the message) when the application is shutdown.
Before Spring Integration 2.2, this property was available on the <delayer> element, because DelayHandler could create its own scheduler on the background.
Since 2.2, the delayer requires an external scheduler instance and waitForTasksToCompleteOnShutdown was deleted.
You should use the scheduler’s own configuration.
|
ThreadPoolTaskScheduler has a property errorHandler , which can be injected with some implementation of org.springframework.util.ErrorHandler .
This handler allows processing an Exception from the thread of the scheduled task sending the delayed message.
By default, it uses an org.springframework.scheduling.support.TaskUtils$LoggingErrorHandler , and you can see a stack trace in the logs.
You might want to consider using an org.springframework.integration.channel.MessagePublishingErrorHandler , which sends an ErrorMessage into an error-channel , either from the failed message’s header or into the default error-channel .
This error handling is performed after a transaction rolls back (if present).
See Release Failures.
|
10.6.2. Delayer and a Message Store
The DelayHandler
persists delayed messages into the message group in the provided MessageStore
.
(The 'groupId' is based on the required 'id' attribute of the <delayer>
element.)
A delayed message is removed from the MessageStore
by the scheduled task immediately before the DelayHandler
sends the message to the output-channel
.
If the provided MessageStore
is persistent (such as JdbcMessageStore
), it provides the ability to not lose messages on the application shutdown.
After application startup, the DelayHandler
reads messages from its message group in the MessageStore
and reschedules them with a delay based on the original arrival time of the message (if the delay is numeric).
For messages where the delay header was a Date
, that Date
is used when rescheduling.
If a delayed message remains in the MessageStore
more than its 'delay', it is sent immediately after startup.
The <delayer>
can be enriched with either of two mutually exclusive elements: <transactional>
and <advice-chain>
.
The List
of these AOP advices is applied to the proxied internal DelayHandler.ReleaseMessageHandler
, which has the responsibility to release the message, after the delay, on a Thread
of the scheduled task.
It might be used, for example, when the downstream message flow throws an exception and the transaction of the ReleaseMessageHandler
is rolled back.
In this case, the delayed message remains in the persistent MessageStore
.
You can use any custom org.aopalliance.aop.Advice
implementation within the <advice-chain>
.
The <transactional>
element defines a simple advice chain that has only the transactional advice.
The following example shows an advice-chain
within a <delayer>
:
<int:delayer id="delayer" input-channel="input" output-channel="output"
expression="headers.delay"
message-store="jdbcMessageStore">
<int:advice-chain>
<beans:ref bean="customAdviceBean"/>
<tx:advice>
<tx:attributes>
<tx:method name="*" read-only="true"/>
</tx:attributes>
</tx:advice>
</int:advice-chain>
</int:delayer>
The DelayHandler
can be exported as a JMX MBean
with managed operations (getDelayedMessageCount
and reschedulePersistedMessages
), which allows the rescheduling of delayed persisted messages at runtime — for example, if the TaskScheduler
has previously been stopped.
These operations can be invoked through a Control Bus
command, as the following example shows:
Message<String> delayerReschedulingMessage =
MessageBuilder.withPayload("@'delayer.handler'.reschedulePersistedMessages()").build();
controlBusChannel.send(delayerReschedulingMessage);
| For more information regarding the message store, JMX, and the control bus, see System Management. |
10.6.3. Release Failures
Starting with version 5.0.8, there are two new properties on the delayer:
-
maxAttempts
(default 5) -
retryDelay
(default 1 second)
When a message is released, if the downstream flow fails, the release will be attempted after the retryDelay
.
If the maxAttempts
is reached, the message is discarded (unless the release is transactional, in which case the message will remain in the store, but will no longer be scheduled for release, until the application is restarted, or the reschedulePersistedMessages()
method is invoked, as discussed above).
In addition, you can configure a delayedMessageErrorChannel
; when a release fails, an ErrorMessage
is sent to that channel with the exception as the payload and has the originalMessage
property.
The ErrorMessage
contains a header IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT
containing the current count.
If the error flow consumes the error message and exits normally, no further action is taken; if the release is transactional, the transaction will commit and the message deleted from the store.
If the error flow throws an exception, the release will be retried up to maxAttempts
as discussed above.
10.7. Scripting Support
Spring Integration 2.1 added support for the JSR223 Scripting for Java specification, introduced in Java version 6. It lets you use scripts written in any supported language (including Ruby, JRuby, Javascript, and Groovy) to provide the logic for various integration components, similar to the way the Spring Expression Language (SpEL) is used in Spring Integration. For more information about JSR223, see the documentation.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-scripting</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-scripting:5.1.5.RELEASE"
In addition you need to add a script engine implementation, e.g. JRuby, Jython.
| Note that this feature requires Java 6 or higher. |
In order to use a JVM scripting language, a JSR223 implementation for that language must be included in your class path. Java 6 natively supports Javascript. The Groovy and JRuby projects provide JSR233 support in their standard distributions.
| Various JSR223 language implementations have been developed by third parties. A particular implementation’s compatibility with Spring Integration depends on how well it conforms to the specification and the implementer’s interpretation of the specification. |
| If you plan to use Groovy as your scripting language, we recommended you use Spring-Integration’s Groovy Support as it offers additional features specific to Groovy. However, this section is relevant as well. |
10.7.1. Script Configuration
Depending on the complexity of your integration requirements, scripts may be provided inline as CDATA in XML configuration or as a reference to a Spring resource that contains the script.
To enable scripting support, Spring Integration defines a ScriptExecutingMessageProcessor
, which binds the message payload to a variable named payload
and the message headers to a headers
variable, both accessible within the script execution context.
All you need to do is write a script that uses these variables.
The following pair of examples show sample configurations that create filters:
<int:filter input-channel="referencedScriptInput">
<int-script:script lang="ruby" location="some/path/to/ruby/script/RubyFilterTests.rb"/>
</int:filter>
<int:filter input-channel="inlineScriptInput">
<int-script:script lang="groovy">
<![CDATA[
return payload == 'good'
]]>
</int-script:script>
</int:filter>
As the preceding examples show, the script can be included inline or can be included by reference to a resource location (by using the location
attribute).
Additionally, the lang
attribute corresponds to the language name (or its JSR223 alias)
Other Spring Integration endpoint elements that support scripting include router
, service-activator
, transformer
, and splitter
.
The scripting configuration in each case would be identical to the above (besides the endpoint element).
Another useful feature of scripting support is the ability to update (reload) scripts without having to restart the application context.
To do so, specify the refresh-check-delay
attribute on the script
element, as the following example shows:
<int-script:script location="..." refresh-check-delay="5000"/>
In the preceding example, the script location is checked for updates every 5 seconds. If the script is updated, any invocation that occurs later than 5 seconds since the update results in running the new script.
Consider the following example:
<int-script:script location="..." refresh-check-delay="0"/>
In the preceding example, the context is updated with any script modifications as soon as such modification occurs, providing a simple mechanism for 'real-time' configuration. Any negative value means the script is not reloaded after initialization of the application context. This is the default behavior. The following example shows a script that never updates:
<int-script:script location="..." refresh-check-delay="-1"/>
| Inline scripts can not be reloaded. |
Script Variable Bindings
Variable bindings are required to enable the script to reference variables externally provided to the script’s execution context.
By default, payload
and headers
are used as binding variables.
You can bind additional variables to a script by using <variable>
elements, as the following example shows:
<script:script lang="js" location="foo/bar/MyScript.js">
<script:variable name="foo" value="thing1"/>
<script:variable name="bar" value="thing2"/>
<script:variable name="date" ref="date"/>
</script:script>
As shown in the preceding example, you can bind a script variable either to a scalar value or to a Spring bean reference.
Note that payload
and headers
are still included as binding variables.
With Spring Integration 3.0, in addition to the variable
element, the variables
attribute has been introduced.
This attribute and the variable
elements are not mutually exclusive, and you can combine them within one script
component.
However, variables must be unique, regardless of where they are defined.
Also, since Spring Integration 3.0, variable bindings are allowed for inline scripts, too, as the following example shows:
<service-activator input-channel="input">
<script:script lang="ruby" variables="thing1=THING1, date-ref=dateBean">
<script:variable name="thing2" ref="thing2Bean"/>
<script:variable name="thing3" value="thing2"/>
<![CDATA[
payload.foo = thing1
payload.date = date
payload.bar = thing2
payload.baz = thing3
payload
]]>
</script:script>
</service-activator>
The preceding example shows a combination of an inline script, a variable
element, and a variables
attribute.
The variables
attribute contains a comma-separated value, where each segment contains an '=' separated pair of the variable and its value.
The variable name can be suffixed with -ref
, as in the date-ref
variable in the preceding example.
That means that the binding variable has the name, date
, but the value is a reference to the dateBean
bean from the application context.
This may be useful when using property placeholder configuration or command-line arguments.
If you need more control over how variables are generated, you can implement your own Java class that uses the ScriptVariableGenerator
strategy, which is defined by the following interface:
public interface ScriptVariableGenerator {
Map<String, Object> generateScriptVariables(Message<?> message);
}
This interface requires you to implement the generateScriptVariables(Message)
method.
The message argument lets you access any data available in the message payload and headers, and the return value is the Map
of bound variables.
This method is called every time the script is executed for a message.
The following example shows how to provide an implementation of ScriptVariableGenerator
and reference it with the script-variable-generator
attribute:
<int-script:script location="foo/bar/MyScript.groovy"
script-variable-generator="variableGenerator"/>
<bean id="variableGenerator" class="foo.bar.MyScriptVariableGenerator"/>
If a script-variable-generator
is not provided, script components use DefaultScriptVariableGenerator
, which merges any provided <variable>
elements with payload
and headers
variables from the Message
in its generateScriptVariables(Message)
method.
You cannot provide both the script-variable-generator attribute and <variable> element(s).
They are mutually exclusive.
|
10.8. Groovy support
In Spring Integration 2.0, we added Groovy support, letting you use the Groovy scripting language to provide the logic for various integration components — similar to the way the Spring Expression Language (SpEL) is supported for routing, transformation, and other integration concerns. For more information about Groovy, see the Groovy documentation, which you can find on the project website.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-groovy</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-groovy:5.1.5.RELEASE"
10.8.1. Groovy Configuration
With Spring Integration 2.1, the configuration namespace for the Groovy support is an extension of Spring Integration’s scripting support and shares the core configuration and behavior described in detail in the Scripting Support section.
Even though Groovy scripts are well supported by generic scripting support, the Groovy support provides the Groovy
configuration namespace, which is backed by the Spring Framework’s org.springframework.scripting.groovy.GroovyScriptFactory
and related components, offering extended capabilities for using Groovy.
The following listing shows two sample configurations:
<int:filter input-channel="referencedScriptInput">
<int-groovy:script location="some/path/to/groovy/file/GroovyFilterTests.groovy"/>
</int:filter>
<int:filter input-channel="inlineScriptInput">
<int-groovy:script><![CDATA[
return payload == 'good'
]]></int-groovy:script>
</int:filter>
As the preceding examples show, the configuration looks identical to the general scripting support configuration.
The only difference is the use of the Groovy namespace, as indicated by the int-groovy
namespace prefix.
Also note that the lang
attribute on the <script>
tag is not valid in this namespace.
10.8.2. Groovy Object Customization
If you need to customize the Groovy object itself (beyond setting variables) you can reference a bean that implements GroovyObjectCustomizer
by using the customizer
attribute.
For example, this might be useful if you want to implement a domain-specific language (DSL) by modifying the MetaClass
and registering functions to be available within the script.
The following example shows how to do so:
<int:service-activator input-channel="groovyChannel">
<int-groovy:script location="somewhere/SomeScript.groovy" customizer="groovyCustomizer"/>
</int:service-activator>
<beans:bean id="groovyCustomizer" class="org.something.MyGroovyObjectCustomizer"/>
Setting a custom GroovyObjectCustomizer
is not mutually exclusive with <variable>
elements or the script-variable-generator
attribute.
It can also be provided when defining an inline script.
Spring Integration 3.0 introduced the variables
attribute, which works in conjunction with the variable
element.
Also, groovy scripts have the ability to resolve a variable to a bean in the BeanFactory
, if a binding variable was not provided with the name.
The following example shows how to use a variable (entityManager
):
<int-groovy:script>
<![CDATA[
entityManager.persist(payload)
payload
]]>
</int-groovy:script>
entityManager
must be an appropriate bean in the application context.
For more information regarding the <variable>
element, the variables
attribute, and the script-variable-generator
attribute, see Script Variable Bindings.
10.8.3. Groovy Script Compiler Customization
The @CompileStatic
hint is the most popular Groovy compiler customization option.
It can be used on the class or method level.
For more information, see the Groovy Reference Manual and, specifically, @CompileStatic.
To utilize this feature for short scripts (in integration scenarios), we are forced to change simple scripts to more Java-like code.
Consider the following <filter>
script:
headers.type == 'good'
The preceding script becomes the following method in Spring Integration:
@groovy.transform.CompileStatic
String filter(Map headers) {
headers.type == 'good'
}
filter(headers)
With that, the filter()
method is transformed and compiled to static Java code, bypassing the Groovy
dynamic phases of invocation, such as getProperty()
factories and CallSite
proxies.
Starting with version 4.3, you can configure the Spring Integration Groovy components with the compile-static
boolean
option, specifying that ASTTransformationCustomizer
for @CompileStatic
should be added to the internal CompilerConfiguration
.
With that in place, you can omit the method declaration with @CompileStatic
in our script code and still get compiled
plain Java code.
In this case, the preceding script can be short but still needs to be a little more verbose than interpreted script, as the following example shows:
binding.variables.headers.type == 'good'
You must access the headers
and payload
(or any other) variables through the groovy.lang.Script
binding
property because, with @CompileStatic
, we do not have the dynamic GroovyObject.getProperty()
capability.
In addition, we introduced the compiler-configuration
bean reference.
With this attribute, you can provide any other required Groovy compiler customizations, such as ImportCustomizer
.
For more information about this feature, see the Groovy Documentation for advanced compiler configuration.
Using compilerConfiguration does not automatically add an ASTTransformationCustomizer for the @CompileStatic annotation, and it overrides the compileStatic option.
If you still need CompileStatic , you should manually add a new ASTTransformationCustomizer(CompileStatic.class) into the CompilationCustomizers of that custom compilerConfiguration .
|
The Groovy compiler customization does not have any effect on the refresh-check-delay option, and reloadable scripts can be statically compiled, too.
|
10.8.4. Control Bus
As described in (Enterprise Integration Patterns), the idea behind the control bus is that you can use the same messaging system for monitoring and managing the components within the framework as is used for “application-level” messaging. In Spring Integration, we build upon the adapters described earlier so that you can send Messages as a means of invoking exposed operations. One option for those operations is Groovy scripts. The following example configures a Groovy script for the control bus:
<int-groovy:control-bus input-channel="operationChannel"/>
The control bus has an input channel that can be accessed to invoke operations on the beans in the application context.
The Groovy control bus runs messages on the input channel as Groovy scripts.
It takes a message, compiles the body to a script, customizes it with a GroovyObjectCustomizer
, and runs it.
The control bus' MessageProcessor
exposes all beans in the application context that are annotated with @ManagedResource
and implement Spring’s Lifecycle
interface or extend Spring’s CustomizableThreadCreator
base class (for example, several of the TaskExecutor
and TaskScheduler
implementations).
Be careful about using managed beans with custom scopes (such as 'request') in the Control Bus' command scripts, especially inside an asynchronous message flow.
If MessageProcessor of the control bus cannot expose a bean from the application context, you may end up with some BeansException during the command script’s run.
For example, if a custom scope’s context is not established, the attempt to get a bean within that scope triggers a BeanCreationException .
|
If you need to further customize the Groovy objects, you can also provide a reference to a bean that implements GroovyObjectCustomizer
through the customizer
attribute, as the following example shows:
<int-groovy:control-bus input-channel="input"
output-channel="output"
customizer="groovyCustomizer"/>
<beans:bean id="groovyCustomizer" class="org.foo.MyGroovyObjectCustomizer"/>
10.9. Adding Behavior to Endpoints
Prior to Spring Integration 2.2, you could add behavior to an entire Integration flow by adding an AOP Advice to a poller’s <advice-chain/>
element.
However, suppose you want to retry, say, just a REST Web Service call, and not any downstream endpoints.
For example, consider the following flow:
inbound-adapter->poller->http-gateway1->http-gateway2->jdbc-outbound-adapter
If you configure some retry-logic into an advice chain on the poller and the call to http-gateway2
failed because of a network glitch, the retry causes both http-gateway1
and http-gateway2
to be called a second time.
Similarly, after a transient failure in the jdbc-outbound-adapter, both HTTP gateways are called a second time before again calling the jdbc-outbound-adapter
.
Spring Integration 2.2 adds the ability to add behavior to individual endpoints.
This is achieved by the addition of the <request-handler-advice-chain/>
element to many endpoints.
The following example shows how to the <request-handler-advice-chain/>
element within an outbound-gateway
:
<int-http:outbound-gateway id="withAdvice"
url-expression="'http://localhost/test1'"
request-channel="requests"
reply-channel="nextChannel">
<int:request-handler-advice-chain>
<ref bean="myRetryAdvice" />
</request-handler-advice-chain>
</int-http:outbound-gateway>
In this case, myRetryAdvice
is applied only locally to this gateway and does not apply to further actions taken downstream after the reply is sent to nextChannel
.
The scope of the advice is limited to the endpoint itself.
|
At this time, you cannot advise an entire However, a |
10.9.1. Provided Advice Classes
In addition to providing the general mechanism to apply AOP advice classes, Spring Integration provides three standard advice classes:
-
RequestHandlerRetryAdvice
(described in Retry Advice) -
RequestHandlerCircuitBreakerAdvice
(described in Circuit Breaker Advice) -
ExpressionEvaluatingRequestHandlerAdvice
(described in Expression Evaluating Advice)
Retry Advice
The retry advice (o.s.i.handler.advice.RequestHandlerRetryAdvice
) leverages the rich retry mechanisms provided by the Spring Retry project.
The core component of spring-retry
is the RetryTemplate
, which allows configuration of sophisticated retry scenarios, including RetryPolicy
and BackoffPolicy
strategies (with a number of implementations) as well as a RecoveryCallback
strategy to determine the action to take when retries are exhausted.
- Stateless Retry
-
Stateless retry is the case where the retry activity is handled entirely within the advice. The thread pauses (if configured to do so) and retries the action.
- Stateful Retry
-
Stateful retry is the case where the retry state is managed within the advice but where an exception is thrown and the caller resubmits the request. An example for stateful retry is when we want the message originator (for example,JMS) to be responsible for resubmitting, rather than performing it on the current thread. Stateful retry needs some mechanism to detect a retried submission.
For more information on spring-retry
, see the project’s Javadoc and the reference documentation for Spring Batch, where spring-retry
originated.
| The default back off behavior is to not back off. Retries are attempted immediately. Using a back off policy that causes threads to pause between attempts may cause performance issues, including excessive memory use and thread starvation. In high-volume environments, back off policies should be used with caution. |
Configuring the Retry Advice
The examples in this section use the following <service-activator>
that always throws an exception:
public class FailingService {
public void service(String message) {
throw new RuntimeException("error");
}
}
- Simple Stateless Retry
-
The default
RetryTemplate
has aSimpleRetryPolicy
which tries three times. There is noBackOffPolicy
, so the three attempts are made back-to-back-to-back with no delay between attempts. There is noRecoveryCallback
, so the result is to throw the exception to the caller after the final failed retry occurs. In a Spring Integration environment, this final exception might be handled by using anerror-channel
on the inbound endpoint. The following example usesRetryTemplate
and shows itsDEBUG
output:<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"/> </request-handler-advice-chain> </int:service-activator> DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...] DEBUG [task-scheduler-2]Retry: count=0 DEBUG [task-scheduler-2]Checking for rethrow: count=1 DEBUG [task-scheduler-2]Retry: count=1 DEBUG [task-scheduler-2]Checking for rethrow: count=2 DEBUG [task-scheduler-2]Retry: count=2 DEBUG [task-scheduler-2]Checking for rethrow: count=3 DEBUG [task-scheduler-2]Retry failed last attempt: count=3
- Simple Stateless Retry with Recovery
-
The following example adds a
RecoveryCallback
to the preceding example and uses anErrorMessageSendingRecoverer
to send anErrorMessage
to a channel:<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"> <property name="recoveryCallback"> <bean class="o.s.i.handler.advice.ErrorMessageSendingRecoverer"> <constructor-arg ref="myErrorChannel" /> </bean> </property> </bean> </request-handler-advice-chain> </int:int:service-activator> DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...] DEBUG [task-scheduler-2]Retry: count=0 DEBUG [task-scheduler-2]Checking for rethrow: count=1 DEBUG [task-scheduler-2]Retry: count=1 DEBUG [task-scheduler-2]Checking for rethrow: count=2 DEBUG [task-scheduler-2]Retry: count=2 DEBUG [task-scheduler-2]Checking for rethrow: count=3 DEBUG [task-scheduler-2]Retry failed last attempt: count=3 DEBUG [task-scheduler-2]Sending ErrorMessage :failedMessage:[Payload=...]
- Stateless Retry with Customized Policies, and Recovery
-
For more sophistication, we can provide the advice with a customized
RetryTemplate
. This example continues to use theSimpleRetryPolicy
but increases the attempts to four. It also adds anExponentialBackoffPolicy
where the first retry waits one second, the second waits five seconds and the third waits 25 (for four attempts in all). The following listing shows the example and itsDEBUG
output:<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"> <property name="recoveryCallback"> <bean class="o.s.i.handler.advice.ErrorMessageSendingRecoverer"> <constructor-arg ref="myErrorChannel" /> </bean> </property> <property name="retryTemplate" ref="retryTemplate" /> </bean> </request-handler-advice-chain> </int:service-activator> <bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate"> <property name="retryPolicy"> <bean class="org.springframework.retry.policy.SimpleRetryPolicy"> <property name="maxAttempts" value="4" /> </bean> </property> <property name="backOffPolicy"> <bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy"> <property name="initialInterval" value="1000" /> <property name="multiplier" value="5.0" /> <property name="maxInterval" value="60000" /> </bean> </property> </bean> 27.058 DEBUG [task-scheduler-1]preSend on channel 'input', message: [Payload=...] 27.071 DEBUG [task-scheduler-1]Retry: count=0 27.080 DEBUG [task-scheduler-1]Sleeping for 1000 28.081 DEBUG [task-scheduler-1]Checking for rethrow: count=1 28.081 DEBUG [task-scheduler-1]Retry: count=1 28.081 DEBUG [task-scheduler-1]Sleeping for 5000 33.082 DEBUG [task-scheduler-1]Checking for rethrow: count=2 33.082 DEBUG [task-scheduler-1]Retry: count=2 33.083 DEBUG [task-scheduler-1]Sleeping for 25000 58.083 DEBUG [task-scheduler-1]Checking for rethrow: count=3 58.083 DEBUG [task-scheduler-1]Retry: count=3 58.084 DEBUG [task-scheduler-1]Checking for rethrow: count=4 58.084 DEBUG [task-scheduler-1]Retry failed last attempt: count=4 58.086 DEBUG [task-scheduler-1]Sending ErrorMessage :failedMessage:[Payload=...]
- Namespace Support for Stateless Retry
-
Starting with version 4.0, the preceding configuration can be greatly simplified, thanks to the namespace support for the retry advice, as the following example shows:
<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean ref="retrier" /> </request-handler-advice-chain> </int:service-activator> <int:handler-retry-advice id="retrier" max-attempts="4" recovery-channel="myErrorChannel"> <int:exponential-back-off initial="1000" multiplier="5.0" maximum="60000" /> </int:handler-retry-advice>
In the preceding example, the advice is defined as a top-level bean so that it can be used in multiple
request-handler-advice-chain
instances. You can also define the advice directly within the chain, as the following example shows:<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <int:retry-advice id="retrier" max-attempts="4" recovery-channel="myErrorChannel"> <int:exponential-back-off initial="1000" multiplier="5.0" maximum="60000" /> </int:retry-advice> </request-handler-advice-chain> </int:service-activator>
A
<handler-retry-advice>
can have a<fixed-back-off>
or<exponential-back-off>
child element or have no child element.  If there is norecovery-channel
, the exception is thrown when retries are exhausted. The namespace can only be used with stateless retry.For more complex environments (custom policies etc), use normal
<bean>
definitions. - Simple Stateful Retry with Recovery
-
To make retry stateful, we need to provide the advice with a
RetryStateGenerator
implementation. This class is used to identify a message as being a resubmission so that theRetryTemplate
can determine the current state of retry for this message. The framework provides aSpelExpressionRetryStateGenerator
, which determines the message identifier by using a SpEL expression. This example again uses the default policies (three attempts with no back off). As with stateless retry, these policies can be customized. The following listing shows the example and itsDEBUG
output:<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"> <property name="retryStateGenerator"> <bean class="o.s.i.handler.advice.SpelExpressionRetryStateGenerator"> <constructor-arg value="headers['jms_messageId']" /> </bean> </property> <property name="recoveryCallback"> <bean class="o.s.i.handler.advice.ErrorMessageSendingRecoverer"> <constructor-arg ref="myErrorChannel" /> </bean> </property> </bean> </int:request-handler-advice-chain> </int:service-activator> 24.351 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 24.368 DEBUG [Container#0-1]Retry: count=0 24.387 DEBUG [Container#0-1]Checking for rethrow: count=1 24.387 DEBUG [Container#0-1]Rethrow in retry for policy: count=1 24.387 WARN [Container#0-1]failure occurred in gateway sendAndReceive org.springframework.integration.MessagingException: Failed to invoke handler ... Caused by: java.lang.RuntimeException: foo ... 24.391 DEBUG [Container#0-1]Initiating transaction rollback on application exception ... 25.412 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 25.412 DEBUG [Container#0-1]Retry: count=1 25.413 DEBUG [Container#0-1]Checking for rethrow: count=2 25.413 DEBUG [Container#0-1]Rethrow in retry for policy: count=2 25.413 WARN [Container#0-1]failure occurred in gateway sendAndReceive org.springframework.integration.MessagingException: Failed to invoke handler ... Caused by: java.lang.RuntimeException: foo ... 25.414 DEBUG [Container#0-1]Initiating transaction rollback on application exception ... 26.418 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 26.418 DEBUG [Container#0-1]Retry: count=2 26.419 DEBUG [Container#0-1]Checking for rethrow: count=3 26.419 DEBUG [Container#0-1]Rethrow in retry for policy: count=3 26.419 WARN [Container#0-1]failure occurred in gateway sendAndReceive org.springframework.integration.MessagingException: Failed to invoke handler ... Caused by: java.lang.RuntimeException: foo ... 26.420 DEBUG [Container#0-1]Initiating transaction rollback on application exception ... 27.425 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 27.426 DEBUG [Container#0-1]Retry failed last attempt: count=3 27.426 DEBUG [Container#0-1]Sending ErrorMessage :failedMessage:[Payload=...]
If you compare the preceding example with the stateless examples, you can see that, with stateful retry, the exception is thrown to the caller on each failure.
- Exception Classification for Retry
-
Spring Retry has a great deal of flexibility for determining which exceptions can invoke retry. The default configuration retries for all exceptions and the exception classifier looks at the top-level exception. If you configure it to, say, retry only on
MyException
and your application throws aSomeOtherException
where the cause is aMyException
, retry does not occur.Since Spring Retry 1.0.3, the
BinaryExceptionClassifier
has a property calledtraverseCauses
(the default isfalse
). Whentrue
, it traverses exception causes until it finds a match or runs out of causes to traverse.To use this classifier for retry, use a
SimpleRetryPolicy
created with the constructor that takes the max attempts, theMap
ofException
objects, and thetraverseCauses
boolean. Then you can inject this policy into theRetryTemplate
.
Circuit Breaker Advice
The general idea of the circuit breaker pattern is that, if a service is not currently available, do not waste time (and resources) trying to use it.
The o.s.i.handler.advice.RequestHandlerCircuitBreakerAdvice
implements this pattern.
When the circuit breaker is in the closed state, the endpoint attempts to invoke the service.
The circuit breaker goes to the open state if a certain number of consecutive attempts fail.
When it is in the open state, new requests “fail fast” and no attempt is made to invoke the service until some time has expired.
When that time has expired, the circuit breaker is set to the half-open state. When in this state, if even a single attempt fails, the breaker immediately goes to the open state. If the attempt succeeds, the breaker goes to the closed state, in which case it does not go to the open state again until the configured number of consecutive failures again occur. Any successful attempt resets the state to zero failures for the purpose of determining when the breaker might go to the open state again.
Typically, this advice might be used for external services, where it might take some time to fail (such as a timeout attempting to make a network connection).
The RequestHandlerCircuitBreakerAdvice
has two properties: threshold
and halfOpenAfter
.
The threshold
property represents the number of consecutive failures that need to occur before the breaker goes open.
It defaults to 5
.
The halfOpenAfter
property represents the time after the last failure that the breaker waits before attempting another request.
The default is 1000 milliseconds.
The following example configures a circuit breaker and shows its DEBUG
and ERROR
output:
<int:service-activator input-channel="input" ref="failer" method="service">
<int:request-handler-advice-chain>
<bean class="o.s.i.handler.advice.RequestHandlerCircuitBreakerAdvice">
<property name="threshold" value="2" />
<property name="halfOpenAfter" value="12000" />
</bean>
</int:request-handler-advice-chain>
</int:service-activator>
05.617 DEBUG [task-scheduler-1]preSend on channel 'input', message: [Payload=...]
05.638 ERROR [task-scheduler-1]org.springframework.messaging.MessageHandlingException: java.lang.RuntimeException: foo
...
10.598 DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...]
10.600 ERROR [task-scheduler-2]org.springframework.messaging.MessageHandlingException: java.lang.RuntimeException: foo
...
15.598 DEBUG [task-scheduler-3]preSend on channel 'input', message: [Payload=...]
15.599 ERROR [task-scheduler-3]org.springframework.messaging.MessagingException: Circuit Breaker is Open for ServiceActivator
...
20.598 DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...]
20.598 ERROR [task-scheduler-2]org.springframework.messaging.MessagingException: Circuit Breaker is Open for ServiceActivator
...
25.598 DEBUG [task-scheduler-5]preSend on channel 'input', message: [Payload=...]
25.601 ERROR [task-scheduler-5]org.springframework.messaging.MessageHandlingException: java.lang.RuntimeException: foo
...
30.598 DEBUG [task-scheduler-1]preSend on channel 'input', message: [Payload=foo...]
30.599 ERROR [task-scheduler-1]org.springframework.messaging.MessagingException: Circuit Breaker is Open for ServiceActivator
In the preceding example, the threshold is set to 2
and halfOpenAfter
is set to 12
seconds.
A new request arrives every 5 seconds.
The first two attempts invoked the service.
The third and fourth failed with an exception indicating that the circuit breaker is open.
The fifth request was attempted because the request was 15 seconds after the last failure.
The sixth attempt fails immediately because the breaker immediately went to open.
Expression Evaluating Advice
The final supplied advice class is the o.s.i.handler.advice.ExpressionEvaluatingRequestHandlerAdvice
.
This advice is more general than the other two advices.
It provides a mechanism to evaluate an expression on the original inbound message sent to the endpoint.
Separate expressions are available to be evaluated, after either success or failure.
Optionally, a message containing the evaluation result, together with the input message, can be sent to a message channel.
A typical use case for this advice might be with an <ftp:outbound-channel-adapter/>
, perhaps to move the file to one directory if the transfer was successful or to another directory if it fails:
The advice has properties to set an expression when successful, an expression for failures, and corresponding channels for each.
For the successful case, the message sent to the successChannel
is an AdviceMessage
, with the payload being the result of the expression evaluation.
An additional property, called inputMessage
, contains the original message sent to the handler.
A message sent to the failureChannel
(when the handler throws an exception) is an ErrorMessage
with a payload of MessageHandlingExpressionEvaluatingAdviceException
.
Like all MessagingException
instances, this payload has failedMessage
and cause
properties, as well as an additional property called evaluationResult
, which contains the result of the expression evaluation.
Starting with version 5.1.3, if channels are configured, but expressions are not provided, the default expression is used to evaluate to the payload of the message.
|
When an exception is thrown in the scope of the advice, by default, that exception is thrown to the caller after any failureExpression
is evaluated.
If you wish to suppress throwing the exception, set the trapException
property to true
.
The following advice shows how to configure an advice with Java DSL:
@SpringBootApplication
public class EerhaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context = SpringApplication.run(EerhaApplication.class, args);
MessageChannel in = context.getBean("advised.input", MessageChannel.class);
in.send(new GenericMessage<>("good"));
in.send(new GenericMessage<>("bad"));
context.close();
}
@Bean
public IntegrationFlow advised() {
return f -> f.handle((GenericHandler<String>) (payload, headers) -> {
if (payload.equals("good")) {
return null;
}
else {
throw new RuntimeException("some failure");
}
}, c -> c.advice(expressionAdvice()));
}
@Bean
public Advice expressionAdvice() {
ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
advice.setSuccessChannelName("success.input");
advice.setOnSuccessExpressionString("payload + ' was successful'");
advice.setFailureChannelName("failure.input");
advice.setOnFailureExpressionString(
"payload + ' was bad, with reason: ' + #exception.cause.message");
advice.setTrapException(true);
return advice;
}
@Bean
public IntegrationFlow success() {
return f -> f.handle(System.out::println);
}
@Bean
public IntegrationFlow failure() {
return f -> f.handle(System.out::println);
}
}
10.9.2. Custom Advice Classes
In addition to the provided advice classes described earlier, you can implement your own advice classes.
While you can provide any implementation of org.aopalliance.aop.Advice
(usually org.aopalliance.intercept.MethodInterceptor
), we generally recommend that you subclass o.s.i.handler.advice.AbstractRequestHandlerAdvice
.
This has the benefit of avoiding the writing of low-level aspect-oriented programming code as well as providing a starting point that is specifically tailored for use in this environment.
Subclasses need to implement the doInvoke()`
method, the definition of which follows:
/**
* Subclasses implement this method to apply behavior to the {@link MessageHandler} callback.execute()
* invokes the handler method and returns its result, or null).
* @param callback Subclasses invoke the execute() method on this interface to invoke the handler method.
* @param target The target handler.
* @param message The message that will be sent to the handler.
* @return the result after invoking the {@link MessageHandler}.
* @throws Exception
*/
protected abstract Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception;
The callback parameter is a convenience to avoid subclasses that deal with AOP directly.
Invoking the callback.execute()
method invokes the message handler.
The target
parameter is provided for those subclasses that need to maintain state for a specific handler, perhaps by maintaining that state in a Map
keyed by the target.
This feature allows the same advice to be applied to multiple handlers.
The RequestHandlerCircuitBreakerAdvice
uses advice this to keep circuit breaker state for each handler.
The message
parameter is the message sent to the handler.
While the advice cannot modify the message before invoking the handler, it can modify the payload (if it has mutable properties).
Typically, an advice would use the message for logging or to send a copy of the message somewhere before or after invoking the handler.
The return value would normally be the value returned by callback.execute()
.
However, the advice does have the ability to modify the return value.
Note that only AbstractReplyProducingMessageHandler
instances return values.
The following example shows a custom advice class that extends AbstractRequestHandlerAdvice
:
public class MyAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
// add code before the invocation
Object result = callback.execute();
// add code after the invocation
return result;
}
}
|
In addition to the For more information, see the ReflectiveMethodInvocation Javadoc. |
10.9.3. Other Advice Chain Elements
While the abstract class mentioned above is a convenience, you can add any Advice
, including a transaction advice, to the chain.
10.9.4. Handling Message Advice
As discussed in the introduction to this section, advice objects in a request handler advice chain are applied to just the current endpoint, not the downstream flow (if any).
For MessageHandler
objects that produce a reply (such as those that extend AbstractReplyProducingMessageHandler
), the advice is applied to an internal method: handleRequestMessage()
(called from MessageHandler.handleMessage()
).
For other message handlers, the advice is applied to MessageHandler.handleMessage()
.
There are some circumstances where, even if a message handler is an AbstractReplyProducingMessageHandler
, the advice must be applied to the handleMessage
method.
For example, the idempotent receiver might return null
, which would cause an exception if the handler’s replyRequired
property is set to true
.
Another example is the BoundRabbitChannelAdvice
— see Strict Message Ordering.
Starting with version 4.3.1, a new HandleMessageAdvice
interface and its base implementation (AbstractHandleMessageAdvice
) have been introduced.
Advice
objects that implement HandleMessageAdvice
are always applied to the handleMessage()
method, regardless of the handler type.
It is important to understand that HandleMessageAdvice
implementations (such as idempotent receiver), when applied to a handlers that return responses, are dissociated from the adviceChain
and properly applied to the MessageHandler.handleMessage()
method.
| Because of this disassociation, the advice chain order is not honored. |
Consider the following configuration:
<some-reply-producing-endpoint ... >
<int:request-handler-advice-chain>
<tx:advice ... />
<bean ref="myHandleMessageAdvice" />
</int:request-handler-advice-chain>
</some-reply-producing-endpoint>
In the preceding example, the <tx:advice>
is applied to the AbstractReplyProducingMessageHandler.handleRequestMessage()
.
However, myHandleMessageAdvice
is applied for to MessageHandler.handleMessage()
.
Therefore, it is invoked before the <tx:advice>
.
To retain the order, you should follow the standard Spring AOP configuration approach and use an endpoint id
together with the .handler
suffix to obtain the target MessageHandler
bean.
Note that, in that case, the entire downstream flow is within the transaction scope.
In the case of a MessageHandler
that does not return a response, the advice chain order is retained.
10.9.5. Transaction Support
Starting with version 5.0, a new TransactionHandleMessageAdvice
has been introduced to make the whole downstream flow transactional, thanks to the HandleMessageAdvice
implementation.
When a regular TransactionInterceptor
is used in the <request-handler-advice-chain>
element (for example, through configuring <tx:advice>
), a started transaction is only applied only for an internal AbstractReplyProducingMessageHandler.handleRequestMessage()
and is not propagated to the downstream flow.
To simplify XML configuration, along with the <request-handler-advice-chain>
, a <transactional>
element has been added to all <outbound-gateway>
and <service-activator>
and related components.
The following example shows <transactional>
in use:
<int-rmi:outbound-gateway remote-channel="foo" host="localhost"
request-channel="good" reply-channel="reply" port="#{@port}">
<int-rmi:transactional/>
</int-rmi:outbound-gateway>
<bean id="transactionManager" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.transaction.PlatformTransactionManager"/>
</bean>
If you are familiar with the JPA integration components, such a configuration is not new, but now we can start a transaction from any point in our flow — not only from the <poller>
or a message-driven channel adapter such as JMS.
Java configuration can be simplified by using the TransactionInterceptorBuilder
, and the result bean name can be used in the messaging annotations adviceChain
attribute, as the following example shows:
@Bean
public ConcurrentMetadataStore store() {
return new SimpleMetadataStore(hazelcastInstance()
.getMap("idempotentReceiverMetadataStore"));
}
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
return new IdempotentReceiverInterceptor(
new MetadataStoreSelector(
message -> message.getPayload().toString(),
message -> message.getPayload().toString().toUpperCase(), store()));
}
@Bean
public TransactionInterceptor transactionInterceptor() {
return new TransactionInterceptorBuilder(true)
.transactionManager(this.transactionManager)
.isolation(Isolation.READ_COMMITTED)
.propagation(Propagation.REQUIRES_NEW)
.build();
}
@Bean
@org.springframework.integration.annotation.Transformer(inputChannel = "input",
outputChannel = "output",
adviceChain = { "idempotentReceiverInterceptor",
"transactionInterceptor" })
public Transformer transformer() {
return message -> message;
}
Note the true
parameter on the TransactionInterceptorBuilder
constructor.
It causes the creation of a TransactionHandleMessageAdvice
, not a regular TransactionInterceptor
.
Java DSL supports an Advice
through the .transactional()
options on the endpoint configuration, as the following example shows:
@Bean
public IntegrationFlow updatingGatewayFlow() {
return f -> f
.handle(Jpa.updatingGateway(this.entityManagerFactory),
e -> e.transactional(true))
.channel(c -> c.queue("persistResults"));
}
10.9.6. Advising Filters
There is an additional consideration when advising Filter
advices.
By default, any discard actions (when the filter returns false
) are performed within the scope of the advice chain.
This could include all the flow downstream of the discard channel.
So, for example, if an element downstream of the discard channel throws an exception and there is a retry advice, the process is retried.
Also, if throwExceptionOnRejection
is set to true
(the exception is thrown within the scope of the advice).
Setting discard-within-advice
to false
modifies this behavior and the discard (or exception) occurs after the advice chain is called.
10.9.7. Advising Endpoints Using Annotations
When configuring certain endpoints by using annotations (@Filter
, @ServiceActivator
, @Splitter
, and @Transformer
), you can supply a bean name for the advice chain in the adviceChain
attribute.
In addition, the @Filter
annotation also has the discardWithinAdvice
attribute, which can be used to configure the discard behavior, as discussed in Advising Filters.
The following example causes the discard to be performed after the advice:
@MessageEndpoint
public class MyAdvisedFilter {
@Filter(inputChannel="input", outputChannel="output",
adviceChain="adviceChain", discardWithinAdvice="false")
public boolean filter(String s) {
return s.contains("good");
}
}
10.9.8. Ordering Advices within an Advice Chain
Advice classes are “around” advices and are applied in a nested fashion. The first advice is the outermost, while the last advice is the innermost (that is, closest to the handler being advised). It is important to put the advice classes in the correct order to achieve the functionality you desire.
For example, suppose you want to add a retry advice and a transaction advice.
You may want to place the retry advice advice first, followed by the transaction advice.
Consequently, each retry is performed in a new transaction.
On the other hand, if you want all the attempts and any recovery operations (in the retry RecoveryCallback
) to be scoped within the transaction, you could put the transaction advice first.
10.9.9. Advised Handler Properties
Sometimes, it is useful to access handler properties from within the advice.
For example, most handlers implement NamedComponent
to let you access the component name.
The target object can be accessed through the target
argument (when subclassing AbstractRequestHandlerAdvice
) or
invocation.getThis()
(when implementing org.aopalliance.intercept.MethodInterceptor
).
When the entire handler is advised (such as when the handler does not produce replies or the advice implements HandleMessageAdvice
), you can cast the target object to an interface, such as NamedComponent
, as shown in the following example:
String componentName = ((NamedComponent) target).getComponentName();
When you implement MethodInterceptor
directly, you could cast the target object as follows:
String componentName = ((NamedComponent) invocation.getThis()).getComponentName();
When only the handleRequestMessage()
method is advised (in a reply-producing handler), you need to access the
full handler, which is an AbstractReplyProducingMessageHandler
.
The following example shows how to do so:
AbstractReplyProducingMessageHandler handler =
((AbstractReplyProducingMessageHandler.RequestHandler) target).getAdvisedHandler();
String componentName = handler.getComponentName();
10.9.10. Idempotent Receiver Enterprise Integration Pattern
Starting with version 4.1, Spring Integration provides an implementation of the Idempotent Receiver Enterprise Integration Pattern.
It is a functional pattern and the whole idempotency logic should be implemented in the application.
However, to simplify the decision-making, the IdempotentReceiverInterceptor
component is provided.
This is an AOP Advice
that is applied to the MessageHandler.handleMessage()
method and that can filter
a request message or mark it as a duplicate
, according to its configuration.
Previously, you could have implemented this pattern by using a custom MessageSelector
in a <filter/>
(see Filter), for example.
However, since this pattern really defines the behavior of an endpoint rather than being an endpoint itself, the idempotent receiver implementation does not provide an endpoint component.
Rather, it is applied to endpoints declared in the application.
The logic of the IdempotentReceiverInterceptor
is based on the provided MessageSelector
and, if the message is not accepted by that selector, it is enriched with the duplicateMessage
header set to true
.
The target MessageHandler
(or downstream flow) can consult this header to implement the correct idempotency logic.
If the IdempotentReceiverInterceptor
is configured with a discardChannel
or throwExceptionOnRejection = true
, the duplicate message is not sent to the target MessageHandler.handleMessage()
.
Rather, it is discarded.
If you want to discard (do nothing with) the duplicate message, the discardChannel
should be configured with a NullChannel
, such as the default nullChannel
bean.
To maintain state between messages and provide the ability to compare messages for the idempotency, we provide the MetadataStoreSelector
.
It accepts a MessageProcessor
implementation (which creates a lookup key based on the Message
) and an optional ConcurrentMetadataStore
(Metadata Store).
See the MetadataStoreSelector
Javadoc for more information.
You can also customize the value
for ConcurrentMetadataStore
by using an additional MessageProcessor
.
By default, MetadataStoreSelector
uses the timestamp
message header.
For convenience, the MetadataStoreSelector
options are configurable directly on the <idempotent-receiver>
component.
The following listing shows all the possible attributes:
<idempotent-receiver
id="" (1)
endpoint="" (2)
selector="" (3)
discard-channel="" (4)
metadata-store="" (5)
key-strategy="" (6)
key-expression="" (7)
value-strategy="" (8)
value-expression="" (9)
throw-exception-on-rejection="" /> (10)
| 1 | The ID of the IdempotentReceiverInterceptor bean.
Optional. |
| 2 | Consumer endpoint name(s) or pattern(s) to which this interceptor is applied.
Separate names (patterns) with commas (, ), such as endpoint="aaa, bbb*, ccc, *ddd, eee*fff" .
Endpoint bean names matching these patterns are then used to retrieve the target endpoint’s MessageHandler bean (using its .handler suffix), and the IdempotentReceiverInterceptor is applied to those beans.
Required. |
| 3 | A MessageSelector bean reference.
Mutually exclusive with metadata-store and key-strategy (key-expression) .
When selector is not provided, one of key-strategy or key-strategy-expression is required. |
| 4 | Identifies the channel to which to send a message when the IdempotentReceiverInterceptor does not accept it.
When omitted, duplicate messages are forwarded to the handler with a duplicateMessage header.
Optional. |
| 5 | A ConcurrentMetadataStore reference.
Used by the underlying MetadataStoreSelector .
Mutually exclusive with selector .
Optional.
The default MetadataStoreSelector uses an internal SimpleMetadataStore that does not maintain state across application executions. |
| 6 | A MessageProcessor reference.
Used by the underlying MetadataStoreSelector .
Evaluates an idempotentKey from the request message.
Mutually exclusive with selector and key-expression .
When a selector is not provided, one of key-strategy or key-strategy-expression is required. |
| 7 | A SpEL expression to populate an ExpressionEvaluatingMessageProcessor .
Used by the underlying MetadataStoreSelector .
Evaluates an idempotentKey by using the request message as the evaluation context root object.
Mutually exclusive with selector and key-strategy .
When a selector is not provided, one of key-strategy or key-strategy-expression is required. |
| 8 | A MessageProcessor reference.
Used by the underlying MetadataStoreSelector .
Evaluates a value for the idempotentKey from the request message.
Mutually exclusive with selector and value-expression .
By default, the 'MetadataStoreSelector' uses the 'timestamp' message header as the Metadata 'value'. |
| 9 | A SpEL expression to populate an ExpressionEvaluatingMessageProcessor .
Used by the underlying MetadataStoreSelector .
Evaluates a value for the idempotentKey by using the request message as the evaluation context root object.
Mutually exclusive with selector and value-strategy .
By default, the 'MetadataStoreSelector' uses the 'timestamp' message header as the metadata 'value'. |
| 10 | Whether to throw an exception if the IdempotentReceiverInterceptor rejects the message.
Defaults to false .
It is applied regardless of whether or not a discard-channel is provided. |
For Java configuration, Spring Integration provides the method-level @IdempotentReceiver
annotation.
It is used to mark a method
that has a messaging annotation (@ServiceActivator
, @Router, and others) to specify which `IdempotentReceiverInterceptor
objects are applied to this endpoint.
The following example shows how to use the @IdempotentReceiver
annotation:
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
return new IdempotentReceiverInterceptor(new MetadataStoreSelector(m ->
m.getHeaders().get(INVOICE_NBR_HEADER)));
}
@Bean
@ServiceActivator(inputChannel = "input", outputChannel = "output")
@IdempotentReceiver("idempotentReceiverInterceptor")
public MessageHandler myService() {
....
}
When you use the Java DSL, you can add the interceptor to the endpoint’s advice chain, as the following example shows:
@Bean
public IntegrationFlow flow() {
...
.handle("someBean", "someMethod",
e -> e.advice(idempotentReceiverInterceptor()))
...
}
The IdempotentReceiverInterceptor is designed only for the MessageHandler.handleMessage(Message<?>) method.
Starting with version 4.3.1, it implements HandleMessageAdvice , with the AbstractHandleMessageAdvice as a base class, for better dissociation.
See Handling Message Advice for more information.
|
10.10. Logging Channel Adapter
The <logging-channel-adapter>
is often used in conjunction with a wire tap, as discussed in Wire Tap.
However, it can also be used as the ultimate consumer of any flow.
For example, consider a flow that ends with a <service-activator>
that returns a result, but you wish to discard that result.
To do that, you could send the result to NullChannel
.
Alternatively, you can route it to an INFO
level <logging-channel-adapter>
.
That way, you can see the discarded message when logging at INFO
level but not see it when logging at (for example) the WARN
level.
With a NullChannel
, you would see only the discarded message when logging at the DEBUG
level.
The following listing shows all the possible attributes for the logging-channel-adapter
element:
<int:logging-channel-adapter
channel="" (1)
level="INFO" (2)
expression="" (3)
log-full-message="false" (4)
logger-name="" /> (5)
| 1 | The channel connecting the logging adapter to an upstream component. |
| 2 | The logging level at which messages sent to this adapter will be logged.
Default: INFO . |
| 3 | A SpEL expression representing exactly what parts of the message are logged.
Default: payload — only the payload is logged.
if log-full-message is specified, this attribute cannot be specified. |
| 4 | When true , the entire message (including headers) is logged.
Default: false — only the payload is logged.
This attribute cannot be specified if expression is specified. |
| 5 | Specifies the name of the logger (known as category in log4j ).
Used to identify log messages created by this adapter.
This enables setting the log name (in the logging subsystem) for individual adapters.
By default, all adapters log under the following name: org.springframework.integration.handler.LoggingHandler . |
10.10.1. Using Java Configuration
The following Spring Boot application shows an example of configuring the LoggingHandler
by using Java configuration:
@SpringBootApplication
public class LoggingJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(LoggingJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToLogger("foo");
}
@Bean
@ServiceActivator(inputChannel = "logChannel")
public LoggingHandler logging() {
LoggingHandler adapter = new LoggingHandler(LoggingHandler.Level.DEBUG);
adapter.setLoggerName("TEST_LOGGER");
adapter.setLogExpressionString("headers.id + ': ' + payload");
return adapter;
}
@MessagingGateway(defaultRequestChannel = "logChannel")
public interface MyGateway {
void sendToLogger(String data);
}
}
10.10.2. Configuring with the Java DSL
The following Spring Boot application shows an example of configuring the logging channel adapter by using the Java DSL:
@SpringBootApplication
public class LoggingJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(LoggingJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToLogger("foo");
}
@Bean
public IntegrationFlow loggingFlow() {
return IntegrationFlows.from(MyGateway.class)
.log(LoggingHandler.Level.DEBUG, "TEST_LOGGER",
m -> m.getHeaders().getId() + ": " + m.getPayload());
}
@MessagingGateway
public interface MyGateway {
void sendToLogger(String data);
}
}
10.11. java.util.function
Interfaces Support
Starting with version 5.1, Spring Integration provides direct support for interfaces in the java.util.function
package.
All messaging endpoints, (Service Activator, Transformer, Filter, etc.) can now refer to Function
(or Consumer
) beans.
The Messaging Annotations can be applied directly on these beans similar to regular MessageHandler
definitions.
For example if you have this Function
bean definition:
@Configuration
public class FunctionConfiguration {
@Bean
public Function<String, String> functionAsService() {
return String::toUpperCase;
}
}
You can use it as a simple reference in an XML configuration file:
<service-activator input-channel="processorViaFunctionChannel" ref="functionAsService"/>
When we configure our flow with Messaging Annotations, the code is straightforward:
@Bean
@Transformer(inputChannel = "functionServiceChannel")
public Function<String, String> functionAsService() {
return String::toUpperCase;
}
When the function returns an array, Collection
(essentially, any Iterable
), Stream
or Reactor Flux
, @Splitter
can be used on such a bean to perform iteration over the result content.
The java.util.function.Consumer
interface can be used for an <int:outbound-channel-adapter>
or, together with the @ServiceActivator
annotation, to perform the final step of a flow:
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
public Consumer<Message<?>> messageConsumerAsService() {
// Has to be an anonymous class for proper type inference
return new Consumer<Message<?>>() {
@Override
public void accept(Message<?> e) {
collector().add(e);
}
};
}
Also, pay attention to the comment in the code snippet above: if you would like to deal with the whole message in your Function
/Consumer
you cannot use a lambda definition.
Because of Java type erasure we cannot determine the target type for the apply()/accept()
method call.
The java.util.function.Supplier
interface can simply be used together with the @InboundChannelAdapter
annotation, or as a ref
in an <int:inbound-channel-adapter>
:
@Bean
@InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
public Supplier<String> pojoSupplier() {
return () -> "foo";
}
With the Java DSL we just need to use a reference to the function bean in the endpoint definitions.
Meanwhile an implementation of the Supplier
interface can be used as regular MessageSource
definition:
@Bean
public Function<String, String> toUpperCaseFunction() {
return String::toUpperCase;
}
@Bean
public Supplier<String> stringSupplier() {
return () -> "foo";
}
@Bean
public IntegrationFlow supplierFlow() {
return IntegrationFlows.from(stringSupplier())
.transform(toUpperCaseFunction())
.channel("suppliedChannel")
.get();
}
This function support is useful when used together with the Spring Cloud Function framework, where we have a function catalog and can refer to its member functions from an integration flow definition.
10.11.1. Kotlin Lambdas
The Framework also has been improved to support Kotlin lambdas for functions so now you can use a combination of the Kotlin language and Spring Integration flow definitions:
@Bean
@Transformer(inputChannel = "functionServiceChannel")
fun kotlinFunction(): (String) -> String {
return { it.toUpperCase() }
}
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
fun kotlinConsumer(): (Message<Any>) -> Unit {
return { print(it) }
}
@Bean
@InboundChannelAdapter(value = "counterChannel",
poller = [Poller(fixedRate = "10", maxMessagesPerPoll = "1")])
fun kotlinSupplier(): () -> String {
return { "baz" }
}
11. Java DSL
The Spring Integration Java configuration and DSL provides a set of convenient builders and a fluent API that lets you configure Spring Integration message flows from Spring @Configuration
classes.
The Java DSL for Spring Integration is essentially a facade for Spring Integration.
The DSL provides a simple way to embed Spring Integration Message Flows into your application by using the fluent Builder
pattern together with existing Java configuration from Spring Framework and Spring Integration.
We also use and support lambdas (available with Java 8) to further simplify Java configuration.
The cafe offers a good example of using the DSL.
The DSL is presented by the IntegrationFlows
factory for the IntegrationFlowBuilder
.
This produces the IntegrationFlow
component, which should be registered as a Spring bean (by using the @Bean
annotation).
The builder pattern is used to express arbitrarily complex structures as a hierarchy of methods that can accept lambdas as arguments.
The IntegrationFlowBuilder
only collects integration components (MessageChannel
instances, AbstractEndpoint
instances, and so on) in the IntegrationFlow
bean for further parsing and registration of concrete beans in the application context by the IntegrationFlowBeanPostProcessor
.
The Java DSL uses Spring Integration classes directly and bypasses any XML generation and parsing. However, the DSL offers more than syntactic sugar on top of XML. One of its most compelling features is the ability to define inline lambdas to implement endpoint logic, eliminating the need for external classes to implement custom logic. In some sense, Spring Integration’s support for the Spring Expression Language (SpEL) and inline scripting address this, but lambdas are easier and much more powerful.
The following example shows how to use Java Configuration for Spring Integration:
@Configuration
@EnableIntegration
public class MyConfiguration {
@Bean
public AtomicInteger integerSource() {
return new AtomicInteger();
}
@Bean
public IntegrationFlow myFlow() {
return IntegrationFlows.from(integerSource::getAndIncrement,
c -> c.poller(Pollers.fixedRate(100)))
.channel("inputChannel")
.filter((Integer p) -> p > 0)
.transform(Object::toString)
.channel(MessageChannels.queue())
.get();
}
}
The result of the preceding configuration example is that it creates, after ApplicationContext
start up, Spring Integration endpoints and message channels.
Java configuration can be used both to replace and augment XML configuration.
You need not replace all of your existing XML configuration to use Java configuration.
11.1. DSL Basics
The org.springframework.integration.dsl
package contains the IntegrationFlowBuilder
API mentioned earlier and a number of IntegrationComponentSpec
implementations, which are also builders and provide the fluent API to configure concrete endpoints.
The IntegrationFlowBuilder
infrastructure provides common enterprise integration patterns (EIP) for message-based applications, such as channels, endpoints, pollers, and channel interceptors.
Endpoints are expressed as verbs in the DSL to improve readability. The following list includes the common DSL method names and the associated EIP endpoint:
-
transform →
Transformer
-
filter →
Filter
-
handle →
ServiceActivator
-
split →
Splitter
-
aggregate →
Aggregator
-
route →
Router
-
bridge →
Bridge
Conceptually, integration processes are constructed by composing these endpoints into one or more message flows.
Note that EIP does not formally define the term 'message flow', but it is useful to think of it as a unit of work that uses well known messaging patterns.
The DSL provides an IntegrationFlow
component to define a composition of channels and endpoints between them, but now IntegrationFlow
plays only the configuration role to populate real beans in the application context and is not used at runtime.
The following example uses the IntegrationFlows
factory to define an IntegrationFlow
bean by using EIP-methods from IntegrationFlowBuilder
:
@Bean
public IntegrationFlow integerFlow() {
return IntegrationFlows.from("input")
.<String, Integer>transform(Integer::parseInt)
.get();
}
The transform
method accepts a lambda as an endpoint argument to operate on the message payload.
The real argument of this method is GenericTransformer<S, T>
.
Consequently, any of the provided transformers (ObjectToJsonTransformer
, FileToStringTransformer
, and other) can be used here.
Under the covers, IntegrationFlowBuilder
recognizes the MessageHandler
and the endpoint for it, with MessageTransformingHandler
and ConsumerEndpointFactoryBean
, respectively.
Consider another example:
@Bean
public IntegrationFlow myFlow() {
return IntegrationFlows.from("input")
.filter("World"::equals)
.transform("Hello "::concat)
.handle(System.out::println)
.get();
}
The preceding example composes a sequence of Filter → Transformer → Service Activator
.
The flow is "'one way'".
That is, it does not provide a reply message but only prints the payload to STDOUT.
The endpoints are automatically wired together by using direct channels.
|
Lambdas And
Message<?> ArgumentsWhen using lambdas in EIP methods, the "input" argument is generally the message payload.
If you wish to access the entire message, use one of the overloaded methods that take a
This will fail at runtime with a Instead, use:
|
|
Bean Definitions override
The Java DSL can register beans for the object defined in-line in the flow definition, as well as can reuse existing, injected beans.
In case of the same bean name defined for in-line object and existing bean definition, a |
11.2. Message Channels
In addition to the IntegrationFlowBuilder
with EIP methods, the Java DSL provides a fluent API to configure MessageChannel
instances.
For this purpose the MessageChannels
builder factory is provided.
The following example shows how to use it:
@Bean
public MessageChannel priorityChannel() {
return MessageChannels.priority(this.mongoDbChannelMessageStore, "priorityGroup")
.interceptor(wireTap())
.get();
}
The same MessageChannels
builder factory can be used in the channel()
EIP method from IntegrationFlowBuilder
to wire endpoints, similar to wiring an input-channel
/output-channel
pair in the XML configuration.
By default, endpoints are wired with DirectChannel
instances where the bean name is based on the following pattern: [IntegrationFlow.beanName].channel#[channelNameIndex]
.
This rule is also applied for unnamed channels produced by inline MessageChannels
builder factory usage.
However all MessageChannels
methods have a variant that is aware of the channelId
that you can use to set the bean names for MessageChannel
instances.
The MessageChannel
references and beanName
can be used as bean-method invocations.
The following example shows the possible ways to use the channel()
EIP method:
@Bean
public MessageChannel queueChannel() {
return MessageChannels.queue().get();
}
@Bean
public MessageChannel publishSubscribe() {
return MessageChannels.publishSubscribe().get();
}
@Bean
public IntegrationFlow channelFlow() {
return IntegrationFlows.from("input")
.fixedSubscriberChannel()
.channel("queueChannel")
.channel(publishSubscribe())
.channel(MessageChannels.executor("executorChannel", this.taskExecutor))
.channel("output")
.get();
}
-
from("input")
means "'find and use theMessageChannel
with the "input" id, or create one'". -
fixedSubscriberChannel()
produces an instance ofFixedSubscriberChannel
and registers it with a name ofchannelFlow.channel#0
. -
channel("queueChannel")
works the same way but uses an existingqueueChannel
bean. -
channel(publishSubscribe())
is the bean-method reference. -
channel(MessageChannels.executor("executorChannel", this.taskExecutor))
is theIntegrationFlowBuilder
that exposesIntegrationComponentSpec
to theExecutorChannel
and registers it asexecutorChannel
. -
channel("output")
registers theDirectChannel
bean withoutput
as its name, as long as no beans with this name already exist.
Note: The preceding IntegrationFlow
definition is valid, and all of its channels are applied to endpoints with BridgeHandler
instances.
Be careful to use the same inline channel definition through MessageChannels factory from different IntegrationFlow instances.
Even if the DSL parser registers non-existent objects as beans, it cannot determine the same object (MessageChannel ) from different IntegrationFlow containers.
The following example is wrong:
|
@Bean
public IntegrationFlow startFlow() {
return IntegrationFlows.from("input")
.transform(...)
.channel(MessageChannels.queue("queueChannel"))
.get();
}
@Bean
public IntegrationFlow endFlow() {
return IntegrationFlows.from(MessageChannels.queue("queueChannel"))
.handle(...)
.get();
}
The result of that bad example is the following exception:
Caused by: java.lang.IllegalStateException:
Could not register object [queueChannel] under bean name 'queueChannel':
there is already object [queueChannel] bound
at o.s.b.f.s.DefaultSingletonBeanRegistry.registerSingleton(DefaultSingletonBeanRegistry.java:129)
To make it work, you need to declare @Bean
for that channel and use its bean method from different IntegrationFlow
instances.
11.3. Pollers
Spring Integration also provides a fluent API that lets you configure PollerMetadata
for AbstractPollingEndpoint
implementations.
You can use the Pollers
builder factory to configure common bean definitions or those created from IntegrationFlowBuilder
EIP methods, as the following example shows:
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerSpec poller() {
return Pollers.fixedRate(500)
.errorChannel("myErrors");
}
See Pollers
and PollerSpec
in the Javadoc for more information.
If you use the DSL to construct a PollerSpec as a @Bean , do not call the get() method in the bean definition.
The PollerSpec is a FactoryBean that generates the PollerMetadata object from the specification and initializes all of its properties.
|
11.4. DSL and Endpoint Configuration
All IntegrationFlowBuilder
EIP methods have a variant that applies the lambda parameter to provide options for AbstractEndpoint
instances: SmartLifecycle
, PollerMetadata
, request-handler-advice-chain
, and others.
Each of them has generic arguments, so it lets you configure an endpoint and even its MessageHandler
in the context, as the following example shows:
@Bean
public IntegrationFlow flow2() {
return IntegrationFlows.from(this.inputChannel)
.transform(new PayloadSerializingTransformer(),
c -> c.autoStartup(false).id("payloadSerializingTransformer"))
.transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice()))
.get();
}
In addition, the EndpointSpec
provides an id()
method to let you register an endpoint bean with a given bean name, rather than a generated one.
If the MessageHandler
is referenced as a bean, then any existing adviceChain
configuration will be overridden if the .advice()
method is present in the DSL definition:
@Bean
public TcpOutboundGateway tcpOut() {
TcpOutboundGateway gateway = new TcpOutboundGateway();
gateway.setConnectionFactory(cf());
gateway.setAdviceChain(Collections.singletonList(fooAdvice()));
return gateway;
}
@Bean
public IntegrationFlow clientTcpFlow() {
return f -> f
.handle(tcpOut(), e -> e.advice(testAdvice()))
.transform(Transformers.objectToString());
}
That is they are not merged, only the testAdvice()
bean is used in this case.
11.5. Transformers
The DSL API provides a convenient, fluent Transformers
factory to be used as inline target object definition within the .transform()
EIP method.
The following example shows how to use it:
@Bean
public IntegrationFlow transformFlow() {
return IntegrationFlows.from("input")
.transform(Transformers.fromJson(MyPojo.class))
.transform(Transformers.serializer())
.get();
}
It avoids inconvenient coding using setters and makes the flow definition more straightforward.
Note that you can use Transformers
to declare target Transformer
instances as @Bean
instances and, again, use them from IntegrationFlow
definition as bean methods.
Nevertheless, the DSL parser takes care of bean declarations for inline objects, if they are not yet defined as beans.
See Transformers in the Javadoc for more information and supported factory methods.
Also see Lambdas And Message<?>
Arguments.
11.6. Inbound Channel Adapters
Typically, message flows start from an inbound channel adapter (such as <int-jdbc:inbound-channel-adapter>
).
The adapter is configured with <poller>
, and it asks a MessageSource<?>
to periodically produce messages.
Java DSL allows for starting IntegrationFlow
from a MessageSource<?>
, too.
For this purpose, the IntegrationFlows
builder factory provides an overloaded IntegrationFlows.from(MessageSource<?> messageSource)
method.
You can configure the MessageSource<?>
as a bean and provide it as an argument for that method.
The second parameter of IntegrationFlows.from()
is a Consumer<SourcePollingChannelAdapterSpec>
lambda that lets you provide options (such as PollerMetadata
or SmartLifecycle
) for the SourcePollingChannelAdapter
.
The following example shows how to use the fluent API and a lambda to create an IntegrationFlow
:
@Bean
public MessageSource<Object> jdbcMessageSource() {
return new JdbcPollingChannelAdapter(this.dataSource, "SELECT * FROM something");
}
@Bean
public IntegrationFlow pollingFlow() {
return IntegrationFlows.from(jdbcMessageSource(),
c -> c.poller(Pollers.fixedRate(100).maxMessagesPerPoll(1)))
.transform(Transformers.toJson())
.channel("furtherProcessChannel")
.get();
}
For those cases that have no requirements to build Message
objects directly, you can use the IntegrationFlows.from()
variant that is based on the java.util.function.Supplier
.
The result of the Supplier.get()
is automatically wrapped in a Message
(if it is not already a Message
).
11.7. Message Routers
Spring Integration natively provides specialized router types, including:
-
HeaderValueRouter
-
PayloadTypeRouter
-
ExceptionTypeRouter
-
RecipientListRouter
-
XPathRouter
As with many other DSL IntegrationFlowBuilder
EIP methods, the route()
method can apply any AbstractMessageRouter
implementation or, for convenience, a String
as a SpEL expression or a ref
-method
pair.
In addition, you can configure route()
with a lambda and use a lambda for a Consumer<RouterSpec<MethodInvokingRouter>>
.
The fluent API also provides AbstractMappingMessageRouter
options such as channelMapping(String key, String channelName)
pairs, as the following example shows:
@Bean
public IntegrationFlow routeFlow() {
return IntegrationFlows.from("routerInput")
.<Integer, Boolean>route(p -> p % 2 == 0,
m -> m.suffix("Channel")
.channelMapping("true", "even")
.channelMapping("false", "odd")
)
.get();
}
The following example shows a simple expression-based router:
@Bean
public IntegrationFlow routeFlow() {
return IntegrationFlows.from("routerInput")
.route("headers['destChannel']")
.get();
}
The routeToRecipients()
method takes a Consumer<RecipientListRouterSpec>
, as the following example shows:
@Bean
public IntegrationFlow recipientListFlow() {
return IntegrationFlows.from("recipientListInput")
.<String, String>transform(p -> p.replaceFirst("Payload", ""))
.routeToRecipients(r -> r
.recipient("thing1-channel", "'thing1' == payload")
.recipient("thing2-channel", m ->
m.getHeaders().containsKey("recipient")
&& (boolean) m.getHeaders().get("recipient"))
.recipientFlow("'thing1' == payload or 'thing2' == payload or 'thing3' == payload",
f -> f.<String, String>transform(String::toUpperCase)
.channel(c -> c.queue("recipientListSubFlow1Result")))
.recipientFlow((String p) -> p.startsWith("thing3"),
f -> f.transform("Hello "::concat)
.channel(c -> c.queue("recipientListSubFlow2Result")))
.recipientFlow(new FunctionExpression<Message<?>>(m ->
"thing3".equals(m.getPayload())),
f -> f.channel(c -> c.queue("recipientListSubFlow3Result")))
.defaultOutputToParentFlow())
.get();
}
The .defaultOutputToParentFlow()
of the .routeToRecipients()
definition lets you set the router’s defaultOutput
as a gateway to continue a process for the unmatched messages in the main flow.
Also see Lambdas And Message<?>
Arguments.
11.8. Splitters
To create a splitter, use the split()
EIP method.
By default, if the payload is an Iterable
, an Iterator
, an Array
, a Stream
, or a reactive Publisher
, the split()
method outputs each item as an individual message.
It accepts a lambda, a SpEL expression, or any AbstractMessageSplitter
implementation.
Alternatively, you can use it without parameters to provide the DefaultMessageSplitter
.
The following example shows how to use the split()
method by providing a lambda:
@Bean
public IntegrationFlow splitFlow() {
return IntegrationFlows.from("splitInput")
.split(s ->
s.applySequence(false).get().getT2().setDelimiters(","))
.channel(MessageChannels.executor(this.taskExecutor()))
.get();
}
The preceding example creates a splitter that splits a message containing a comma-delimited String
.
Note: The getT2()
method comes from a Tuple
Collection
, which is the result of EndpointSpec.get()
, and represents a pair of ConsumerEndpointFactoryBean
and DefaultMessageSplitter
for the preceding example.
Also see Lambdas And Message<?>
Arguments.
11.9. Aggregators and Resequencers
An Aggregator
is conceptually the opposite of a Splitter
.
It aggregates a sequence of individual messages into a single message and is necessarily more complex.
By default, an aggregator returns a message that contains a collection of payloads from incoming messages.
The same rules are applied for the Resequencer
.
The following example shows a canonical example of the splitter-aggregator pattern:
@Bean
public IntegrationFlow splitAggregateFlow() {
return IntegrationFlows.from("splitAggregateInput")
.split()
.channel(MessageChannels.executor(this.taskExecutor()))
.resequence()
.aggregate()
.get();
}
The split()
method splits the list into individual messages and sends them to the ExecutorChannel
.
The resequence()
method reorders messages by sequence details found in the message headers.
The aggregate()
method collects those messages.
However, you can change the default behavior by specifying a release strategy and correlation strategy, among other things. Consider the following example:
.aggregate(a ->
a.correlationStrategy(m -> m.getHeaders().get("myCorrelationKey"))
.releaseStrategy(g -> g.size() > 10)
.messageStore(messageStore()))
The preceding example correlates messages that have myCorrelationKey
headers and releases the messages once at least ten have been accumulated.
Similar lambda configurations are provided for the resequence()
EIP method.
11.10. Service Activators and the .handle()
method
The .handle()
EIP method’s goal is to invoke any MessageHandler
implementation or any method on some POJO.
Another option is to define an “activity” by using lambda expressions.
Consequently, we introduced a generic GenericHandler<P>
functional interface.
Its handle
method requires two arguments: P payload
and MessageHeaders headers
(starting with version 5.1).
Having that, we can define a flow as follows:
@Bean
public IntegrationFlow myFlow() {
return IntegrationFlows.from("flow3Input")
.<Integer>handle((p, h) -> p * 2)
.get();
}
The preceding example doubles any integer it receives.
However, one main goal of Spring Integration is loose coupling
, through runtime type conversion from message payload to the target arguments of the message handler.
Since Java does not support generic type resolution for lambda classes, we introduced a workaround with an additional payloadType
argument for the most EIP methods and LambdaMessageProcessor
.
Doing so delegates the hard conversion work to Spring’s ConversionService
, which uses the provided type
and the requested message to target method arguments.
The following example shows what the resulting IntegrationFlow
might look like:
@Bean
public IntegrationFlow integerFlow() {
return IntegrationFlows.from("input")
.<byte[], String>transform(p - > new String(p, "UTF-8"))
.handle(Integer.class, (p, h) -> p * 2)
.get();
}
We also can register some BytesToIntegerConverter
within ConversionService
to get rid of that additional .transform()
:
@Bean
@IntegrationConverter
public BytesToIntegerConverter bytesToIntegerConverter() {
return new BytesToIntegerConverter();
}
@Bean
public IntegrationFlow integerFlow() {
return IntegrationFlows.from("input")
.handle(Integer.class, (p, h) -> p * 2)
.get();
}
Also see Lambdas And Message<?>
Arguments.
11.11. Operator log()
For convenience, to log the message journey through the Spring Integration flow (<logging-channel-adapter>
), a log()
operator is presented.
Internally, it is represented by the WireTap
ChannelInterceptor
with a LoggingHandler
as its subscriber.
It is responsible for logging the incoming message into the next endpoint or the current channel.
The following example shows how to use LoggingHandler
:
.filter(...)
.log(LoggingHandler.Level.ERROR, "test.category", m -> m.getHeaders().getId())
.route(...)
In the preceding example, an id
header is logged at the ERROR
level onto test.category
only for messages that passed the filter and before routing.
When this operator is used at the end of a flow, it is a one-way handler and the flow ends.
To make it as a reply-producing flow, you can either use a simple bridge()
after the log()
or, starting with version 5.1, you can use a logAndReply()
operator instead.
logAndReply
can only be used at the end of a flow.
11.12. MessageChannelSpec.wireTap()
Spring Integration includes a .wireTap()
fluent API MessageChannelSpec
builders.
The following example shows how to use the wireTap
method to log input:
@Bean
public QueueChannelSpec myChannel() {
return MessageChannels.queue()
.wireTap("loggingFlow.input");
}
@Bean
public IntegrationFlow loggingFlow() {
return f -> f.log();
}
|
If the
|
When the current MessageChannel
does not implement ChannelInterceptorAware
, an implicit DirectChannel
and BridgeHandler
are injected into the IntegrationFlow
, and the WireTap
is added to this new DirectChannel
.
The following example does not have any channel declaration:
.handle(...)
.log()
}
In the preceding example (and any time no channel has been declared), an implicit DirectChannel
is injected in the current position of the IntegrationFlow
and used as an output channel for the currently configured ServiceActivatingHandler
(from the .handle()
, described earlier).
11.13. Working With Message Flows
IntegrationFlowBuilder
provides a top-level API to produce integration components wired to message flows.
When your integration may be accomplished with a single flow (which is often the case), this is convenient.
Alternately IntegrationFlow
instances can be joined via MessageChannel
instances.
By default, MessageFlow
behaves as a “chain” in Spring Integration parlance.
That is, the endpoints are automatically and implicitly wired by DirectChannel
instances.
The message flow is not actually constructed as a chain, which offers much more flexibility.
For example, you may send a message to any component within the flow, if you know its inputChannel
name (that is, if you explicitly define it).
You may also reference externally defined channels within a flow to allow the use of channel adapters (to enable remote transport protocols, file I/O, and so on), instead of direct channels.
As such, the DSL does not support the Spring Integration chain
element, because it does not add much value in this case.
Since the Spring Integration Java DSL produces the same bean definition model as any other configuration options and is based on the existing Spring Framework @Configuration
infrastructure, it can be used together with XML definitions and wired with Spring Integration messaging annotation configuration.
You can also define direct IntegrationFlow
instances by using a lambda.
The following example shows how to do so:
@Bean
public IntegrationFlow lambdaFlow() {
return f -> f.filter("World"::equals)
.transform("Hello "::concat)
.handle(System.out::println);
}
The result of this definition is the same set of integration components that are wired with an implicit direct channel.
The only limitation here is that this flow is started with a named direct channel - lambdaFlow.input
.
Also, a Lambda flow cannot start from MessageSource
or MessageProducer
.
Starting with version 5.1, this kind of IntegrationFlow
is wrapped to the proxy to expose lifecycle control and provide access to the inputChannel
of the internally associated StandardIntegrationFlow
.
Starting with version 5.0.6, the generated bean names for the components in an IntegrationFlow
include the flow bean followed by a dot (.
) as a prefix.
For example, the ConsumerEndpointFactoryBean
for the .transform("Hello "::concat)
in the preceding sample results in a bean name of lambdaFlow.o.s.i.config.ConsumerEndpointFactoryBean#0
.
(The o.s.i
is a shortened from org.springframework.integration
to fit on the page.)
The Transformer
implementation bean for that endpoint has a bean name of lambdaFlow.transformer#0
(starting with version 5.1), where instead of a fully qualified name of the MethodInvokingTransformer
class, its component type is used.
The same pattern is applied for all the NamedComponent
s when the bean name has to be generated within the flow.
These generated bean names are prepended with the flow ID for purposes such as parsing logs or grouping components together in some analysis tool, as well as to avoid a race condition when we concurrently register integration flows at runtime.
See Dynamic and Runtime Integration Flows for more information.
11.14. FunctionExpression
We introduced the FunctionExpression
class (an implementation of SpEL’s Expression
interface) to let us use lambdas and generics
.
The Function<T, R>
option is provided for the DSL components, along with an expression
option, when there is the implicit Strategy
variant from Core Spring Integration.
The following example shows how to use a function expression:
.enrich(e -> e.requestChannel("enrichChannel")
.requestPayload(Message::getPayload)
.propertyFunction("date", m -> new Date()))
The FunctionExpression
also supports runtime type conversion, as is done in SpelExpression
.
11.15. Sub-flows support
Some of if…else
and publish-subscribe
components provide the ability to specify their logic or mapping by using sub-flows.
The simplest sample is .publishSubscribeChannel()
, as the following example shows:
@Bean
public IntegrationFlow subscribersFlow() {
return flow -> flow
.publishSubscribeChannel(Executors.newCachedThreadPool(), s -> s
.subscribe(f -> f
.<Integer>handle((p, h) -> p / 2)
.channel(c -> c.queue("subscriber1Results")))
.subscribe(f -> f
.<Integer>handle((p, h) -> p * 2)
.channel(c -> c.queue("subscriber2Results"))))
.<Integer>handle((p, h) -> p * 3)
.channel(c -> c.queue("subscriber3Results"));
}
You can achieve the same result with separate IntegrationFlow
@Bean
definitions, but we hope you find the sub-flow style of logic composition useful.
We find that it results in shorter (and so more readable) code.
A similar publish-subscribe
sub-flow composition provides the .routeToRecipients()
method.
Another example is using .discardFlow()
instead of .discardChannel()
on the .filter()
method.
The .route()
deserves special attention.
Consider the following example:
@Bean
public IntegrationFlow routeFlow() {
return f -> f
.<Integer, Boolean>route(p -> p % 2 == 0,
m -> m.channelMapping("true", "evenChannel")
.subFlowMapping("false", sf ->
sf.<Integer>handle((p, h) -> p * 3)))
.transform(Object::toString)
.channel(c -> c.queue("oddChannel"));
}
The .channelMapping()
continues to work as it does in regular Router
mapping, but the .subFlowMapping()
tied that sub-flow to the main flow.
In other words, any router’s sub-flow returns to the main flow after .route()
.
|
Sometimes, you need to refer to an existing
Caused by: org.springframework.beans.factory.BeanCreationException: The 'currentComponent' (org.springframework.integration.router.MethodInvokingRouter@7965a51c) is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow. When you configure a sub-flow as a lambda, the framework handles the request-reply interaction with the sub-flow and a gateway is not needed. |
Sub-flows can be nested to any depth, but we do not recommend doing so. In fact, even in the router case, adding complex sub-flows within a flow would quickly begin to look like a plate of spaghetti and be difficult for a human to parse.
11.16. Using Protocol Adapters
All of the examples shown so far illustrate how the DSL supports a messaging architecture by using the Spring Integration programming model. However, we have yet to do any real integration. Doing so requires access to remote resources over HTTP, JMS, AMQP, TCP, JDBC, FTP, SMTP, and so on or access to the local file system. Spring Integration supports all of these and more. Ideally, the DSL should offer first class support for all of them, but it is a daunting task to implement all of these and keep up as new adapters are added to Spring Integration. So the expectation is that the DSL is continually catching up with Spring Integration.
Consequently, we provide the high-level API to seamlessly define protocol-specific messaging.
We do so with the factory and builder patterns and with lambdas.
You can think of the factory classes as “Namespace Factories”, because they play the same role as the XML namespace for components from the concrete protocol-specific Spring Integration modules.
Currently, Spring Integration Java DSL supports the Amqp
, Feed
, Jms
, Files
, (S)Ftp
, Http
, JPA
, MongoDb
, TCP/UDP
, Mail
, WebFlux
, and Scripts
namespace factories.
The following example shows how to use three of them (Amqp
, Jms
, and Mail
):
@Bean
public IntegrationFlow amqpFlow() {
return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue()))
.transform("hello "::concat)