CompletableFuture<Invoice> order(Order order);

<int:gateway service-interface="foo.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.

MyCompletableFuture<Invoice> order(Order order);

<int:gateway service-interface="foo.Service" default-request-channel="orders" />

In this scenario, the caller thread will return 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 is emitted indicating that the async executor cannot be used for this scenario.

CompletableFuture s can be used to perform additional manipulation on the reply, such as:

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 the Project Reactor with gateway interface methods, utilizing 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 invoking your Consumer when the result is returned to the gateway.

	Important
	The Mono isn’t flushed immediately by the framework. Hence the underlying message flow won’t be started before the gateway method returns (as it is with Future<?> Executor task). The flow will be started when the Mono is subscribed. Alternatively, the Mono (being a Composable) might be a part of Reactor stream, when the subscribe() is related to the entire Flux. For example:

@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 is a simple callback scenario:

Mono<Invoice> mono = service.process(myOrder);

mono.subscribe(invoice -> handleInvoice(invoice));

The calling thread continues, with handleInvoice() being called when the flow completes.

==== Gateway behavior when no response arrives

As it was explained earlier, the Gateway provides a convenient way of interacting with a Messaging system via POJO method invocations, but realizing that 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 (e.g., a reply message might not arrive - which is equivalent to a method not returning). It is important to go over several scenarios especially in the Sync Gateway case and understand the default behavior of the Gateway and how to deal with these scenarios to make the Sync Gateway behavior more predictable regardless of the outcome of the message flow that was initialed from such Gateway.

There are certain attributes that could be configured to make Sync Gateway behavior more predictable, but some of them might not always work as you might have expected. One of them is reply-timeout (at the method level or default-reply-timeout at the gateway level). So, lets look at the reply-timeout attribute and see how it can/can’t influence the behavior of the Sync Gateway in various scenarios. We will look at single-threaded scenario (all components downstream are connected via Direct Channel) and multi-threaded scenarios (e.g., somewhere downstream you may have Pollable or Executor Channel which breaks single-thread boundary)

Long running process downstream

Sync Gateway - single-threaded. If a component downstream is still running (e.g., infinite loop or a very slow service), then setting a reply-timeout has no effect and the Gateway method call will not return until such downstream service exits (via return or exception). Sync Gateway - multi-threaded. If a component downstream is still running (e.g., infinite loop or a very slow service), in a multi-threaded message flow setting the reply-timeout will have an effect by allowing gateway method invocation to return once the timeout has been reached, since the GatewayProxyFactoryBean  will simply poll on the reply channel waiting for a message until the timeout expires. However it could result in a null return from the Gateway method if the timeout has been reached before the actual reply was produced. It is also important to understand that the reply message (if produced) will be 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 this 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 will hang indefinitely unless: a) a reply-timeout has been configured or b) the requires-reply attribute has been set on the downstream component (e.g., service-activator) that might return null. In this case, an Exception would be thrown and propagated to the Gateway.Sync Gateway - multi-threaded. Behavior is the same as above.

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 will hang indefinitely unless a reply-timeout has been configured  Sync Gateway - multi-threaded Behavior is the same as above.

Downstream component results in Runtime Exception (regardless of the method signature)

Sync Gateway - single-threaded. If a component downstream throws a Runtime Exception, such exception will be propagated via an Error Message back to the gateway and re-thrown. Sync Gateway - multi-threaded Behavior is the same as above.

Important

It is also important to understand that by default reply-timeout is unbounded* which means that if not explicitly set there are several scenarios (described above) where your Gateway method invocation might hang indefinitely. So, make sure you analyze your flow and if there is even a remote possibility of one of these scenarios to occur, set the reply-timeout attribute to a safe value or, even better, 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 does return null internally. But also, realize that there are some scenarios (see the very first one) where reply-timeout will not help. That means it is also important to analyze your message flow and decide when to use a Sync Gateway vs an Async Gateway. As you’ve seen the latter case is simply a matter of defining Gateway methods that return Future instances. Then, you are guaranteed to receive that return value, and you will 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 will result 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 will behave like that containing a service-activator with the requires-reply attribute. In other words, it will help to ensure a timely response from the Gateway method invocation.

Note

* reply-timeout is unbounded for <gateway/> elements (created by the GatewayProxyFactoryBean). Inbound gateways for external integration (ws, http, etc.) share many characteristics and attributes with these gateways. However, for those inbound gateways, the default reply-timeout is 1000 milliseconds (1 second). If a downstream async 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.

	Important
	It is important to understand that the timer starts when the thread returns to the gateway, i.e. 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 will be immediately available; for asynchronous flows, the thread will wait for up to this time.

Also see the section called “CompletableFuture” in the Java DSL chapter for options to define gateways via IntegrationFlows.

=== Service Activator

==== Introduction

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 and, as with most of the configuration options you’ll see here, the same behavior actually applies for most of the other components we have seen.

==== Configuring Service Activator

To create a Service Activator, use the service-activator element with the input-channel and ref attributes:

<int:service-activator input-channel="exampleChannel" ref="exampleHandler"/>

The configuration above selects all methods from the exampleHandler which meet one of the Messaging requirements:

The target method for invocation at runtime is selected for each request message by their payload type. Or as a fallback to 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 Section 8.1.7, “Content Type Conversion” with the target method being invoked after conversion.

To delegate to an explicitly defined method of any object, simply add the "method" attribute.

<int:service-activator input-channel="exampleChannel" ref="somePojo" method="someMethod"/>

In either case, when the service method returns a non-null value, the endpoint will attempt to send the reply message to an appropriate reply channel. To determine the reply channel, it will first check if an "output-channel" was provided in the endpoint configuration:

<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 will then check the request Message’s replyChannel header value. If that value is available, it will then check its type. If it is a MessageChannel, the reply message will be sent to that channel. If it is a String, then the endpoint will attempt to resolve the channel name to a channel instance. If the channel cannot be resolved, then a DestinationResolutionException will be thrown. It it can be resolved, the Message will be sent there. If the request Message doesn’t have replyChannel header and 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 Section 4.1.6, “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, 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 will be assumed that it is a Message payload, which will be extracted from the message and injected into such service method. This is generally the recommended 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 the section called “CompletableFuture”

	Note
	The service method is not required to have any arguments at all, which means you can implement event-style Service Activators, where all you care about is an invocation of the service method, not worrying about the contents of the message. Think of it as a NULL JMS message. An example use-case for such an implementation could be a simple counter/monitor of messages deposited on the input channel.

Starting with version 4.1 the framework correct converts Message properties (payload and headers) to the Java 8 Optional POJO method parameters:

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

}

Using a ref attribute is generally recommended 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:

<int:service-activator id="exampleServiceActivator" input-channel="inChannel"
            output-channel = "outChannel" method="foo">
    <beans:bean class="org.foo.ExampleServiceActivator"/>
</int:service-activator>

	Note
	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 will result in an Exception being thrown.

Important

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 will get a configuration exception.

Service Activators and the Spring Expression Language (SpEL)

Since Spring Integration 2.0, Service Activators can also benefit from SpEL (http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/expressions.html).

For example, you may now invoke any bean method without pointing to the bean via a ref attribute or including it as an inner bean definition. For example:

<int:service-activator input-channel="in" output-channel="out"
	expression="@accountService.processAccount(payload, headers.accountId)"/>

	<bean id="accountService" class="foo.bar.Account"/>

In the above configuration instead of injecting accountService using a ref or as an inner bean, we are simply using SpEL’s @beanId notation and invoking a method which takes a type compatible with Message payload. We are also passing a header value. As you can see, any valid SpEL expression can be evaluated against any content in the Message. For simple scenarios your Service Activators do not even have to reference a bean if all logic can be encapsulated by such an expression.

<int:service-activator input-channel="in" output-channel="out" expression="payload * 2"/>

In the above configuration our service logic is to simply multiply the payload value by 2, and SpEL lets us handle it relatively easy.

See the section called “CompletableFuture” in Java DSL chapter for more information about configuring Service Activator.

==== Asynchronous Service Activator

The service activator is invoked by the calling thread; this would be some 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 (setAsync(true) when using Java configuration). If the service returns a ListenableFuture<?> when this is 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 using a PollableChannel because the poller thread is freed up to perform other services within the framework.

If the service completes the future with an Exception, normal error processing will occur - an ErrorMessage is sent to the errorChannel message header, if present or otherwise to the default errorChannel (if available).

=== Delayer

==== Introduction

A Delayer is a simple endpoint that allows a Message flow to be delayed by a certain interval. When a Message is delayed, the original sender will not block. Instead, the delayed Messages will be 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 will be used for the actual execution of releasing the Messages. Below you will find several examples of configuring a Delayer.

==== 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 expression sub-element) that are used to determine the number of milliseconds that each Message should be delayed. The following delays all messages by 3 seconds:

<int:delayer id="delayer" input-channel="input"
             default-delay="3000" output-channel="output"/>

If you need per-Message determination of the delay, then you can also provide the SpEL expression using the expression attribute:

<int:delayer id="delayer" input-channel="input" output-channel="output"
             default-delay="3000" expression="headers['delay']"/>

In the example above, the 3 second delay would only apply when the expression evaluates to null for a given inbound Message. If you only want to apply a delay to Messages that have a valid result of the expression evaluation, then you can use a default-delay of 0 (the default). For any Message that has a delay of 0 (or less), the Message will be sent immediately, on the calling Thread.

The java configuration equivalent of the second example is:

@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;
}

and with the Java DSL:

@Bean
public IntegrationFlow flow() {
    return IntegrationFlows.from("input")
            .delay("delayer.messageGroupId", d -> d
                    .defaultDelay(3_000L)
                    .delayExpression("headers['delay']"))
            .channel("output")
            .get();
}

	Note
	The XML parser uses a message group id <beanName>.messageGroupId.

Tip

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 will be counted from the current time (e.g. a value of 5000 would delay the Message for at least 5 seconds from the time it is received by the Delayer). With a Date instance, the Message will not be released until the time represented by that Date object. In either case, a value that equates to a non-positive delay, or a Date in the past, will not result in any delay. Instead, it will be 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) will be applied.

Important

The expression evaluation may throw an evaluation Exception for various reasons, including an invalid expression, or other conditions. By default, such exceptions are ignored (logged at 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 above. 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.

Tip

Notice in the example above that the delay expression is specified as headers['delay']. This is the SpEL Indexer syntax to access a Map element (MessageHeaders implements Map), it invokes: headers.get("delay"). For simple map element names (that do not contain .) you can also use the SpEL dot accessor syntax, where the above header expression can be specified as headers.delay. But, different results are achieved if the header is missing. In the first case, the expression will evaluate to null; the second will result in something like: org.springframework.expression.spel.SpelEvaluationException: EL1008E:(pos 8): Field or property 'delay' cannot be found on object of type 'org.springframework.messaging.MessageHeaders' So, 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: the section called “CompletableFuture”. If you want to delegate to a different scheduler, you can provide a reference through the delayer element’s scheduler attribute:

<int:delayer id="delayer" input-channel="input" output-channel="output"
    expression="headers.delay"
    scheduler="exampleTaskScheduler"/>

<task:scheduler id="exampleTaskScheduler" pool-size="3"/>

Tip

If you configure an external ThreadPoolTaskScheduler you can set on this scheduler property waitForTasksToCompleteOnShutdown = true. It allows successful completion of delay tasks, which 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 delayer requires an external scheduler instance and waitForTasksToCompleteOnShutdown was deleted; you should use the scheduler’s own configuration.

Tip

Also keep in mind ThreadPoolTaskScheduler has a property errorHandler which can be injected with some implementation of org.springframework.util.ErrorHandler. This handler allows to process 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 will 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 the section called “CompletableFuture”.

==== Delayer and a Message Store

The DelayHandler persists delayed Messages into the Message Group in the provided MessageStore. (The groupId is based on required id attribute of <delayer> element.) A delayed message is removed from the MessageStore by the scheduled task just before the DelayHandler sends the Message to the output-channel. If the provided MessageStore is persistent (e.g. 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 is used when rescheduling. If a delayed Message remained in the MessageStore more than its delay, it will be sent immediately after startup.

The <delayer> can be enriched with mutually exclusive sub-elements <transactional> or <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 ReleaseMessageHandler's transaction will be rolled back. In this case the delayed Message will remain in the persistent MessageStore. You can use any custom org.aopalliance.aop.Advice implementation within the <advice-chain>. A sample configuration of the <delayer> may look like this:

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

	Note
	For more information regarding the message store, JMX, and the control bus, see the section called “CompletableFuture”.

==== Release Failures

Starting with version 5.0.8, there are two new properties on the delayer:

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.

=== Scripting support

With Spring Integration 2.1 we’ve added support for the JSR223 Scripting for Java specification, introduced in Java version 6. This allows you to 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 please refer to the documentation

	Important
	Note that this feature requires Java 6 or higher. Sun developed a JSR223 reference implementation which works with Java 5 but it is not officially supported and we have not tested it with Spring Integration.

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 distribution. Other language implementations may be available or under development. Please refer to the appropriate project website for more information.

	Important
	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/or the implementer’s interpretation of the specification.

	Tip
	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 you will find this section relevant as well.

==== 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 containing the script. To enable scripting support Spring Integration defines a ScriptExecutingMessageProcessor which will bind the Message Payload to a variable named payload and the Message Headers to a headers variable, both accessible within the script execution context. All that is left for you to do is write a script that uses these variables. Below are a couple of sample configurations:

Filter

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

Here, you see that the script can be included inline or can reference a resource location via the location attribute. Additionally the lang attribute corresponds to the language name (or JSR223 alias)

Other Spring Integration endpoint elements which 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 accomplish this, specify the refresh-check-delay attribute on the script element:

<int-script:script location="..." refresh-check-delay="5000"/>

In the above example, the script location will be checked for updates every 5 seconds. If the script is updated, any invocation that occurs later than 5 seconds since the update will result in execution of the new script.

<int-script:script location="..." refresh-check-delay="0"/>

In the above example the context will be updated with any script modifications as soon as such modification occurs, providing a simple mechanism for real-time configuration. Any negative number value means the script will not be reloaded after initialization of the application context. This is the default behavior.

	Important
	Inline scripts can not be reloaded.

<int-script:script location="..." refresh-check-delay="-1"/>

Script variable bindings

Variable bindings are required to enable the script to reference variables externally provided to the script’s execution context. As we have seen, payload and headers are used as binding variables by default. You can bind additional variables to a script via <variable> sub-elements:

<script:script lang="js" location="foo/bar/MyScript.js">
    <script:variable name="foo" value="foo"/>
    <script:variable name="bar" value="bar"/>
    <script:variable name="date" ref="date"/>
</script:script>

As shown in the above example, you can bind a script variable either to a scalar value or a Spring bean reference. Note that payload and headers will still be included as binding variables.

With Spring Integration 3.0, in addition to the variable sub-element, the variables attribute has been introduced. This attribute and variable sub-elements aren’t 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:

<service-activator input-channel="input">
    <script:script lang="ruby" variables="foo=FOO, date-ref=dateBean">
        <script:variable name="bar" ref="barBean"/>
        <script:variable name="baz" value="bar"/>
        <![CDATA[
            payload.foo = foo
            payload.date = date
            payload.bar = bar
            payload.baz = baz
            payload
        ]]>
    </script:script>
</service-activator>

The example above shows a combination of an inline script, a variable sub-element and a variables attribute. The variables attribute is 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 above. That means that the binding variable will have the name date, but the value will be 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 using the ScriptVariableGenerator strategy:

public interface ScriptVariableGenerator {

    Map<String, Object> generateScriptVariables(Message<?> message);

}

This interface requires you to implement the method generateScriptVariables(Message). The Message argument allows you to access any data available in the Message payload and headers and the return value is the Map of bound variables. This method will be called every time the script is executed for a Message. All you need to do is 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> s with payload and headers variables from the Message in its generateScriptVariables(Message) method.

	Important
	You cannot provide both the script-variable-generator attribute and <variable> sub-element(s) as they are mutually exclusive.

=== Groovy support

In Spring Integration 2.0 we added Groovy support allowing you to 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 please refer to the Groovy documentation which you can find on the project website.

==== Groovy configuration

With Spring Integration 2.1, Groovy Support’s configuration namespace 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, 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. Below are a couple of sample configurations:

Filter

<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 above examples show, the configuration looks identical to the general Scripting Support configuration. The only difference is the use of the Groovy namespace as indicated in the examples by the int-groovy namespace prefix. Also note that the lang attribute on the <script> tag is not valid in this namespace.

Groovy object customization

If you need to customize the Groovy object itself, beyond setting variables, you can reference a bean that implements GroovyObjectCustomizer via 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:

<int:service-activator input-channel="groovyChannel">
    <int-groovy:script location="foo/SomeScript.groovy" customizer="groovyCustomizer"/>
</int:service-activator>

<beans:bean id="groovyCustomizer" class="org.foo.MyGroovyObjectCustomizer"/>

Setting a custom GroovyObjectCustomizer is not mutually exclusive with <variable> sub-elements or the script-variable-generator attribute. It can also be provided when defining an inline script.

With Spring Integration 3.0, in addition to the variable sub-element, the variables attribute has been introduced. 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:

<int-groovy:script>
    <![CDATA[
        entityManager.persist(payload)
        payload
    ]]>
</int-groovy:script>

where variable entityManager is an appropriate bean in the application context.

For more information regarding <variable>, variables, and script-variable-generator, see the paragraph Script variable bindings of the section called “CompletableFuture”.

Groovy Script Compiler Customization

The @CompileStatic hint is the most popular Groovy compiler customization option, which can be used on the class or method level. See more information in the Groovy Reference Manual and, specifically, @CompileStatic. To utilize this feature for short scripts (in integration scenarios), we are forced to change a simple script like this (a <filter> script):

headers.type == 'good'

to more Java-like code:

@groovy.transform.CompileStatic
String filter(Map headers) {
	headers.type == 'good'
}

filter(headers)

With that, the filter() method will be transformed and compiled to static Java code, bypassing the Groovy dynamic phases of invocation, like getProperty() factories and CallSite proxies.

Starting with version 4.3, Spring Integration Groovy components can be configured with the compile-static boolean option, specifying that ASTTransformationCustomizer for @CompileStatic should be added to the internal CompilerConfiguration. With that in place, we can omit the method declaration with @CompileStatic in our script code and still get compiled plain Java code. In this case our script can still be short but still needs to be a little more verbose than interpreted script:

binding.variables.headers.type == 'good'

Where we can access the headers and payload (or any other) variables only through the groovy.lang.Script binding property since, with @CompileStatic, we don’t have the dynamic GroovyObject.getProperty() capability.

In addition, the compiler-configuration bean reference has been introduced. With this attribute, you can provide any other required Groovy compiler customizations, e.g. ImportCustomizer. For more information about this feature, please, refer to the Groovy Documentation: Advanced compiler configuration.

	Note
	Using compilerConfiguration does not automatically add a ASTTransformationCustomizer for @CompileStatic and overrides the compileStatic option. If CompileStatic is still requirement, a new ASTTransformationCustomizer(CompileStatic.class) should be manually added into the CompilationCustomizers of that custom compilerConfiguration.

	Note
	The Groovy compiler customization does not have any effect to the refresh-check-delay option and reloadable scripts can be statically compiled, too.

==== Control Bus

As described in (EIP), the idea behind the Control Bus is that the same messaging system can be used 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 above so that it’s possible to send Messages as a means of invoking exposed operations. One option for those operations is Groovy scripts.

<int-groovy:control-bus input-channel="operationChannel"/>

The Control Bus has an input channel that can be accessed for invoking operations on the beans in the application context.

The Groovy Control Bus executes messages on the input channel as Groovy scripts. It takes a message, compiles the body to a Script, customizes it with a GroovyObjectCustomizer, and then executes it. The Control Bus' MessageProcessor exposes all beans in the application context that are annotated with @ManagedResource, implement Spring’s Lifecycle interface or extend Spring’s CustomizableThreadCreator base class (e.g. several of the TaskExecutor and TaskScheduler implementations).

Important

Be careful about using managed beans with custom scopes (e.g. request) in the Control Bus' command scripts, especially inside an async message flow. If The Control Bus' MessageProcessor can’t expose a bean from the application context, you may end up with some BeansException during command script’s executing. For example, if a custom scope’s context is not established, the attempt to get a bean within that scope will trigger a BeanCreationException.

If you need to further customize the Groovy objects, you can also provide a reference to a bean that implements GroovyObjectCustomizer via the customizer attribute.

<int-groovy:control-bus input-channel="input"
        output-channel="output"
        customizer="groovyCustomizer"/>

<beans:bean id="groovyCustomizer" class="org.foo.MyGroovyObjectCustomizer"/>

=== Adding Behavior to Endpoints

==== Introduction

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, let’s say 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 would cause 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 would be 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. For example:

<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 will only be applied locally to this gateway and will not apply to further actions taken downstream after the reply is sent to the nextChannel. The scope of the advice is limited to the endpoint itself.

Important

At this time, you cannot advise an entire <chain/> of endpoints. The schema does not allow a <request-handler-advice-chain/> as a child element of the chain itself. However, a <request-handler-advice-chain/> can be added to individual reply-producing endpoints within a <chain/> element. An exception is that, in a chain that produces no reply, because the last element in the chain is an outbound-channel-adapter, that last element cannot be advised. If you need to advise such an element, it must be moved outside of the chain (with the output-channel of the chain being the input-channel of the adapter. The adapter can then be advised as normal. For chains that produce a reply, every child element can be advised.

==== Provided Advice Classes

In addition to providing the general mechanism to apply AOP Advice classes in this way, three standard Advices are provided:

These are each described in detail in the following sections.

===== 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, where the thread pauses (if so configured) 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 (e.g. JMS) to be responsible for resubmitting, rather than performing it on the current thread. Stateful retry needs some mechanism to detect a retried submission.

Further Information

For more information on spring-retry, refer to the project’s javadocs, as well as the reference documentation for Spring Batch, where spring-retry originated.

	Warning
	The default back off behavior is no 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 following examples use a simple <service-activator/> that always throws an exception:

public class FailingService {

    public void service(String message) {
        throw new RuntimeException("foo");
    }
}

Simple Stateless Retry

This example uses the default RetryTemplate which has a SimpleRetryPolicy which tries 3 times. There is no BackOffPolicy so the 3 attempts are made back-to-back-to-back with no delay between attempts. There is no RecoveryCallback 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 using an error-channel on the inbound endpoint.

<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

This example adds a RecoveryCallback to the above example; it uses a ErrorMessageSendingRecoverer to send an ErrorMessage 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 the SimpleRetryPolicy but it increases the attempts to 4. It also adds an ExponentialBackoffPolicy where the first retry waits 1 second, the second waits 5 seconds and the third waits 25 (for 4 attempts in all).

<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 above configuration can be greatly simplified with the namespace support for the retry advice:

<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 this example, the advice is defined as a top level bean so it can be used in multiple request-handler-advice-chain s. You can also define the advice directly within the chain:

<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/> with no child element uses no back off; it can have a fixed-back-off or exponential-back-off child element. If there is no recovery-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 the RetryTemplate can determine the current state of retry for this message. The framework provides a SpelExpressionRetryStateGenerator which determines the message identifier using a SpEL expression. This is shown below; this example again uses the default policies (3 attempts with no back off); of course, as with stateless retry, these policies can be customized.

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

Comparing 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 will retry for all exceptions and the exception classifier just looks at the top level exception. If you configure it to, say, only retry on BarException and your application throws a FooException where the cause is a BarException, retry will not occur.

Since Spring Retry 1.0.3, the BinaryExceptionClassifier has a property traverseCauses (default false). When true it will traverse exception causes until it finds a match or there is no cause.

To use this classifier for retry, use a SimpleRetryPolicy created with the constructor that takes the max attempts, the Map of Exception s and the boolean (traverseCauses), and inject this policy into the RetryTemplate.

===== Circuit Breaker Advice

The general idea of the Circuit Breaker Pattern is that, if a service is not currently available, then don’t 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 will attempt 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 will "fail fast" and no attempt will be 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 will immediately go to the open state; if the attempt succeeds, the breaker will go to the closed state, in which case, it won’t 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 will wait before attempting another request. Default is 1000 milliseconds.

Example:

<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 above example, the threshold is set to 2 and halfOpenAfter is set to 12 seconds; a new request arrives every 5 seconds. You can see that the first two attempts invoked the service; the third and fourth failed with an exception indicating 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, either after 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, and an additional property inputMessage which 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 s, this payload has failedMessage and cause properties, as well as an additional property evaluationResult, containing the result of the expression evaluation.

When an exception is thrown in the scope of the advice, by default, that exception is thrown to caller after any failureExpression is evaluated. If you wish to suppress throwing the exception, set the trapException property to true.

Example - Configuring the 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);
    }

}

==== Custom Advice Classes

In addition to the provided Advice classes above, you can implement your own Advice classes. While you can provide any implementation of org.aopalliance.aop.Advice (usually org.aopalliance.intercept.MethodInterceptor), it is generally recommended that you subclass o.s.i.handler.advice.AbstractRequestHandlerAdvice. This has the benefit of avoiding writing 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:

/**
 * 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 simply a convenience to avoid subclasses dealing 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 allows the same advice to be applied to multiple handlers. The RequestHandlerCircuitBreakerAdvice uses this to keep circuit breaker state for each handler.

The message parameter is the message that will be 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 and/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(); but the advice does have the ability to modify the return value. Note that only AbstractReplyProducingMessageHandler s return a value.

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;
    }
}

Note

In addition to the execute() method, the ExecutionCallback provides an additional method cloneAndExecute(). This method must be used in cases where the invocation might be called multiple times within a single execution of doInvoke(), such as in the RequestHandlerRetryAdvice. This is required because the Spring AOP org.springframework.aop.framework.ReflectiveMethodInvocation object maintains state of which advice in a chain was last invoked; this state must be reset for each call. For more information, see the ReflectiveMethodInvocation JavaDocs.

==== Other Advice Chain Elements

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

==== Handle 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 s that produce a reply (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 and this would cause an exception if the handler’s replyRequired property is true.

Starting with version 4.3.1, a new HandleMessageAdvice and the AbstractHandleMessageAdvice base implementation have been introduced. Advice s that implement HandleMessageAdvice will always be 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 handler that returns a response, are dissociated from the adviceChain and properly applied to the MessageHandler.handleMessage() method. Bear in mind, however, that this means the advice chain order is not complied with; and, with configuration such as:

<some-reply-producing-endpoint ... >
    <int:request-handler-advice-chain>
        <tx:advice ... />
        <bean ref="myHandleMessageAdvice" />
    </int:request-handler-advice-chain>
</some-reply-producing-endpoint>

The <tx:advice> is applied to the AbstractReplyProducingMessageHandler.handleRequestMessage(), but myHandleMessageAdvice is applied for to MessageHandler.handleMessage() and, therefore, invoked before the <tx:advice>. To retain the order, you should follow with standard Spring AOP configuration approach and use endpoint id together with the .handler suffix to obtain the target MessageHandler bean. Note, however, that in that case, the entire downstream flow would be within the transaction scope.

In the case of a MessageHandler that does not return a response, the advice chain order is retained.

==== 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 regular TransactionInterceptor is used in the <request-handler-advice-chain>, for example via <tx:advice> configuration, a started transaction is only applied only for an internal AbstractReplyProducingMessageHandler.handleRequestMessage() and isn’t propagated to the downstream flow.

To simplify XML configuration, alongside with the <request-handler-advice-chain>, a <transactional> sub-element has been added to all <outbound-gateway> and <service-activator> & family components:

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

For whom is familiar with JPA Integration components such a configuration isn’t new, but now we can start transaction from any point in our flow, not only from the <poller> or Message Driven Channel Adapter like in JMS.

Java & Annotation configuration can be simplified via newly introduced TransactionInterceptorBuilder and the result bean name can be used in the Messaging Annotations adviceChain attribute:

@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 for the TransactionInterceptorBuilder constructor, which means produce TransactionHandleMessageAdvice, not regular TransactionInterceptor.

Java DSL supports such an Advice via .transactional() options on the endpoint configuration:

@Bean
public IntegrationFlow updatingGatewayFlow() {
    return f -> f
        .handle(Jpa.updatingGateway(this.entityManagerFactory),
                e -> e.transactional(true))
        .channel(c -> c.queue("persistResults"));
}

==== Advising Filters

There is an additional consideration when advising Filter s. 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 will be retried. This is also the case 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.

==== Advising Endpoints Using Annotations

When configuring certain endpoints 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 the section called “CompletableFuture”. An example with the discard being performed after the advice is shown below.

@MessageEndpoint
public class MyAdvisedFilter {

    @Filter(inputChannel="input", outputChannel="output",
            adviceChain="adviceChain", discardWithinAdvice="false")
    public boolean filter(String s) {
        return s.contains("good");
    }
}

==== Ordering Advices within an Advice Chain

Advice classes are "around" advices and are applied in a nested fashion. The first advice is the outermost, the last advice the innermost (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, let’s say 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. Then, each retry will be 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 would put the transaction advice first.

==== Advised Handler Properties

Sometimes, it is useful to access handler properties from within the advice. For example, most handlers implement NamedComponent and you can access the component name.

The target object can be accessed via 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 simply cast the target object to the desired implemented interface, such as NamedComponent.

String componentName = ((NamedComponent) target).getComponentName();

or

String componentName = ((NamedComponent) invocation.getThis()).getComponentName();

when implementing MethodInterceptor directly.

When only the handleRequestMessage() method is advised (in a reply-producing handler), you need to access the full handler, which is an AbstractReplyProducingMessageHandler…​

AbstractReplyProducingMessageHandler handler =
    ((AbstractReplyProducingMessageHandler.RequestHandler) target).getAdvisedHandler();

String componentName = handler.getComponentName();

==== 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, which is applied to the MessageHandler.handleMessage() method and can filter a request message or mark it as a duplicate, according to its configuration.

Previously, users could have implemented this pattern, by using a custom MessageSelector in a <filter/> (Section 6.2, “Filter”), for example. However, since this pattern is really behavior of an endpoint rather than being an endpoint itself, the Idempotent Receiver implementation doesn’t 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 isn’t accepted by that selector, it will be 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 and/or throwExceptionOnRejection = true, the duplicate Message won’t be sent to the target MessageHandler.handleMessage(), but discarded. If you simply 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, the MetadataStoreSelector is provided. It accepts a MessageProcessor implementation (which creates a lookup key based on the Message) and an optional ConcurrentMetadataStore (the section called “CompletableFuture”). See the MetadataStoreSelector JavaDocs for more information. The value for ConcurrentMetadataStore also can be customized using additional MessageProcessor. By default MetadataStoreSelector uses timestamp message header.

For convenience, the MetadataStoreSelector options are configurable directly on the <idempotent-receiver> component:

<idempotent-receiver
        id=""  
        endpoint=""  
        selector=""  
        discard-channel=""  
        metadata-store=""  
        key-strategy=""  
        key-expression=""  
        value-strategy=""  
        value-expression=""  
        throw-exception-on-rejection="" />  

For Java configuration, the method level IdempotentReceiver annotation is provided. It is used to mark a method that has a Messaging annotation (@ServiceActivator, @Router etc.) to specify which IdempotentReceiverInterceptor s will be applied to this endpoint:

@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() {
    ....
}

And with the Java DSL, the interceptor is added to the endpoint’s advice chain:

@Bean
public IntegrationFlow flow() {
    ...
        .handle("someBean", "someMethod",
            e -> e.advice(idempotentReceiverInterceptor())
    ...
}

	Note
	The IdempotentReceiverInterceptor is designed only for the MessageHandler.handleMessage(Message<?>) method and starting with version 4.3.1 it implements HandleMessageAdvice, with the AbstractHandleMessageAdvice as a base class, for better dissociation. See the section called “CompletableFuture” for more information.

=== Logging Channel Adapter

The <logging-channel-adapter/> is often used in conjunction with a Wire Tap, as discussed in the section called “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, say, WARN level. With a NullChannel, you would only see the discarded message when logging at DEBUG level.

<int:logging-channel-adapter
    channel="" 
    level="INFO" 
    expression="" 
    log-full-message="false" 
    logger-name="" /> 

==== Configuring with Java Configuration

The following Spring Boot application provides an example of configuring the LoggingHandler 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);

    }

}

==== Configuring with the Java DSL

The following Spring Boot application provides an example of configuring the logging channel adapter 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);

    }

}

== Java DSL

The Spring Integration JavaConfig and DSL provides a set of convenient Builders and a fluent API to configure Spring Integration message flows from Spring @Configuration classes.

=== Example Configurations

@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();
    }
}

As the result after ApplicationContext start up Spring Integration endpoints and Message Channels will be created as is the case after XML parsing. Such configuration can be used to replace XML configuration or along side with it.

=== Introduction

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 using the fluent Builder pattern together with existing Java and Annotation configurations from Spring Framework and Spring Integration as well. Another useful tool to simplify configuration is Java 8 Lambdas.

The cafe is 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 (@Bean). The builder pattern is used to express arbitrarily complex structures as a hierarchy of methods that may accept Lambdas as arguments.

The IntegrationFlowBuilder just collects integration components (MessageChannel s, AbstractEndpoint s etc.) 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 Java Lambdas are easier and much more powerful.

=== DSL Basics

The org.springframework.integration.dsl package contains the IntegrationFlowBuilder API mentioned above and a bunch of IntegrationComponentSpec implementations which are builders too and provide the fluent API to configure concrete endpoints. The IntegrationFlowBuilder infrastructure provides common 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:

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 isn’t used at runtime:

@Bean
public IntegrationFlow integerFlow() {
    return IntegrationFlows.from("input")
            .<String, Integer>transform(Integer::parseInt)
            .get();
}

Here we use the IntegrationFlows factory to define an IntegrationFlow bean using EIP-methods from IntegrationFlowBuilder.

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>, hence any out-of-the-box transformers (ObjectToJsonTransformer, FileToStringTransformer etc.) can be used here.

Under the covers, IntegrationFlowBuilder recognizes the MessageHandler and endpoint for that: MessageTransformingHandler and ConsumerEndpointFactoryBean, respectively. Let’s look at another example:

@Bean
public IntegrationFlow myFlow() {
    return IntegrationFlows.from("input")
                .filter("World"::equals)
                .transform("Hello "::concat)
                .handle(System.out::println)
                .get();
}

The above example composes a sequence of Filter -> Transformer -> Service Activator. The flow is one way, that is it does not provide a a reply message but simply prints the payload to STDOUT. The endpoints are automatically wired together using direct channels.

	Lambdas And Message<?> Arguments
	When 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 Class<?> as the first parameter. For example, this won’t work:

This will fail at runtime with a ClassCastException because the lambda doesn’t retain the argument type and the framework will attempt to cast the payload to a Message<?>.

Instead, use:

=== Message Channels

In addition to the IntegrationFlowBuilder with EIP-methods the Java DSL provides a fluent API to configure MessageChannel s. For this purpose the MessageChannels builder factory is provided:

@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 an input-channel/output-channel pair in the XML configuration. By default endpoints are wired via DirectChannel s where the bean name is based on the pattern: [IntegrationFlow.beanName].channel#[channelNameIndex]. This rule is applied for unnamed channels produced by inline MessageChannels builder factory usage, too. However all MessageChannels methods have a channelId -aware variant to create the bean names for MessageChannel s. The MessageChannel references can be used as well as beanName, as bean-method invocations. Here is a sample with possible variants of channel() EIP-method usage:

@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();
}

Note: the IntegrationFlow definition shown above is valid and all of its channels are applied to endpoints with BridgeHandler s.

	Important
	Be careful to use the same inline channel definition via MessageChannels factory from different IntegrationFlow s. Even if the DSL parsers register non-existing objects as beans, it can’t determine the same object (MessageChannel) from different IntegrationFlow containers. This 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();
}

You end up with:

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 working there is just need to declare @Bean for that channel and use its bean-method from different IntegrationFlow s.

=== Pollers

A similar fluent API is provided to configure PollerMetadata for AbstractPollingEndpoint implementations. The Pollers builder factory can be used to configure common bean definitions or those created from IntegrationFlowBuilder EIP-methods:

@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerSpec poller() {
    return Pollers.fixedRate(500)
        .errorChannel("myErrors");
}

See Pollers and PollerSpec Java Docs for more information.

	Important
	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 will generate the PollerMetadata object from the specification and initialize all of its properties as needed.

=== DSL and Endpoint Configuration

All IntegrationFlowBuilder EIP-methods have a variant to apply the Lambda parameter to provide options for AbstractEndpoint s: SmartLifecycle, PollerMetadata, request-handler-advice-chain etc. Each of them has generic arguments, so it allows you to simply configure an endpoint and even its MessageHandler in the context:

@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 allow you to register an endpoint bean with a given bean name, rather than a generated one.

=== Transformers

The DSL API provides a convenient, fluent Transformers factory to be used as inline target object definition within .transform() EIP-method:

@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 Transformers can be use to declare target Transformer s as @Bean s and, again, use them from IntegrationFlow definition as bean-methods. Nevertheless, the DSL parser takes care about bean declarations for inline objects, if they aren’t defined as beans yet.

See Transformers Java Docs for more information and supported factory methods.

Also see Lambdas And Message<?> Arguments.

=== Inbound Channel Adapters

Typically message flows start from some Inbound Channel Adapter (e.g. <int-jdbc:inbound-channel-adapter>). The adapter is configured with <poller> and it asks a MessageSource<?> for producing messages periodically. Java DSL allows to start IntegrationFlow from a MessageSource<?>, too. For this purpose IntegrationFlows builder factory provides overloaded IntegrationFlows.from(MessageSource<?> messageSource) method. The MessageSource<?> may be configured as a bean and provided as argument for that method. The second parameter of IntegrationFlows.from() is a Consumer<SourcePollingChannelAdapterSpec> Lambda and allows to provide options for the SourcePollingChannelAdapter, e.g. PollerMetadata or SmartLifecycle:

@Bean
public MessageSource<Object> jdbcMessageSource() {
    return new JdbcPollingChannelAdapter(this.dataSource, "SELECT * FROM foo");
}

@Bean
public IntegrationFlow pollingFlow() {
    return IntegrationFlows.from(jdbcMessageSource(),
                c -> c.poller(Pollers.fixedRate(100).maxMessagesPerPoll(1)))
            .transform(Transformers.toJson())
            .channel("furtherProcessChannel")
            .get();
}

There is also an IntegrationFlows.from() variant based on the java.util.function.Supplier if there is no requirements to build Message objects directly. The result of the Supplier.get() is wrapped to the Message (if it isn’t message already) by Framework automatically.

The next sections discuss selected endpoints which require further explanation.

=== Message Routers

Spring Integration natively provides specialized router types including:

As with many other DSL IntegrationFlowBuilder EIP-methods the route() method can apply any out-of-the-box AbstractMessageRouter implementation, or for convenience a String as a SpEL expression, or a ref/method pair. In addition route() can be configured with a Lambda - the inline method invocation case, and with a Lambda for a Consumer<RouterSpec<MethodInvokingRouter>>. The fluent API also provides AbstractMappingMessageRouter options like channelMapping(String key, String channelName) pairs:

@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();
}

A simple expression-based router:

@Bean
public IntegrationFlow routeFlow() {
    return IntegrationFlows.from("routerInput")
            .route("headers['destChannel']")
            .get();
}

The routeToRecipients() method takes a Consumer<RecipientListRouterSpec>:

@Bean
public IntegrationFlow recipientListFlow() {
    return IntegrationFlows.from("recipientListInput")
            .<String, String>transform(p -> p.replaceFirst("Payload", ""))
                        .routeToRecipients(r -> r
                .recipient("foo-channel", "'foo' == payload")
                .recipient("bar-channel", m ->
                    m.getHeaders().containsKey("recipient")
                        && (boolean) m.getHeaders().get("recipient"))
                .recipientFlow("'foo' == payload or 'bar' == payload or 'baz' == payload",
                    f -> f.<String, String>transform(String::toUpperCase)
                        .channel(c -> c.queue("recipientListSubFlow1Result")))
                .recipientFlow((String p) -> p.startsWith("baz"),
                    f -> f.transform("Hello "::concat)
                        .channel(c -> c.queue("recipientListSubFlow2Result")))
                .recipientFlow(new FunctionExpression<Message<?>>(m ->
                                             "bax".equals(m.getPayload())),
                    f -> f.channel(c -> c.queue("recipientListSubFlow3Result")))
                .defaultOutputToParentFlow())
            .get();
}

The .defaultOutputToParentFlow() of the .routeToRecipients() allows to make 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.

=== Splitters

A splitter is created using the split() EIP-method. By default, if the payload is a Iterable, Iterator, Array, Stream or Reactive Publisher, this will output each item as an individual message. This takes a Lambda, SpEL expression, any AbstractMessageSplitter implementation, or can be used without parameters to provide the DefaultMessageSplitter. For example:

@Bean
public IntegrationFlow splitFlow() {
    return IntegrationFlows.from("splitInput")
              .split(s ->
                      s.applySequence(false).get().getT2().setDelimiters(","))
              .channel(MessageChannels.executor(this.taskExecutor()))
              .get();
}

This creates a splitter that splits a message containing a comma delimited String. Note: the getT2() method comes from Tuple Collection which is the result of EndpointSpec.get() and represents a pair of ConsumerEndpointFactoryBean and DefaultMessageSplitter for the example above.

Also see Lambdas And Message<?> Arguments.

=== Aggregators and Resequencers

An Aggregator is conceptually the converse of a Splitter. It aggregates a sequence of individual messages into a single message and is necessarily more complex. By default, an aggregator will return a message containing a collection of payloads from incoming messages. The same rules are applied for the Resequencer:

@Bean
public IntegrationFlow splitAggregateFlow() {
    return IntegrationFlows.from("splitAggregateInput")
            .split()
            .channel(MessageChannels.executor(this.taskExecutor()))
            .resequence()
            .aggregate()
            .get();
}

The above is a canonical example of splitter/aggregator pattern. The split() method splits the list into individual messages and sends them to the ExecutorChannel. The resequence() method reorders messages by sequence details from message headers. The aggregate() method just collects those messages to the result list.

However, you may change the default behavior by specifying a release strategy and correlation strategy, among other things. Consider the following:

.aggregate(a ->
        a.correlationStrategy(m -> m.getHeaders().get("myCorrelationKey"))
            .releaseStrategy(g -> g.size() > 10)
            .messageStore(messageStore()))

The similar Lambda configurations are provided for the resequence() EIP-method.

=== ServiceActivators (.handle())

The .handle() EIP-method’s goal is to invoke any MessageHandler implementation or any method on some POJO. Another option to define "activity" via Lambda expression. Hence a generic GenericHandler<P> functional interface has been introduced. Its handle method requires two arguments - P payload and Map<String, Object> headers. Having that we can define a flow like this:

@Bean
public IntegrationFlow myFlow() {
    return IntegrationFlows.from("flow3Input")
        .<Integer>handle((p, h) -> p * 2)
        .get();
}

However one main goal of Spring Integration an achieving of loose coupling via runtime type conversion from message payload to target arguments of message handler. Since Java doesn’t support generic type resolution for Lambda classes, we introduced a workaround with additional payloadType argument for the most EIP-methods and LambdaMessageProcessor, which delegates the hard conversion work to the Spring’s ConversionService using provided type and requested message to target method arguments. The IntegrationFlow might look like this:

@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();
}

Of course we register some custom BytesToIntegerConverter within ConversionService and get rid of that additional .transform().

Also see Lambdas And Message<?> Arguments.

=== Operator log()

For convenience to log the message journey throw the Spring Integration flow (<logging-channel-adapter>), a log() operator is presented. Underneath it is represented by the WireTap ChannelInterceptor and LoggingHandler as subscriber. It is responsible to log message incoming into the next endpoint or for the current channel:

.filter(...)
.log(LoggingHandler.Level.ERROR, "test.category", m -> m.getHeaders().getId())
.route(...)

In this example an id header will be logged with ERROR level onto "test.category" only for messages passed the filter and before routing.

=== MessageChannelSpec.wireTap()

A .wireTap() fluent API exists for MessageChannelSpec builders. A target configuration gains much more from Java DSL usage:

@Bean
public QueueChannelSpec myChannel() {
    return MessageChannels.queue()
            .wireTap("loggingFlow.input");
}

@Bean
public IntegrationFlow loggingFlow() {
    return f -> f.log();
}

@Bean
MessageChannel myChannel() {
    return new DirectChannel();
}

...
    .channel(myChannel())
    .log()
}

When current MessageChannel doesn’t implement ChannelInterceptorAware, an implicit DirectChannel and BridgeHandler are injected into the IntegrationFlow and the WireTap is added to this new DirectChannel. And when there is not any channel declaration like in this sample:

.handle(...)
.log()
}

an implicit DirectChannel is injected in the current position of the IntegrationFlow and it is used as an output channel for the currently configured ServiceActivatingHandler (the .handle() above).

	Important
	If log() or wireTap() are used in the end of flow they are considered one-way MessageHandler s. If the integration flow is expected to return a reply, a bridge() should be added to the end, after log() or wireTap():

@Bean
public IntegrationFlow sseFlow() {
    return IntegrationFlows
        .from(WebFlux.inboundGateway("/sse")
            .requestMapping(m ->
                m.produces(MediaType.TEXT_EVENT_STREAM_VALUE)))
        .handle((p, h) -> Flux.just("foo", "bar", "baz"))
        .log(LoggingHandler.Level.WARN)
        .bridge()
        .get();
}

=== Working With Message Flows

As we have seen, IntegrationFlowBuilder provides a top level API to produce Integration components wired to message flows. This is convenient if your integration may be accomplished with a single flow (which is often the case). Alternately IntegrationFlow s can be joined via MessageChannel s.

By default, the MessageFlow behaves as a Chain in Spring Integration parlance. That is, the endpoints are automatically wired implicitly via DirectChannel s. The message flow is not actually constructed as a chain, affording much more flexibility. For example, you may send a message to any component within the flow, if you know its inputChannel name, i.e., 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 the like, instead of direct channels. As such, the DSL does not support the Spring Integration chain element since it doesn’t add much value.

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 Integration XML definitions and wired with Spring Integration Messaging Annotations configuration.

Another alternative to define direct IntegrationFlow s is based on a fact that IntegrationFlow can be declared as Lambda too:

@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 bunch of Integration components wired with implicit direct channel. Only limitation is here, that this flow is started with named direct channel - lambdaFlow.input. And Lambda flow can’t start from MessageSource or MessageProducer.

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 sample above, will end up with te bean name like lambdaFlow.org.springframework.integration.config. ConsumerEndpointFactoryBean#0. The Transformer implementation bean for that endpoint will have a bean name such as lambdaFlow.org.springframework.integration.transformer. MethodInvokingTransformer#0. These generated bean names are prepended with the flow id prefix 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 the section called “CompletableFuture” for more information.

=== FunctionExpression

The FunctionExpression (an implementation of SpEL Expression) has been introduced to get a gain of Java and Lambda usage for the method and its generics context. The Function<T, R> option is provided for the DSL components alongside with expression option, when there is the implicit Strategy variant from Core Spring Integration. The usage may look like:

.enrich(e -> e.requestChannel("enrichChannel")
            .requestPayload(Message::getPayload)
            .propertyFunction("date", m -> new Date()))

The FunctionExpression also supports runtime type conversion as it is done in the standard SpelExpression.

=== Sub Flows support

Some of if...else and publish-subscribe components provide the support to specify their logic or mapping using Sub Flows. The simplest sample is .publishSubscribeChannel():

@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"));
}

Of course the same result we can achieve with separate IntegrationFlow @Bean definitions, but we hope you’ll find the subflow style of logic composition useful.

Similar publish-subscribe subflow composition provides .routeToRecipients().

Another sample is .discardFlow() on the .filter() instead of .discardChannel().

The .route() deserves special attention. As a sample:

@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 in regular Router mapping, but the .subFlowMapping() tied that subflow with main flow. In other words, any router’s subflow returns to the main flow after .route().

Of course, subflows can be nested with any depth, but we don’t recommend to do that because, in fact, even in the router case, adding complex subflows within a flow would quickly begin to look like a plate of spaghetti and difficult for a human to parse.

=== Using Protocol Adapters

All of the examples so far illustrate how the DSL supports a messaging architecture using the Spring Integration programming model, but we haven’t done any real integration yet. This requires access to remote resources via http, jms, amqp, tcp, jdbc, ftp, smtp, and the like, 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 will continually be catching up with Spring Integration.

Anyway we are providing the hi-level API to define protocol-specific seamlessly. This is achieved with Factory and Builder patterns and, of course, with Lambdas. The factory classes can be considered "Namespace Factories", because they play the same role as XML namespace for components from the concrete protocol-specific Spring Integration modules. Currently, Spring Integration Java DSL supports Amqp, Feed, Jms, Files, (S)Ftp, Http, JPA, MongoDb, TCP/UDP, Mail, WebFlux and Scripts namespace factories:

@Bean
public IntegrationFlow amqpFlow() {
    return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue()))
            .transform("hello "::concat)
            .transform(String.class, String::toUpperCase)
            .get();
}

@Bean
public IntegrationFlow jmsOutboundGatewayFlow() {
    return IntegrationFlows.from("jmsOutboundGatewayChannel")
            .handle(Jms.outboundGateway(this.jmsConnectionFactory)
                        .replyContainer(c ->
                                    c.concurrentConsumers(3)
                                            .sessionTransacted(true))
                        .requestDestination("jmsPipelineTest"))
            .get();
}

@Bean
public IntegrationFlow sendMailFlow() {
    return IntegrationFlows.from("sendMailChannel")
            .handle(Mail.outboundAdapter("localhost")
                            .port(smtpPort)
                            .credentials("user", "pw")
                            .protocol("smtp")
                            .javaMailProperties(p -> p.put("mail.debug", "true")),
                    e -> e.id("sendMailEndpoint"))
            .get();
}

We show here the usage of namespace factories as inline adapters declarations, however they can be used from @Bean definitions to make the IntegrationFlow method-chain more readable.

We are soliciting community feedback on these namespace factories before we spend effort on others; we’d also appreciate some prioritization for which adapters/gateways we should support next.

See more Java DSL samples in the protocol-specific chapter throughout this reference manual.

All other protocol channel adapters may be configured as generic beans and wired to the IntegrationFlow:

@Bean
public QueueChannelSpec wrongMessagesChannel() {
    return MessageChannels
            .queue()
            .wireTap("wrongMessagesWireTapChannel");
}

@Bean
public IntegrationFlow xpathFlow(MessageChannel wrongMessagesChannel) {
    return IntegrationFlows.from("inputChannel")
            .filter(new StringValueTestXPathMessageSelector("namespace-uri(/*)", "my:namespace"),
                    e -> e.discardChannel(wrongMessagesChannel))
            .log(LoggingHandler.Level.ERROR, "test.category", m -> m.getHeaders().getId())
            .route(xpathRouter(wrongMessagesChannel))
            .get();
}

@Bean
public AbstractMappingMessageRouter xpathRouter(MessageChannel wrongMessagesChannel) {
    XPathRouter router = new XPathRouter("local-name(/*)");
    router.setEvaluateAsString(true);
    router.setResolutionRequired(false);
    router.setDefaultOutputChannel(wrongMessagesChannel);
    router.setChannelMapping("Tags", "splittingChannel");
    router.setChannelMapping("Tag", "receivedChannel");
    return router;
}

=== IntegrationFlowAdapter

The IntegrationFlow as an interface can be implemented directly and specified as component for scanning:

@Component
public class MyFlow implements IntegrationFlow {

    @Override
    public void configure(IntegrationFlowDefinition<?> f) {
        f.<String, String>transform(String::toUpperCase);
    }

}

And yes, it is picked up by the IntegrationFlowBeanPostProcessor and correctly parsed and registered in the application context.

For convenience and loosely coupled architecture the IntegrationFlowAdapter base class implementation is provided. It requires a buildFlow() method implementation to produce an IntegrationFlowDefinition using one of from() support methods:

@Component
public class MyFlowAdapter extends IntegrationFlowAdapter {

    private final AtomicBoolean invoked = new tomicBoolean();

    public Date nextExecutionTime(TriggerContext triggerContext) {
          return this.invoked.getAndSet(true) ? null : new Date();
    }

    @Override
    protected IntegrationFlowDefinition<?> buildFlow() {
        return from(this, "messageSource",
                      e -> e.poller(p -> p.trigger(this::nextExecutionTime)))
                 .split(this)
  	   .transform(this)
  	   .aggregate(a -> a.processor(this, null), null)
  	   .enrichHeaders(Collections.singletonMap("foo", "FOO"))
  	   .filter(this)
  	   .handle(this)
  	   .channel(c -> c.queue("myFlowAdapterOutput"));
    }

    public String messageSource() {
         return "B,A,R";
    }

    @Splitter
    public String[] split(String payload) {
         return StringUtils.commaDelimitedListToStringArray(payload);
    }

    @Transformer
    public String transform(String payload) {
         return payload.toLowerCase();
    }

    @Aggregator
    public String aggregate(List<String> payloads) {
           return payloads.stream().collect(Collectors.joining());
    }

    @Filter
    public boolean filter(@Header Optional<String> foo) {
            return foo.isPresent();
    }

    @ServiceActivator
    public String handle(String payload, @Header String foo) {
           return payload + ":" + foo;
    }

}

=== Dynamic and runtime Integration Flows

The IntegrationFlow s and therefore all its dependant components can be registered at runtime. This was done previously by the BeanFactory.registerSingleton() hook and now via newly introduced in the Spring Framework 5.0 programmatic BeanDefinition registration with the instanceSupplier hook:

BeanDefinition beanDefinition =
         BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) bean.getClass(), () -> bean)
               .getRawBeanDefinition();

((BeanDefinitionRegistry) this.beanFactory).registerBeanDefinition(beanName, beanDefinition);

and all the necessary bean initialization and lifecycle is done automatically as it is with the standard context configuration bean definitions.

To simplify the development experience Spring Integration introduced IntegrationFlowContext to register and manage IntegrationFlow instances at runtime:

@Autowired
private AbstractServerConnectionFactory server1;

@Autowired
private IntegrationFlowContext flowContext;

...

@Test
public void testTcpGateways() {
    TestingUtilities.waitListening(this.server1, null);

    IntegrationFlow flow = f -> f
            .handle(Tcp.outboundGateway(Tcp.netClient("localhost", this.server1.getPort())
                    .serializer(TcpCodecs.crlf())
                    .deserializer(TcpCodecs.lengthHeader1())
                    .id("client1"))
                .remoteTimeout(m -> 5000))
            .transform(Transformers.objectToString());

    IntegrationFlowRegistration theFlow = this.flowContext.registration(flow).register();
    assertThat(theFlow.getMessagingTemplate().convertSendAndReceive("foo", String.class), equalTo("FOO"));
}

This is useful when we have multi configuration options and have to create several instances of similar flows. So, we can iterate our options and create and register IntegrationFlow s within loop. Another variant when our source of data isn’t Spring-based and we must create it on the fly. Such a sample is Reactive Streams event source:

Flux<Message<?>> messageFlux =
    Flux.just("1,2,3,4")
        .map(v -> v.split(","))
        .flatMapIterable(Arrays::asList)
        .map(Integer::parseInt)
        .map(GenericMessage<Integer>::new);

QueueChannel resultChannel = new QueueChannel();

IntegrationFlow integrationFlow =
    IntegrationFlows.from(messageFlux)
        .<Integer, Integer>transform(p -> p * 2)
        .channel(resultChannel)
        .get();

this.integrationFlowContext.registration(integrationFlow)
            .register();

The IntegrationFlowRegistrationBuilder (as a result of the IntegrationFlowContext.registration()) can be used to specify a bean name for the IntegrationFlow to register, to control its autoStartup and also for additional, non Integration beans registration. Usually those additional beans are connection factories (AMQP, JMS, (S)FTP, TCP/UDP etc.), serializers/deserializers or any other required support components.

Such a dynamically registered IntegrationFlow and all its dependant beans can be removed afterwards using IntegrationFlowRegistration.destroy() callback. See IntegrationFlowContext JavaDocs for more information.

Note

Starting with version 5.0.6, all generated bean names in an IntegrationFlow definition are prepended with flow id as a prefix. It is recommended to always specify an explicit flow id, otherwise a synchronization barrier is initiated in the IntegrationFlowContext to generate the bean name for the IntegrationFlow and register its beans. We synchronize on these two operations to avoid a race condition when the same generated bean name may be used for different IntegrationFlow instances.

Also, starting with version 5.0.6, the registration builder API has a new method useFlowIdAsPrefix(). This is useful if you wish to declare multiple instances of the same flow and avoid bean name collisions if components in the flows have the same id.

For example:

private void registerFlows() {
    IntegrationFlowRegistration flow1 =
              this.flowContext.registration(buildFlow(1234))
                    .id("tcp1")
                    .useFlowIdAsPrefix()
                    .register();

    IntegrationFlowRegistration flow2 =
              this.flowContext.registration(buildFlow(1235))
                    .id("tcp2")
                    .useFlowIdAsPrefix()
                    .register();
}

private IntegrationFlow buildFlow(int port) {
    return f -> f
            .handle(Tcp.outboundGateway(Tcp.netClient("localhost", port)
                    .serializer(TcpCodecs.crlf())
                    .deserializer(TcpCodecs.lengthHeader1())
                    .id("client"))
                .remoteTimeout(m -> 5000))
            .transform(Transformers.objectToString());
}

In this case, the message handler for the first flow can be referenced with bean name tcp1.client.handler.

	Note
	an id is required when using useFlowIdAsPrefix().

=== IntegrationFlow as Gateway

The IntegrationFlow can start from the service interface providing GatewayProxyFactoryBean component:

public interface ControlBusGateway {

    void send(String command);
}

...

@Bean
public IntegrationFlow controlBusFlow() {
    return IntegrationFlows.from(ControlBusGateway.class)
            .controlBus()
            .get();
}

All the proxy for interface methods are supplied with the channel to send messages to the next integration component in the IntegrationFlow. The service interface can be marked with the @MessagingGateway as well as methods with the @Gateway annotations. Nevertheless the requestChannel is ignored and overridden with that internal channel for the next component in the IntegrationFlow. Otherwise such a configuration via IntegrationFlow won’t make sense.

By default a GatewayProxyFactoryBean gets a conventional bean name like [FLOW_BEAN_NAME.gateway]. That id can be changed via @MessagingGateway.name() attribute or the overloaded from(Class<?> serviceInterface, String beanName) factory method.

With the Java 8 on board we even can create such an Integration Gateway with the java.util.function interfaces:

@Bean
public IntegrationFlow errorRecovererFlow() {
    return IntegrationFlows.from(Function.class, "errorRecovererFunction")
            .handle((GenericHandler<?>) (p, h) -> {
                throw new RuntimeException("intentional");
            }, e -> e.advice(retryAdvice()))
            .get();
}

That can be used lately as:

@Autowired
@Qualifier("errorRecovererFunction")
private Function<String, String> errorRecovererFlowGateway;

== System Management

=== Metrics and Management

==== Configuring Metrics Capture

	Note
	Prior to version 4.2 metrics were only available when JMX was enabled. See the section called “CompletableFuture”.

To enable MessageSource, MessageChannel and MessageHandler metrics, add an <int:management/> bean to the application context, or annotate one of your @Configuration classes with @EnableIntegrationManagement. MessageSource s only maintain counts, MessageChannel s and MessageHandler s maintain duration statistics in addition to counts. See the section called “CompletableFuture” and the section called “CompletableFuture” below.

This causes the automatic registration of the IntegrationManagementConfigurer bean in the application context. Only one such bean can exist in the context and it must have the bean name integrationManagementConfigurer if registered manually via a <bean/> definition. This bean applies it’s configuration to beans after all beans in the context have been instantiated.

In addition to metrics, you can control debug logging in the main message flow. It has been found that in very high volume applications, even calls to isDebugEnabled() can be quite expensive with some logging subsystems. You can disable all such logging to avoid this overhead; exception logging (debug or otherwise) are not affected by this setting.

A number of options are available:

<int:management
    default-logging-enabled="true" 
    default-counts-enabled="false" 
    default-stats-enabled="false" 
    counts-enabled-patterns="foo, !baz, ba*" 
    stats-enabled-patterns="fiz, buz" 
    metrics-factory="myMetricsFactory" /> 

@Configuration
@EnableIntegration
@EnableIntegrationManagement(
    defaultLoggingEnabled = "true", 
    defaultCountsEnabled = "false", 
    defaultStatsEnabled = "false", 
    countsEnabled = { "foo", "${count.patterns}" }, 
    statsEnabled = { "qux", "!*" }, 
    MetricsFactory = "myMetricsFactory") 
public static class ContextConfiguration {
...
}

At runtime, counts and statistics can be obtained by calling IntegrationManagementConfigurer getChannelMetrics, getHandlerMetrics and getSourceMetrics, returning MessageChannelMetrics, MessageHandlerMetrics and MessageSourceMetrics respectively.

See the javadocs for complete information about these classes.

When JMX is enabled (see the section called “CompletableFuture”), these metrics are also exposed by the IntegrationMBeanExporter.

Starting with version 5.0.2, the framework will automatically detect if there is a single MetricsFactory bean in the application context and use it instead of the default metrics factory.

==== Micrometer Integration

Starting with version 5.0.3, the presence of a Micrometer MeterRegistry in the application context will trigger support for Micrometer metrics in addition to the inbuilt metrics (inbuilt metrics will be removed in a future release).

	Important
	Micrometer was first supported in version 5.0.2, but changes were made to the Micrometer Meters in version 5.0.3 to make them more suitable for use in dimensional systems. Further changes were made in 5.0.4; if using Micrometer, a minimum of version 5.0.4 is recommended since some of the changes in 5.0.4 were breaking API changes.

Simply add a MeterRegistry bean of choice to the application context. If the IntegrationManagementConfigurer detects exactly one MeterRegistry bean, it will configure a MicrometerMetricsCaptor bean with name integrationMicrometerMetricsCaptor.

For each MessageHandler and MessageChannel, timers are registered. For each MessageSource, a counter is registered.

This only applies to objects that extend AbstractMessageHandler, AbstractMessageChannel and AbstractMessageSource respectively (which is the case for most framework components).

With Micrometer metrics, the statsEnabled flag takes no effect, since statistics capture is delegated to Micrometer. The countsEnabled flag controls whether the Micrometer Meter s are updated when processing each message.

The Timer Meters for send operations on message channels have the following name/tags:

(A failure result with a none exception means the channel send() operation returned false).

The Counter Meters for receive operations on pollable message channels have the following names/tags:

The Timer Meters for operations on message handlers have the following name/tags:

The Counter meters for message sources have the following names/tags:

In addition, there are three Gauge Meters:

spring.integration.channels - the number of MessageChannels in the application. spring.integration.handlers - the number of MessageHandlers in the application. spring.integration.sources - the number of MessageSources in the application.

==== MessageChannel Metric Features

These legacy metrics will be removed in a future release; see the section called “CompletableFuture”.

Message channels report metrics according to their concrete type. If you are looking at a DirectChannel, you will see statistics for the send operation. If it is a QueueChannel, you will also see statistics for the receive operation, as well as the count of messages that are currently buffered by this QueueChannel. In both cases there are some metrics that are simple counters (message count and error count), and some that are estimates of averages of interesting quantities. The algorithms used to calculate these estimates are described briefly in the section below.

==== MessageHandler Metric Features

These legacy metrics will be removed in a future release; see the section called “CompletableFuture”.

The following table shows the statistics maintained for message handlers. Some metrics are simple counters (message count and error count), and one is an estimate of averages of send duration. The algorithms used to calculate these estimates are described briefly in the table below:

==== Time-Based Average Estimates

A feature of the time-based average estimates is that they decay with time if no new measurements arrive. To help interpret the behaviour over time, the time (in seconds) since the last measurement is also exposed as a metric.

There are two basic exponential models: decay per measurement (appropriate for duration and anything where the number of measurements is part of the metric), and decay per time unit (more suitable for rate measurements where the time in between measurements is part of the metric). Both models depend on the fact that

S(n) = sum(i=0,i=n) w(i) x(i) has a special form when w(i) = r^i, with r=constant:

S(n) = x(n) + r S(n-1) (so you only have to store S(n-1), not the whole series x(i), to generate a new metric estimate from the last measurement). The algorithms used in the duration metrics use r=exp(-1/M) with M=10. The net effect is that the estimate S(n) is more heavily weighted to recent measurements and is composed roughly of the last M measurements. So M is the "window" or lapse rate of the estimate In the case of the vanilla moving average, i is a counter over the number of measurements. In the case of the rate we interpret i as the elapsed time, or a combination of elapsed time and a counter (so the metric estimate contains contributions roughly from the last M measurements and the last T seconds).

==== Metrics Factory

A strategy interface MetricsFactory has been introduced allowing you to provide custom channel metrics for your MessageChannel s and MessageHandler s. By default, a DefaultMetricsFactory provides default implementation of MessageChannelMetrics and MessageHandlerMetrics which are described above. To override the default MetricsFactory configure it as described above, by providing a reference to your MetricsFactory bean instance. You can either customize the default implementations as described in the next bullet, or provide completely different implementations by extending AbstractMessageChannelMetrics and/or AbstractMessageHandlerMetrics.

Also see the section called “CompletableFuture”.

In addition to the default metrics factory described above, the framework provides the AggregatingMetricsFactory. This factory creates AggregatingMessageChannelMetrics and AggregatingMessageHandlerMetrics. In very high volume scenarios, the cost of capturing statistics can be prohibitive (2 calls to the system time and storing the data for each message). The aggregating metrics aggregate the response time over a sample of messages. This can save significant CPU time.

	Caution
	The statistics will be skewed if messages arrive in bursts. These metrics are intended for use with high, constant-volume, message rates.

<bean id="aggregatingMetricsFactory"
            class="org.springframework.integration.support.management.AggregatingMetricsFactory">
    <constructor-arg value="1000" /> <!-- sample size -->
</bean>

The above configuration aggregates the duration over 1000 messages. Counts (send, error) are maintained per-message but the statistics are per 1000 messages.

See the section called “CompletableFuture” and the Javadocs for the ExponentialMovingAverage* classes for more information about these values.

By default, the DefaultMessageChannelMetrics and DefaultMessageHandlerMetrics use a window of 10 measurements, a rate period of 1 second (rate per second) and a decay lapse period of 1 minute.

If you wish to override these defaults, you can provide a custom MetricsFactory that returns appropriately configured metrics and provide a reference to it to the MBean exporter as described above.

Example:

public static class CustomMetrics implements MetricsFactory {

    @Override
    public AbstractMessageChannelMetrics createChannelMetrics(String name) {
        return new DefaultMessageChannelMetrics(name,
                new ExponentialMovingAverage(20, 1000000.),
                new ExponentialMovingAverageRate(2000, 120000, 30, true),
                new ExponentialMovingAverageRatio(130000, 40, true),
                new ExponentialMovingAverageRate(3000, 140000, 50, true));
    }

    @Override
    public AbstractMessageHandlerMetrics createHandlerMetrics(String name) {
        return new DefaultMessageHandlerMetrics(name, new ExponentialMovingAverage(20, 1000000.));
    }

}

The customizations described above are wholesale and will apply to all appropriate beans exported by the MBean exporter. This is the extent of customization available using XML configuration.

Individual beans can be provided with different implementations using java @Configuration or programmatically at runtime, after the application context has been refreshed, by invoking the configureMetrics methods on AbstractMessageChannel and AbstractMessageHandler.

Previously, the time-based metrics (see the section called “CompletableFuture”) were calculated in real time. The statistics are now calculated when retrieved instead. This resulted in a significant performance improvement, at the expense of a small amount of additional memory for each statistic. As discussed in the bullet above, the statistics can be disabled altogether, while retaining the MBean allowing the invocation of Lifecycle methods.

=== JMX Support

Spring Integration provides Channel Adapters for receiving and publishing JMX Notifications. There is also an_Inbound Channel Adapter_ for polling JMX MBean attribute values, and an Outbound Channel Adapter for invoking JMX MBean operations.

==== Notification Listening Channel Adapter

The Notification-listening Channel Adapter requires a JMX ObjectName for the MBean that publishes notifications to which this listener should be registered. A very simple configuration might look like this:

<int-jmx:notification-listening-channel-adapter id="adapter"
    channel="channel"
    object-name="example.domain:name=publisher"/>

	Tip
	The notification-listening-channel-adapter registers with an MBeanServer at startup, and the default bean name is mbeanServer which happens to be the same bean name generated when using Spring’s <context:mbean-server/> element. If you need to use a different name, be sure to include the_mbean-server_ attribute.

The adapter can also accept a reference to a NotificationFilter and a handback Object to provide some context that is passed back with each Notification. Both of those attributes are optional. Extending the above example to include those attributes as well as an explicit MBeanServer bean name would produce the following:

<int-jmx:notification-listening-channel-adapter id="adapter"
    channel="channel"
    mbean-server="someServer"
    object-name="example.domain:name=somePublisher"
    notification-filter="notificationFilter"
    handback="myHandback"/>

The Notification-listening Channel Adapter is event-driven and registered with the MBeanServer directly. It does not require any poller configuration.

Note

For this component only, the object-name attribute can contain an ObjectName pattern (e.g. "org.foo:type=Bar,name=*") and the adapter will receive notifications from all MBeans with ObjectNames that match the pattern. In addition, the object-name attribute can contain a SpEL reference to a <util:list/> of ObjectName patterns: <jmx:notification-listening-channel-adapter id="manyNotificationsAdapter" channel="manyNotificationsChannel" object-name="#{patterns}"/> <util:list id="patterns"> <value>org.foo:type=Foo,name=*</value> <value>org.foo:type=Bar,name=*</value> </util:list> The names of the located MBean(s) will be logged when DEBUG level logging is enabled.

==== Notification Publishing Channel Adapter

The Notification-publishing Channel Adapter is relatively simple. It only requires a JMX ObjectName in its configuration as shown below.

<context:mbean-export/>

<int-jmx:notification-publishing-channel-adapter id="adapter"
    channel="channel"
    object-name="example.domain:name=publisher"/>

It does also require that an MBeanExporter be present in the context. That is why the <context:mbean-export/> element is shown above as well.

When Messages are sent to the channel for this adapter, the Notification is created from the Message content. If the payload is a String it will be passed as the message text for the Notification. Any other payload type will be passed as the userData of the Notification.

JMX Notifications also have a type, and it should be a dot-delimited String. There are two ways to provide the type. Precedence will always be given to a Message header value associated with the JmxHeaders.NOTIFICATION_TYPE key. On the other hand, you can rely on a fallback default-notification-type attribute provided in the configuration.

<context:mbean-export/>

<int-jmx:notification-publishing-channel-adapter id="adapter"
    channel="channel"
    object-name="example.domain:name=publisher"
    default-notification-type="some.default.type"/>

==== Attribute Polling Channel Adapter

The Attribute Polling Channel Adapter is useful when you have a requirement, to periodically check on some value that is available through an MBean as a managed attribute. The poller can be configured in the same way as any other polling adapter in Spring Integration (or it’s possible to rely on the default poller). The object-name and attribute-name are required. An MBeanServer reference is also required, but it will automatically check for a bean named mbeanServer by default, just like the Notification-listening Channel Adapter described above.

<int-jmx:attribute-polling-channel-adapter id="adapter"
    channel="channel"
    object-name="example.domain:name=someService"
    attribute-name="InvocationCount">
        <int:poller max-messages-per-poll="1" fixed-rate="5000"/>
</int-jmx:attribute-polling-channel-adapter>

==== Tree Polling Channel Adapter

The Tree Polling Channel Adapter queries the JMX MBean tree and sends a message with a payload that is the graph of objects that matches the query. By default the MBeans are mapped to primitives and simple Objects like Map, List and arrays - permitting simple transformation, for example, to JSON. An MBeanServer reference is also required, but it will automatically check for a bean named mbeanServer by default, just like the Notification-listening Channel Adapter described above. A basic configuration would be:

<int-jmx:tree-polling-channel-adapter id="adapter"
    channel="channel"
    query-name="example.domain:type=*">
        <int:poller max-messages-per-poll="1" fixed-rate="5000"/>
</int-jmx:tree-polling-channel-adapter>

This will include all attributes on the MBeans selected. You can filter the attributes by providing an MBeanObjectConverter that has an appropriate filter configured. The converter can be provided as a reference to a bean definition using the converter attribute, or as an inner <bean/> definition. A DefaultMBeanObjectConverter is provided which can take a MBeanAttributeFilter in its constructor argument.

Two standard filters are provided; the NamedFieldsMBeanAttributeFilter allows you to specify a list of attributes to include and the NotNamedFieldsMBeanAttributeFilter allows you to specify a list of attributes to exclude. You can also implement your own filter

==== Operation Invoking Channel Adapter

The operation-invoking-channel-adapter enables Message-driven invocation of any managed operation exposed by an MBean. Each invocation requires the operation name to be invoked and the ObjectName of the target MBean. Both of these must be explicitly provided via adapter configuration:

<int-jmx:operation-invoking-channel-adapter id="adapter"
    object-name="example.domain:name=TestBean"
    operation-name="ping"/>

Then the adapter only needs to be able to discover the mbeanServer bean. If a different bean name is required, then provide the mbean-server attribute with a reference.

The payload of the Message will be mapped to the parameters of the operation, if any. A Map-typed payload with String keys is treated as name/value pairs, whereas a List or array would be passed as a simple argument list (with no explicit parameter names). If the operation requires a single parameter value, then the payload can represent that single value, and if the operation requires no parameters, then the payload would be ignored.

If you want to expose a channel for a single common operation to be invoked by Messages that need not contain headers, then that option works well.

==== Operation Invoking Outbound Gateway

Similar to the operation-invoking-channel-adapter Spring Integration also provides a operation-invoking-outbound-gateway, which could be used when dealing with non-void operations and a return value is required. Such return value will be sent as message payload to the reply-channel specified by this Gateway.

<int-jmx:operation-invoking-outbound-gateway request-channel="requestChannel"
   reply-channel="replyChannel"
   object-name="o.s.i.jmx.config:type=TestBean,name=testBeanGateway"
   operation-name="testWithReturn"/>

If the reply-channel attribute is not provided, the reply message will be sent to the channel that is identified by the IntegrationMessageHeaderAccessor.REPLY_CHANNEL header. That header is typically auto-created by the entry point into a message flow, such as any Gateway component. However, if the message flow was started by manually creating a Spring Integration Message and sending it directly to a Channel, then you must specify the message header explicitly or use the provided reply-channel attribute.

==== MBean Exporter

Spring Integration components themselves may be exposed as MBeans when the IntegrationMBeanExporter is configured. To create an instance of the IntegrationMBeanExporter, define a bean and provide a reference to an MBeanServer and a domain name (if desired). The domain can be left out, in which case the default domain is org.springframework.integration.

<int-jmx:mbean-export id="integrationMBeanExporter"
            default-domain="my.company.domain" server="mbeanServer"/>

<bean id="mbeanServer" class="org.springframework.jmx.support.MBeanServerFactoryBean">
    <property name="locateExistingServerIfPossible" value="true"/>
</bean>

Important

The MBean exporter is orthogonal to the one provided in Spring core - it registers message channels and message handlers, but not itself. You can expose the exporter itself, and certain other components in Spring Integration, using the standard <context:mbean-export/> tag. The exporter has a some metrics attached to it, for instance a count of the number of active handlers and the number of queued messages. It also has a useful operation, as discussed in the section called “CompletableFuture”.

Starting with Spring Integration 4.0 the @EnableIntegrationMBeanExport annotation has been introduced for convenient configuration of a default (integrationMbeanExporter) bean of type IntegrationMBeanExporter with several useful options at the @Configuration class level. For example:

@Configuration
@EnableIntegration
@EnableIntegrationMBeanExport(server = "mbeanServer", managedComponents = "input")
public class ContextConfiguration {

	@Bean
	public MBeanServerFactoryBean mbeanServer() {
		return new MBeanServerFactoryBean();
	}
}

If there is a need to provide more options, or have several IntegrationMBeanExporter beans e.g. for different MBean Servers, or to avoid conflicts with the standard Spring MBeanExporter (e.g. via @EnableMBeanExport), you can simply configure an IntegrationMBeanExporter as a generic bean.

===== MBean ObjectNames

All the MessageChannel, MessageHandler and MessageSource instances in the application are wrapped by the MBean exporter to provide management and monitoring features. The generated JMX object names for each component type are listed in the table below:

o.s.i:type=MessageChannel,name=<channelName>

o.s.i:type=MessageSource,name=<channelName>,bean=<source>

o.s.i:type=MessageSource,name=<channelName>,bean=<source>

The bean attribute in the object names for sources and handlers takes one of the values in the table below:

Custom elements can be appended to the object name by providing a reference to a Properties object in the object-name-static-properties attribute.

Also, since Spring Integration 3.0, you can use a custom ObjectNamingStrategy using the object-naming-strategy attribute. This permits greater control over the naming of the MBeans. For example, to group all Integration MBeans under an Integration type. A simple custom naming strategy implementation might be:

public class Namer implements ObjectNamingStrategy {

	private final ObjectNamingStrategy realNamer = new KeyNamingStrategy();
	@Override
	public ObjectName getObjectName(Object managedBean, String beanKey) throws MalformedObjectNameException {
		String actualBeanKey = beanKey.replace("type=", "type=Integration,componentType=");
		return realNamer.getObjectName(managedBean, actualBeanKey);
	}

}

The beanKey argument is a String containing the standard object name beginning with the default-domain and including any additional static properties. This example simply moves the standard type part to componentType and sets the type to Integration, enabling selection of all Integration MBeans in one query:"my.domain:type=Integration,*. This also groups the beans under one tree entry under the domain in tools like VisualVM.

	Note
	The default naming strategy is a MetadataNamingStrategy. The exporter propagates the default-domain to that object to allow it to generate a fallback object name if parsing of the bean key fails. If your custom naming strategy is a MetadataNamingStrategy (or subclass), the exporter will not propagate the default-domain; you will need to configure it on your strategy bean.

Starting with version 5.0.9; any bean names (represented by the name key in the object name) can be quoted if they contain any characters that are not allowed in a Java identifier (or period .). This requires setting spring.integration.jmx.quote.names=true in a META-INF/spring.integration.properties file on the class path. In 5.1 it will not be configurable.

===== JMX Improvements

Version 4.2 introduced some important improvements, representing a fairly major overhaul to the JMX support in the framework. These resulted in a significant performance improvement of the JMX statistics collection and much more control thereof, but has some implications for user code in a few specific (uncommon) situations. These changes are detailed below, with a caution where necessary.

Previously, MessageSource, MessageChannel and MessageHandler metrics were captured by wrapping the object in a JDK dynamic proxy to intercept appropriate method calls and capture the statistics. The proxy was added when an integration MBean exporter was declared in the context.

Now, the statistics are captured by the beans themselves; see the section called “CompletableFuture” for more information.

Warning

This change means that you no longer automatically get an MBean or statistics for custom MessageHandler implementations, unless those custom handlers extend AbstractMessageHandler. The simplest way to resolve this is to extend AbstractMessageHandler. If that’s not possible, or desired, another work-around is to implement the MessageHandlerMetrics interface. For convenience, a DefaultMessageHandlerMetrics is provided to capture and report statistics. Invoke the beforeHandle and afterHandle at the appropriate times. Your MessageHandlerMetrics methods can then delegate to this object to obtain each statistic. Similarly, MessageSource implementations must extend AbstractMessageSource or implement MessageSourceMetrics. Message sources only capture a count so there is no provided convenience class; simply maintain the count in an AtomicLong field.

The removal of the proxy has two additional benefits; 1) stack traces in exceptions are reduced (when JMX is enabled) because the proxy is not on the stack; 2) cases where 2 MBeans were exported for the same bean now only export a single MBean with consolidated attributes/operations (see the MBean consolidation bullet below).

System.nanoTime() is now used to capture times instead of System.currentTimeMillis(). This may provide more accuracy on some JVMs, espcially when durations of less than 1 millisecond are expected

Previously, when JMX was enabled, all sources, channels, handlers captured statistics. It is now possible to control whether the statisics are enabled on an individual component. Further, it is possible to capture simple counts on MessageChannel s and MessageHandler s instead of the complete time-based statistics. This can have significant performance implications because you can selectively configure where you need detailed statistics, as well as enable/disable at runtime.

See the section called “CompletableFuture”.

Similar to the @ManagedResource annotation, the @IntegrationManagedResource marks a class as eligible to be exported as an MBean; however, it will only be exported if there is an IntegrationMBeanExporter in the application context.

Certain Spring Integration classes (in the org.springframework.integration) package) that were previously annotated with`@ManagedResource` are now annotated with both @ManagedResource and @IntegrationManagedResource. This is for backwards compatibility (see the next bullet). Such MBeans will be exported by any context MBeanServeror an IntegrationMBeanExporter (but not both - if both exporters are present, the bean is exported by the integration exporter if the bean matches a managed-components pattern).

Certain classes within the framework (mapping routers for example) have additional attributes/operations over and above those provided by metrics and Lifecycle. We will use a Router as an example here.

Previously, beans of these types were exported as two distinct MBeans:

1) the metrics MBean (with an objectName such as: intDomain:type=MessageHandler,name=myRouter,bean=endpoint). This MBean had metrics attributes and metrics/Lifecycle operations.

2) a second MBean (with an objectName such as: ctxDomain:name=org.springframework.integration.config.RouterFactoryBean#0 ,type=MethodInvokingRouter) was exported with the channel mappings attribute and operations.

Now, the attributes and operations are consolidated into a single MBean. The objectName will depend on the exporter. If exported by the integration MBean exporter, the objectName will be, for example: intDomain:type=MessageHandler,name=myRouter,bean=endpoint. If exported by another exporter, the objectName will be, for example: ctxDomain:name=org.springframework.integration.config.RouterFactoryBean#0 ,type=MethodInvokingRouter. There is no difference between these MBeans (aside from the objectName), except that the statistics will not be enabled (the attributes will be 0) by exporters other than the integration exporter; statistics can be enabled at runtime using the JMX operations. When exported by the integration MBean exporter, the initial state can be managed as described above.

	Warning
	If you are currently using the second MBean to change, for example, channel mappings, and you are using the integration MBean exporter, note that the objectName has changed because of the MBean consolidation. There is no change if you are not using the integration MBean exporter.

Previously, the managed-components patterns were inclusive only. If a bean name matched one of the patterns it would be included. Now, the pattern can be negated by prefixing it with !. i.e. "!foo*, foox" will match all beans that don’t start with foo, except foox. Patterns are evaluated left to right and the first match (positive or negative) wins and no further patterns are applied.

	Warning
	The addition of this syntax to the pattern causes one possible (although perhaps unlikely) problem. If you have a bean "!foo"and you included a pattern "!foo" in your MBean exporter’s managed-components patterns; it will no long match; the pattern will now match all beans not named foo. In this case, you can escape the ! in the pattern with \. The pattern "\!foo" means match a bean named "!foo".

The IntegrationMBeanExporter no longer implements SmartLifecycle; this means that start() and stop() operations are no longer available to register/unregister MBeans. The MBeans are now registered during context initialization and unregistered when the context is destroyed.

===== Orderly Shutdown Managed Operation

The MBean exporter provides a JMX operation to shut down the application in an orderly manner, intended for use before terminating the JVM.

public void stopActiveComponents(long howLong)

Its use and operation are described in the section called “CompletableFuture”.

=== Message History

The key benefit of a messaging architecture is loose coupling where participating components do not maintain any awareness about one another. This fact alone makes your application extremely flexible, allowing you to change components without affecting the rest of the flow, change messaging routes,   message consuming styles (polling vs event driven), and so on. However, this unassuming style of architecture could prove to be difficult when things go wrong. When debugging, you would probably like to get as much information about the message as you can (its origin, channels it has traversed, etc.)

Message History is one of those patterns that helps by giving you an option to maintain some level of awareness of a message path either for debugging purposes or to maintain an audit trail. Spring integration provides a simple way to configure your message flows to maintain the Message History by adding a header to the Message and updating that header every time a message passes through a tracked component.

==== Message History Configuration

To enable Message History all you need is to define the message-history element in your configuration.

<int:message-history/>

Now every named component (component that has an id defined) will be tracked. The framework will set the history header in your Message. Its value is very simple - List<Properties>.

<int:gateway id="sampleGateway" 
    service-interface="org.springframework.integration.history.sample.SampleGateway"
    default-request-channel="bridgeInChannel"/>

<int:chain id="sampleChain" input-channel="chainChannel" output-channel="filterChannel">
  <int:header-enricher>
    <int:header name="baz" value="baz"/>
  </int:header-enricher>
</int:chain>

The above configuration will produce a very simple Message History structure:

[{name=sampleGateway, type=gateway, timestamp=1283281668091},
 {name=sampleChain, type=chain, timestamp=1283281668094}]

To get access to Message History all you need is access the MessageHistory header. For example:

Iterator<Properties> historyIterator =
    message.getHeaders().get(MessageHistory.HEADER_NAME, MessageHistory.class).iterator();
assertTrue(historyIterator.hasNext());
Properties gatewayHistory = historyIterator.next();
assertEquals("sampleGateway", gatewayHistory.get("name"));
assertTrue(historyIterator.hasNext());
Properties chainHistory = historyIterator.next();
assertEquals("sampleChain", chainHistory.get("name"));

You might not want to track all of the components. To limit the history to certain components based on their names, all you need is provide the tracked-components attribute and specify a comma-delimited list of component names and/or patterns that match the components you want to track.

<int:message-history tracked-components="*Gateway, sample*, foo"/>

In the above example, Message History will only be maintained for all of the components that end with Gateway, start with sample, or match the name foo exactly.

Starting with version 4.0, you can also use the @EnableMessageHistory annotation in a @Configuration class. In addition, the MessageHistoryConfigurer bean is now exposed as a JMX MBean by the IntegrationMBeanExporter (see the section called “CompletableFuture”), allowing the patterns to be changed at runtime. Note, however, that the bean must be stopped (turning off message history) in order to change the patterns. This feature might be useful to temporarily turn on history to analyze a system. The MBean’s object name is "<domain>:name=messageHistoryConfigurer,type=MessageHistoryConfigurer".

	Important
	If multiple beans (declared by @EnableMessageHistory and/or <message-history/>) they all must have identical component name patterns (when trimmed and sorted). Do not use a generic <bean/> definition for the MessageHistoryConfigurer.

Note

Remember that by definition the Message History header is immutable (you can’t re-write history, although some try). Therefore, when writing Message History values, the components are either creating brand new Messages (when the component is an origin), or they are copying the history from a request Message, modifying it and setting the new list on a reply Message. In either case, the values can be appended even if the Message itself is crossing thread boundaries. That means that the history values can greatly simplify debugging in an asynchronous message flow.

=== Message Store

Enterprise Integration Patterns (EIP) identifies several patterns that have the capability to buffer messages. For example, an Aggregator buffers messages until they can be released and a QueueChannel buffers messages until consumers explicitly receive those messages from that channel. Because of the failures that can occur at any point within your message flow, EIP components that buffer messages also introduce a point where messages could be lost.

To mitigate the risk of losing Messages, EIP defines the Message Store pattern which allows EIP components to store Messages typically in some type of persistent store (e.g. RDBMS).

Spring Integration provides support for the Message Store pattern by a) defining a org.springframework.integration.store.MessageStore strategy interface, b) providing several implementations of this interface, and c) exposing a message-store attribute on all components that have the capability to buffer messages so that you can inject any instance that implements the MessageStore interface.

Details on how to configure a specific Message Store implementation and/or how to inject a MessageStore implementation into a specific buffering component are described throughout the manual (see the specific component, such as QueueChannel, Aggregator, Delayer etc.), but here are a couple of samples to give you an idea:

QueueChannel

<int:channel id="myQueueChannel">
    <int:queue message-store="refToMessageStore"/>
<int:channel>

Aggregator

<int:aggregator … message-store="refToMessageStore"/>

By default Messages are stored in-memory using org.springframework.integration.store.SimpleMessageStore, an implementation of MessageStore. That might be fine for development or simple low-volume environments where the potential loss of non-persistent messages is not a concern. However, the typical production application will need a more robust option, not only to mitigate the risk of message loss but also to avoid potential out-of-memory errors. Therefore, we also provide MessageStore implementations for a variety of data-stores. Below is a complete list of supported implementations:

Important

However be aware of some limitations while using persistent implementations of the MessageStore. The Message data (payload and headers) is serialized and deserialized using different serialization strategies depending on the implementation of the MessageStore. For example, when using JdbcMessageStore, only Serializable data is persisted by default. In this case non-Serializable headers are removed before serialization occurs. Also be aware of the protocol specific headers that are injected by transport adapters (e.g., FTP, HTTP, JMS etc.). For example, <http:inbound-channel-adapter/> maps HTTP-headers into Message Headers and one of them is an ArrayList of non-Serializable org.springframework.http.MediaType instances. However you are able to inject your own implementation of the Serializer and/or Deserializer strategy interfaces into some MessageStore implementations (such as JdbcMessageStore) to change the behaviour of serialization and deserialization. Special attention must be paid to the headers that represent certain types of data. For example, if one of the headers contains an instance of some Spring Bean, upon deserialization you may end up with a different instance of that bean, which directly affects some of the implicit headers created by the framework (e.g., REPLY_CHANNEL or ERROR_CHANNEL). Currently they are not serializable, but even if they were, the deserialized channel would not represent the expected instance. Beginning with Spring Integration version 3.0, this issue can be resolved with a header enricher, configured to replace these headers with a name after registering the channel with the HeaderChannelRegistry. Also when configuring a message-flow like this: gateway → queue-channel (backed by a persistent Message Store) → service-activator That gateway creates a Temporary Reply Channel, and it will be lost by the time the service-activator’s poller reads from the queue. Again, you can use the header enricher to replace the headers with a String representation. For more information, refer to the Section 7.2.2, “Header Enricher”.

Spring Integration 4.0 introduced two new interfaces ChannelMessageStore - to implement operations specific for QueueChannel s, PriorityCapableChannelMessageStore - to mark MessageStore implementation to be used for PriorityChannel s and to provide priority order for persisted Messages. The real behaviour depends on implementation. The Framework provides these implementations, which can be used as a persistent MessageStore for QueueChannel and PriorityChannel:

Caution with SimpleMessageStore

Starting with version 4.1, the SimpleMessageStore no longer copies the message group when calling getMessageGroup(). For large message groups, this was a significant performance problem. 4.0.1 introduced a boolean copyOnGet allowing this to be controlled. When used internally by the aggregator, this was set to false to improve performance. It is now false by default. Users accessing the group store outside of components such as aggregators, will now get a direct reference to the group being used by the aggregator, instead of a copy. Manipulation of the group outside of the aggregator may cause unpredictable results. For this reason, users should not perform such manipulation, or set the copyOnGet property to true.

==== MessageGroupFactory

Starting with version 4.3, some MessageGroupStore implementations can be injected with a custom MessageGroupFactory strategy to create/customize the MessageGroup instances used by the MessageGroupStore. This defaults to a SimpleMessageGroupFactory which produces SimpleMessageGroup s based on the GroupType.HASH_SET (LinkedHashSet) internal collection. Other possible options are SYNCHRONISED_SET and BLOCKING_QUEUE, where the last one can be used to reinstate the previous SimpleMessageGroup behavior. Also the PERSISTENT option is available. See the next section for more information. Starting with _version 5.0.1, the LIST option is also available for use-cases when the order and uniqueness of messages in the group doesn’t matter.

==== Persistence MessageGroupStore and Lazy-Load

Starting with version 4.3, all persistence MessageGroupStore s retrieve MessageGroup s and their messages from the store with the Lazy-Load manner. In most cases it is useful for the Correlation MessageHandler s (Section 6.4, “Aggregator” and Section 6.5, “Resequencer”), when it would be an overhead to load entire MessageGroup from the store on each correlation operation.

To switch off the lazy-load behavior the AbstractMessageGroupStore.setLazyLoadMessageGroups(false) option can be used from the configuration.

Our performance tests for lazy-load on MongoDB MessageStore (the section called “CompletableFuture”) and <aggregator> (Section 6.4, “Aggregator”) with custom release-strategy like:

<int:aggregator input-channel="inputChannel"
                output-channel="outputChannel"
                message-store="mongoStore"
                release-strategy-expression="size() == 1000"/>

demonstrate this results for 1000 simple messages:

StopWatch 'Lazy-Load Performance': running time (millis) = 38918
-----------------------------------------
ms     %     Task name
-----------------------------------------
02652  007%  Lazy-Load
36266  093%  Eager

=== Metadata Store

Many external systems, services or resources aren’t transactional (Twitter, RSS, file system etc.) and there is no any ability to mark the data as read. Or there is just need to implement the Enterprise Integration Pattern Idempotent Receiver in some integration solutions. To achieve this goal and store some previous state of the Endpoint before the next interaction with external system, or deal with the next Message, Spring Integration provides the Metadata Store component being an implementation of the org.springframework.integration.metadata.MetadataStore interface with a general key-value contract.

The Metadata Store is designed to store various types of generic meta-data (e.g., published date of the last feed entry that has been processed) to help components such as the Feed adapter deal with duplicates. If a component is not directly provided with a reference to a MetadataStore, the algorithm for locating a metadata store is as follows: First, look for a bean with id metadataStore in the ApplicationContext. If one is found then it will be used, otherwise it will create a new instance of SimpleMetadataStore which is an in-memory implementation that will only persist metadata within the lifecycle of the currently running Application Context. This means that upon restart you may end up with duplicate entries.

If you need to persist metadata between Application Context restarts, these persistent MetadataStores are provided by the framework:

The PropertiesPersistingMetadataStore is backed by a properties file and a PropertiesPersister.

By default, it only persists the state when the application context is closed normally. It implements Flushable so you can persist the state at will, be invoking flush().

<bean id="metadataStore"
    class="org.springframework.integration.metadata.PropertiesPersistingMetadataStore"/>

Alternatively, you can provide your own implementation of the MetadataStore interface (e.g. JdbcMetadataStore) and configure it as a bean in the Application Context.

Starting with version 4.0, SimpleMetadataStore, PropertiesPersistingMetadataStore and RedisMetadataStore implement ConcurrentMetadataStore. These provide for atomic updates and can be used across multiple component or application instances.

==== Idempotent Receiver and Metadata Store

The Metadata Store is useful for implementing the EIP Idempotent Receiver pattern, when there is need to filter an incoming Message if it has already been processed, and just discard it or perform some other logic on discarding. The following configuration is an example of how to do this:

<int:filter input-channel="serviceChannel"
			output-channel="idempotentServiceChannel"
			discard-channel="discardChannel"
			expression="@metadataStore.get(headers.businessKey) == null"/>

<int:publish-subscribe-channel id="idempotentServiceChannel"/>

<int:outbound-channel-adapter channel="idempotentServiceChannel"
                              expression="@metadataStore.put(headers.businessKey, '')"/>

<int:service-activator input-channel="idempotentServiceChannel" ref="service"/>

The value of the idempotent entry may be some expiration date, after which that entry should be removed from Metadata Store by some scheduled reaper.

Also see the section called “CompletableFuture”.

==== MetadataStoreListener

Some metadata stores (currently only zookeeper) support registering a listener to receive events when items change.

public interface MetadataStoreListener {

	void onAdd(String key, String value);

	void onRemove(String key, String oldValue);

	void onUpdate(String key, String newValue);
}

See the javadocs for more information. The MetadataStoreListenerAdapter can be subclassed if you are only interested in a subset of events.

=== Control Bus

As described in (EIP), the idea behind the Control Bus is that the same messaging system can be used 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 above so that it’s possible to send Messages as a means of invoking exposed operations.

<int:control-bus input-channel="operationChannel"/>

The Control Bus has an input channel that can be accessed for invoking operations on the beans in the application context. It also has all the common properties of a service activating endpoint, e.g. you can specify an output channel if the result of the operation has a return value that you want to send on to a downstream channel.

The Control Bus executes messages on the input channel as Spring Expression Language expressions. It takes a message, compiles the body to an expression, adds some context, and then executes it. The default context supports any method that has been annotated with @ManagedAttribute or @ManagedOperation. It also supports the methods on Spring’s Lifecycle interface, and it supports methods that are used to configure several of Spring’s TaskExecutor and TaskScheduler implementations. The simplest way to ensure that your own methods are available to the Control Bus is to use the @ManagedAttribute and/or @ManagedOperation annotations. Since those are also used for exposing methods to a JMX MBean registry, it’s a convenient by-product (often the same types of operations you want to expose to the Control Bus would be reasonable for exposing via JMX). Resolution of any particular instance within the application context is achieved in the typical SpEL syntax. Simply provide the bean name with the SpEL prefix for beans (@). For example, to execute a method on a Spring Bean a client could send a message to the operation channel as follows:

Message operation = MessageBuilder.withPayload("@myServiceBean.shutdown()").build();
operationChannel.send(operation)

The root of the context for the expression is the Message itself, so you also have access to the payload and headers as variables within your expression. This is consistent with all the other expression support in Spring Integration endpoints.

With Java and Annotations the Control Bus can be configured as follows:

@Bean
@ServiceActivator(inputChannel = "operationChannel")
public ExpressionControlBusFactoryBean controlBus() {
    return new ExpressionControlBusFactoryBean();
}

Or, when using Java DSL flow definitions:

@Bean
public IntegrationFlow controlBusFlow() {
    return IntegrationFlows.from("controlBus")
              .controlBus()
              .get();
}

Or, if you prefer Lambda style with automatic DirectChannel creation:

@Bean
public IntegrationFlow controlBus() {
    return IntegrationFlowDefinition::controlBus;
}

In this case, the channel is named controlBus.input.

=== Orderly Shutdown

As described in the section called “CompletableFuture”, the MBean exporter provides a JMX operation stopActiveComponents, which is used to stop the application in an orderly manner. The operation has a single long parameter. The parameter indicates how long (in milliseconds) the operation will wait to allow in-flight messages to complete. The operation works as follows:

The first step calls beforeShutdown() on all beans that implement OrderlyShutdownCapable. This allows such components to prepare for shutdown. Examples of components that implement this interface, and what they do with this call include: JMS and AMQP message-driven adapters stop their listener containers; TCP server connection factories stop accepting new connections (while keeping existing connections open); TCP inbound endpoints drop (log) any new messages received; http inbound endpoints return 503 - Service Unavailable for any new requests.

The second step stops any active channels, such as JMS- or AMQP-backed channels.

The third step stops all MessageSource s.

The fourth step stops all inbound MessageProducer s (that are not OrderlyShutdownCapable).

The fifth step waits for any remaining time left, as defined by the value of the long parameter passed in to the operation. This is intended to allow any in-flight messages to complete their journeys. It is therefore important to select an appropriate timeout when invoking this operation.

The sixth step calls afterShutdown() on all OrderlyShutdownCapable components. This allows such components to perform final shutdown tasks (closing all open sockets, for example).

As discussed in the section called “CompletableFuture” this operation can be invoked using JMX. If you wish to programmatically invoke the method, you will need to inject, or otherwise get a reference to, the IntegrationMBeanExporter. If no id attribute is provided on the <int-jmx:mbean-export/> definition, the bean will have a generated name. This name contains a random component to avoid ObjectName collisions if multiple Spring Integration contexts exist in the same JVM (MBeanServer).

For this reason, if you wish to invoke the method programmatically, it is recommended that you provide the exporter with an id attribute so it can easily be accessed in the application context.

Finally, the operation can be invoked using the <control-bus>; see the monitoring Spring Integration sample application for details.

	Important
	The above algorithm was improved in version 4.1. Previously, all task executors and schedulers were stopped. This could cause mid-flow messages in QueueChannel s to remain. Now, the shutdown leaves pollers running in order to allow these messages to be drained and processed.

=== Integration Graph

Starting with version 4.3, Spring Integration provides access to an application’s runtime object model which can, optionally, include component metrics. It is exposed as a graph, which may be used to visualize the current state of the integration application. The o.s.i.support.management.graph package contains all the required classes to collect, build and render the runtime state of Spring Integration components as a single tree-like Graph object. The IntegrationGraphServer should be declared as a bean to build, retrieve and refresh the Graph object. The resulting Graph object can be serialized to any format, although JSON is flexible and convenient to parse and represent on the client side. A simple Spring Integration application with only the default components would expose a graph as follows:

{
  "contentDescriptor": {
    "providerVersion": "4.3.0.RELEASE",
    "providerFormatVersion": 1.0,
    "provider": "spring-integration",
    "name": "myApplication"
  },
  "nodes": [
    {
      "nodeId": 1,
      "name": "nullChannel",
      "stats": null,
      "componentType": "channel"
    },
    {
      "nodeId": 2,
      "name": "errorChannel",
      "stats": null,
      "componentType": "publish-subscribe-channel"
    },
    {
      "nodeId": 3,
      "name": "_org.springframework.integration.errorLogger",
      "stats": {
        "duration": {
          "count": 0,
          "min": 0.0,
          "max": 0.0,
          "mean": 0.0,
          "standardDeviation": 0.0,
          "countLong": 0
        },
        "errorCount": 0,
        "standardDeviationDuration": 0.0,
        "countsEnabled": true,
        "statsEnabled": true,
        "loggingEnabled": false,
        "handleCount": 0,
        "meanDuration": 0.0,
        "maxDuration": 0.0,
        "minDuration": 0.0,
        "activeCount": 0
      },
      "componentType": "logging-channel-adapter",
      "output": null,
      "input": "errorChannel"
    }
  ],
  "links": [
    {
      "from": 2,
      "to": 3,
      "type": "input"
    }
  ]
}

As you can see, the graph consists of three top-level elements.

The contentDescriptor graph element is pretty straightforward and contains general information about the application providing the data. The name can be customized on the IntegrationGraphServer bean or via spring.application.name application context environment property. Other properties are provided by the framework and allows you to distinguish a similar model from other sources.

The links graph element represents connections between nodes from the nodes graph element and, therefore, between integration components in the source Spring Integration application. For example from a MessageChannel to an EventDrivenConsumer with some MessageHandler; or from an AbstractReplyProducingMessageHandler to a MessageChannel. For the convenience and to allow to determine a link purpose, the model is supplied with the type attribute. The possible types are:

The information from this element can be used by a visualizing tool to render connections between nodes from the nodes graph element, where the from and to numbers represent the value from the nodeId property of the linked nodes. For example the link type can be used to determine the proper port on the target node:

              +---(discard)
              |
         +----o----+
         |         |
         |         |
         |         |
(input)--o         o---(output)
         |         |
         |         |
         |         |
         +----o----+
              |
              +---(error)

The nodes graph element is perhaps the most interesting because its elements contain not only the runtime components with their componentType s and name s, but can also optionally contain metrics exposed by the component. Node elements contain various properties which are generally self-explanatory. For example, expression-based components include the expression property containing the primary expression string for the component. To enable the metrics, add an @EnableIntegrationManagement to some @Configuration class or add an <int:management/> element to your XML configuration. You can control exactly which components in the framework collect statistics. See the section called “CompletableFuture” for complete information. See the stats attribute from the _org.springframework.integration.errorLogger component in the JSON example above. The nullChannel and errorChannel don’t provide statistics information in this case, because the configuration for this example was:

@Configuration
@EnableIntegration
@EnableIntegrationManagement(statsEnabled = "_org.springframework.integration.errorLogger.handler",
      countsEnabled = "!*",
      defaultLoggingEnabled = "false")
public class ManagementConfiguration {

    @Bean
    public IntegrationGraphServer integrationGraphServer() {
        return new IntegrationGraphServer();
    }

}

The nodeId represents a unique incremental identifier to distinguish one component from another. It is also used in the links element to represent a relationship (connection) of this component to others, if any. The input and output attributes are for the inputChannel and outputChannel properties of the AbstractEndpoint, MessageHandler, SourcePollingChannelAdapter or MessageProducerSupport. See the next paragraph for more information.

==== Graph Runtime Model

Spring Integration components have various levels of complexity. For example, any polled MessageSource also has a SourcePollingChannelAdapter and a MessageChannel to which to send messages from the source data periodically. Other components might be middleware request-reply components, e.g. JmsOutboundGateway, with a consuming AbstractEndpoint to subscribe to (or poll) the requestChannel (input) for messages, and a replyChannel (output) to produce a reply message to send downstream. Meanwhile, any MessageProducerSupport implementation (e.g. ApplicationEventListeningMessageProducer) simply wraps some source protocol listening logic and sends messages to the outputChannel.

Within the graph, Spring Integration components are represented using the IntegrationNode class hierarchy, which you can find in the o.s.i.support.management.graph package. For example the ErrorCapableDiscardingMessageHandlerNode could be used for the AggregatingMessageHandler (because it has a discardChannel option) and can produce errors when consuming from a PollableChannel using a PollingConsumer. Another sample is CompositeMessageHandlerNode - for a MessageHandlerChain when subscribed to a SubscribableChannel, using an EventDrivenConsumer.

	Note
	The @MessagingGateway (see Section 8.4, “Messaging Gateways”) provides nodes for each its method, where the name attribute is based on the gateway’s bean name and the short method signature. For example the gateway:

@MessagingGateway(defaultRequestChannel = "four")
public interface Gate {

	void foo(String foo);

	void foo(Integer foo);

	void bar(String bar);

}

produces nodes like:

{
  "nodeId" : 10,
  "name" : "gate.bar(class java.lang.String)",
  "stats" : null,
  "componentType" : "gateway",
  "output" : "four",
  "errors" : null
},
{
  "nodeId" : 11,
  "name" : "gate.foo(class java.lang.String)",
  "stats" : null,
  "componentType" : "gateway",
  "output" : "four",
  "errors" : null
},
{
  "nodeId" : 12,
  "name" : "gate.foo(class java.lang.Integer)",
  "stats" : null,
  "componentType" : "gateway",
  "output" : "four",
  "errors" : null
}

This IntegrationNode hierarchy can be used for parsing the graph model on the client side, as well as for the understanding the general Spring Integration runtime behavior. See also Section 3.8, “Programming Tips and Tricks” for more information.

=== Integration Graph Controller

If your application is WEB-based (or built on top of Spring Boot using an embedded web container) and the Spring Integration HTTP or WebFlux module (see the section called “CompletableFuture” and the section called “CompletableFuture”) is present on the classpath, you can use a IntegrationGraphController to expose the IntegrationGraphServer functionality as a REST service. For this purpose, the @EnableIntegrationGraphController @Configuration class annotation and the <int-http:graph-controller/> XML element, are available in the HTTP module. Together with the @EnableWebMvc annotation (or <mvc:annotation-driven/> for xml definitions), this configuration registers an IntegrationGraphController @RestController where its @RequestMapping.path can be configured on the @EnableIntegrationGraphController annotation or <int-http:graph-controller/> element. The default path is /integration.

The IntegrationGraphController @RestController provides these services:

Any Security and Cross Origin restrictions for the IntegrationGraphController can be achieved with the standard configuration options and components provided by Spring Security and Spring MVC projects. A simple example of that follows:

<mvc:annotation-driven />

<mvc:cors>
	<mvc:mapping path="/myIntegration/**"
				 allowed-origins="http://localhost:9090"
				 allowed-methods="GET" />
</mvc:cors>

<security:http>
    <security:intercept-url pattern="/myIntegration/**" access="ROLE_ADMIN" />
</security:http>


<int-http:graph-controller path="/myIntegration" />

The Java & Annotation Configuration variant follows; note that, for convenience, the annotation provides an allowedOrigins attribute; this just provides GET access to the path. For more sophistication, you can configure the CORS mappings using standard Spring MVC mechanisms.

@Configuration
@EnableWebMvc // or @EnableWebFlux
@EnableWebSecurity // or @EnableWebFluxSecurity
@EnableIntegration
@EnableIntegrationGraphController(path = "/testIntegration", allowedOrigins="http://localhost:9090")
public class IntegrationConfiguration extends WebSecurityConfigurerAdapter {

    @Override
    protected void configure(HttpSecurity http) throws Exception {
	    http
            .authorizeRequests()
               .antMatchers("/testIntegration/**").hasRole("ADMIN")
            // ...
            .formLogin();
    }

    //...

}

= Integration Endpoints

This section covers the various Channel Adapters and Messaging Gateways provided by Spring Integration to support Message-based communication with external systems.

== Endpoint Quick Reference Table

As discussed in the sections above, Spring Integration provides a number of endpoints used to interface with external systems, file systems etc. The following is a summary of the various endpoints with quick links to the appropriate chapter.

To recap, Inbound Channel Adapters are used for one-way integration bringing data into the messaging application. Outbound Channel Adapters are used for one-way integration to send data out of the messaging application. Inbound Gateways are used for a bidirectional integration flow where some other system invokes the messaging application and receives a reply.Outbound Gateways are used for a bidirectional integration flow where the messaging application invokes some external service or entity, expecting a result.