In addition to providing the general mechanism to apply AOP advice classes, Spring Integration provides these out-of-the-box advice implementations:

Starting with version 5.3, a ReactiveRequestHandlerAdvice can be used for request message handlers producing a Mono replies. A BiFunction<Message<?>, Mono<?>, Publisher<?>> has to be provided for this advice and it is called from the Mono.transform() operator on a reply produced by the intercepted handleRequestMessage() method implementation. Typically, such a Mono customization is necessary when we would like to control network fluctuations via timeout(), retry() and similar support operators. For example when we can an HTTP request over WebFlux client, we could use below configuration to not wait for response more than 5 seconds:

The message argument is the request message for the message handler and can be used to determine request-scope attributes. The mono argument is the result of this message handler’s handleRequestMessage() method implementation. A nested Mono.transform() can also be called from this function to apply, for example, a Reactive Circuit Breaker.

Starting with version 6.1, the ContextHolderRequestHandlerAdvice has been introduced. This advice takes some value from the request message as and stores it in the context holder. The value is clear from the context when an execution is finished on the target MessageHandler. The best way to think about this advice is similar to the programming flow where we store some value into a ThreadLocal, get access to it from the target call and then clean up the ThreadLocal after execution. The ContextHolderRequestHandlerAdvice requires these constructor arguments: a Function<Message<?>, Object> as a value provider, Consumer<Object> as a context set callback and Runnable as a context clean up hook.

Following is a sample how a ContextHolderRequestHandlerAdvice can be used in combination with a o.s.i.file.remote.session.DelegatingSessionFactory:

And it is just enough to send a message to the in channel with a FACTORY_KEY header set to either one or two. The ContextHolderRequestHandlerAdvice sets the value from that header into a DelegatingSessionFactory via its setThreadKey. Then when FtpOutboundGateway executes an ls command a proper delegating SessionFactory is chosen from the DelegatingSessionFactory according to the value in its ThreadLocal. When the result is produced from the FtpOutboundGateway, a ThreadLocal value in the DelegatingSessionFactory is cleared according to the clearThreadKey() call from the ContextHolderRequestHandlerAdvice. See Delegating Session Factory for more information.

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:

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:

While the abstract class mentioned above is a convenience, you can add any Advice, including a transaction advice, to the chain.

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.

Consider the following configuration:

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.

Starting with version 5.3, the HandleMessageAdviceAdapter is provided to apply any MethodInterceptor for the MessageHandler.handleMessage() method and, therefore, the whole sub-flow. For example, a RetryOperationsInterceptor could be applied to the whole sub-flow starting from some endpoint; this is not possible, by default, because the consumer endpoint applies advices only to the AbstractReplyProducingMessageHandler.RequestHandler.handleRequestMessage().

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 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:

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:

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:

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.

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:

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 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.

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:

When you implement MethodInterceptor directly, you could cast the target object as follows:

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:

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.

Normally, the selector selects a message for acceptance if there is no existing value for the key. In some cases, it is useful to compare the current and new values for a key, to determine whether the message should be accepted. Starting with version 5.3, the compareValues property is provided which references a BiPredicate<String, String>; the first parameter is the old value; return true to accept the message and replace the old value with the new value in the MetadataStore. This can be useful to reduce the number of keys; for example, when processing lines in a file, you can store the file name in the key and the current line number in the value. Then, after a restart, you can skip lines that have already been processed. See Idempotent Downstream Processing a Split File for an example.

For convenience, the MetadataStoreSelector options are configurable directly on the <idempotent-receiver> component. The following listing shows all the possible attributes:

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:

When you use the Java DSL, you can add the interceptor to the endpoint’s advice chain, as the following example shows: