8. Messaging Endpoints

8.1 Message Endpoints

The first part of this chapter covers some background theory and reveals quite a bit about the underlying API that drives Spring Integration’s various messaging components. This information can be helpful if you want to really understand what’s going on behind the scenes. However, if you want to get up and running with the simplified namespace-based configuration of the various elements, feel free to skip ahead toSection 8.1.4, “Namespace Support” for now.

As mentioned in the overview, Message Endpoints are responsible for connecting the various messaging components to channels. Over the next several chapters, you will see a number of different components that consume Messages. Some of these are also capable of sending reply Messages. Sending Messages is quite straightforward. As shown above in Section 4.1, “Message Channels”, it’s easy to send a Message to a Message Channel. However, receiving is a bit more complicated. The main reason is that there are two types of consumers: Polling Consumers and Event Driven Consumers.

Of the two, Event Driven Consumers are much simpler. Without any need to manage and schedule a separate poller thread, they are essentially just listeners with a callback method. When connecting to one of Spring Integration’s subscribable Message Channels, this simple option works great. However, when connecting to a buffering, pollable Message Channel, some component has to schedule and manage the polling thread(s). Spring Integration provides two different endpoint implementations to accommodate these two types of consumers. Therefore, the consumers themselves can simply implement the callback interface. When polling is required, the endpoint acts as a container for the consumer instance. The benefit is similar to that of using a container for hosting Message Driven Beans, but since these consumers are simply Spring-managed Objects running within an ApplicationContext, it more closely resembles Spring’s own MessageListener containers.

8.1.1 Message Handler

Spring Integration’s MessageHandler interface is implemented by many of the components within the framework. In other words, this is not part of the public API, and a developer would not typically implement MessageHandler directly. Nevertheless, it is used by a Message Consumer for actually handling the consumed Messages, and so being aware of this strategy interface does help in terms of understanding the overall role of a consumer. The interface is defined as follows:

public interface MessageHandler {

    void handleMessage(Message<?> message);

}

Despite its simplicity, this provides the foundation for most of the components that will be covered in the following chapters (Routers, Transformers, Splitters, Aggregators, Service Activators, etc). Those components each perform very different functionality with the Messages they handle, but the requirements for actually receiving a Message are the same, and the choice between polling and event-driven behavior is also the same. Spring Integration provides two endpoint implementations that host these callback-based handlers and allow them to be connected to Message Channels.

8.1.2 Event Driven Consumer

Because it is the simpler of the two, we will cover the Event Driven Consumer endpoint first. You may recall that the SubscribableChannel interface provides a subscribe() method and that the method accepts a MessageHandler parameter (as shown in the section called “SubscribableChannel”):

subscribableChannel.subscribe(messageHandler);

Since a handler that is subscribed to a channel does not have to actively poll that channel, this is an Event Driven Consumer, and the implementation provided by Spring Integration accepts a a SubscribableChannel and a MessageHandler:

SubscribableChannel channel = context.getBean("subscribableChannel", SubscribableChannel.class);

EventDrivenConsumer consumer = new EventDrivenConsumer(channel, exampleHandler);

8.1.3 Polling Consumer

Spring Integration also provides a PollingConsumer, and it can be instantiated in the same way except that the channel must implement PollableChannel:

PollableChannel channel = context.getBean("pollableChannel", PollableChannel.class);

PollingConsumer consumer = new PollingConsumer(channel, exampleHandler);
[Note]Note

For more information regarding Polling Consumers, please also read Section 4.2, “Poller” as well as Section 4.3, “Channel Adapter”.

There are many other configuration options for the Polling Consumer. For example, the trigger is a required property:

PollingConsumer consumer = new PollingConsumer(channel, handler);

consumer.setTrigger(new IntervalTrigger(30, TimeUnit.SECONDS));

Spring Integration currently provides two implementations of the Trigger interface: IntervalTrigger and CronTrigger. The IntervalTrigger is typically defined with a simple interval (in milliseconds), but also supports an initialDelay property and a boolean fixedRate property (the default is false, i.e. fixed delay):

IntervalTrigger trigger = new IntervalTrigger(1000);
trigger.setInitialDelay(5000);
trigger.setFixedRate(true);

The CronTrigger simply requires a valid cron expression (see the Javadoc for details):

CronTrigger trigger = new CronTrigger("*/10 * * * * MON-FRI");

In addition to the trigger, several other polling-related configuration properties may be specified:

PollingConsumer consumer = new PollingConsumer(channel, handler);

consumer.setMaxMessagesPerPoll(10);
consumer.setReceiveTimeout(5000);

The maxMessagesPerPoll property specifies the maximum number of messages to receive within a given poll operation. This means that the poller will continue calling receive() without waiting until either null is returned or that max is reached. For example, if a poller has a 10 second interval trigger and a maxMessagesPerPoll setting of 25, and it is polling a channel that has 100 messages in its queue, all 100 messages can be retrieved within 40 seconds. It grabs 25, waits 10 seconds, grabs the next 25, and so on.

The receiveTimeout property specifies the amount of time the poller should wait if no messages are available when it invokes the receive operation. For example, consider two options that seem similar on the surface but are actually quite different: the first has an interval trigger of 5 seconds and a receive timeout of 50 milliseconds while the second has an interval trigger of 50 milliseconds and a receive timeout of 5 seconds. The first one may receive a message up to 4950 milliseconds later than it arrived on the channel (if that message arrived immediately after one of its poll calls returned). On the other hand, the second configuration will never miss a message by more than 50 milliseconds. The difference is that the second option requires a thread to wait, but as a result it is able to respond much more quickly to arriving messages. This technique, known as long polling, can be used to emulate event-driven behavior on a polled source.

A Polling Consumer may also delegate to a Spring TaskExecutor, as illustrated in the following example:

PollingConsumer consumer = new PollingConsumer(channel, handler);

TaskExecutor taskExecutor = context.getBean("exampleExecutor", TaskExecutor.class);
consumer.setTaskExecutor(taskExecutor);

Furthermore, a PollingConsumer has a property called adviceChain. This property allows you to specify a List of AOP Advices for handling additional cross cutting concerns including transactions. These advices are applied around the doPoll() method. For more in-depth information, please see the sections AOP Advice chains and Transaction Support under Section 8.1.4, “Namespace Support”.

The examples above show dependency lookups, but keep in mind that these consumers will most often be configured as Spring bean definitions. In fact, Spring Integration also provides a FactoryBean called ConsumerEndpointFactoryBean that creates the appropriate consumer type based on the type of channel, and there is full XML namespace support to even further hide those details. The namespace-based configuration will be featured as each component type is introduced.

[Note]Note

Many of the MessageHandler implementations are also capable of generating reply Messages. As mentioned above, sending Messages is trivial when compared to the Message reception. Nevertheless,when and how many reply Messages are sent depends on the handler type. For example, an Aggregator waits for a number of Messages to arrive and is often configured as a downstream consumer for a Splitter which may generate multiple replies for each Message it handles. When using the namespace configuration, you do not strictly need to know all of the details, but it still might be worth knowing that several of these components share a common base class, the AbstractReplyProducingMessageHandler, and it provides a setOutputChannel(..) method.

8.1.4 Namespace Support

Throughout the reference manual, you will see specific configuration examples for endpoint elements, such as router, transformer, service-activator, and so on. Most of these will support an input-channel attribute and many will support an output-channel attribute. After being parsed, these endpoint elements produce an instance of either the PollingConsumer or the EventDrivenConsumer depending on the type of the input-channel that is referenced: PollableChannel or SubscribableChannel respectively. When the channel is pollable, then the polling behavior is determined based on the endpoint element’s poller sub-element and its attributes.

Configuration_Below you find a _poller with all available configuration options:

<int:poller cron=""                                  1
            default="false"                          2
            error-channel=""                         3
            fixed-delay=""                           4
            fixed-rate=""                            5
            id=""                                    6
            max-messages-per-poll=""                 7
            receive-timeout=""                       8
            ref=""                                   9
            task-executor=""                         10
            time-unit="MILLISECONDS"                 11
            trigger="">                              12
            <int:advice-chain />                     13
            <int:transactional />                    14
</int:poller>

1

Provides the ability to configure Pollers using Cron expressions. The underlying implementation uses an org.springframework.scheduling.support.CronTrigger. If this attribute is set, none of the following attributes must be specified: fixed-delay, trigger, fixed-rate, ref.

2

By setting this attribute to true, it is possible to define exactly one (1) global default poller. An exception is raised if more than one default poller is defined in the application context. Any endpoints connected to a PollableChannel (PollingConsumer) or any SourcePollingChannelAdapter that does not have any explicitly configured poller will then use the global default Poller. Optional. Defaults to false.

3

Identifies the channel which error messages will be sent to if a failure occurs in this poller’s invocation. To completely suppress Exceptions, provide a reference to the nullChannel. Optional.

4

The fixed delay trigger uses a PeriodicTrigger under the covers. If the time-unit attribute is not used, the specified value is represented in milliseconds. If this attribute is set, none of the following attributes must be specified: fixed-rate, trigger, cron, ref.

5

The fixed rate trigger uses a PeriodicTrigger under the covers. If the time-unit attribute is not used the specified value is represented in milliseconds. If this attribute is set, none of the following attributes must be specified: fixed-delay, trigger, cron, ref.

6

The Id referring to the Poller’s underlying bean-definition, which is of type org.springframework.integration.scheduling.PollerMetadata. The id attribute is required for a top-level poller element unless it is the default poller (default="true").

7

Please see Section 4.3.1, “Configuring An Inbound Channel Adapter” for more information. Optional. If not specified the default values used depends on the context. If a PollingConsumer is used, this atribute will default to -1. However, if a SourcePollingChannelAdapter is used, then the max-messages-per-poll attribute defaults to 1.

8

Value is set on the underlying class `PollerMetadata`Optional. If not specified it defaults to 1000 (milliseconds).

9

Bean reference to another top-level poller. The ref attribute must not be present on the top-level poller element. However, if this attribute is set, none of the following attributes must be specified: fixed-rate, trigger, cron, fixed-delay.

10

Provides the ability to reference a custom task executor. Please see the section below titled TaskExecutor Support for further information. Optional.

11

This attribute specifies the java.util.concurrent.TimeUnit enum value on the underlying org.springframework.scheduling.support.PeriodicTrigger. Therefore, this attribute can ONLY be used in combination with the fixed-delay or fixed-rate attributes. If combined with either cron or a trigger reference attribute, it will cause a failure. The minimal supported granularity for a PeriodicTrigger is MILLISECONDS. Therefore, the only available options are MILLISECONDS and SECONDS. If this value is not provided, then any fixed-delay or fixed-rate value will be interpreted as MILLISECONDS by default. Basically this enum provides a convenience for SECONDS-based interval trigger values. For hourly, daily, and monthly settings, consider using a cron trigger instead.

12

Reference to any spring configured bean which implements the org.springframework.scheduling.Trigger interface. Optional. However, if this attribute is set, none of the following attributes must be specified:fixed-delay, fixed-rate, cron, ref.

13

Allows to specify extra AOP Advices to handle additional cross cutting concerns. Please see the section below titled Transaction Support for further information. Optional.

14

Pollers can be made transactional. Please see the section below titled AOP Advice chains for further information. Optional.

Examples

For example, a simple interval-based poller with a 1-second interval would be configured like this:

<int:transformer input-channel="pollable"
    ref="transformer"
    output-channel="output">
    <int:poller fixed-rate="1000"/>
</int:transformer>

As an alternative to fixed-rate you can also use the fixed-delay attribute.

For a poller based on a Cron expression, use the cron attribute instead:

<int:transformer input-channel="pollable"
    ref="transformer"
    output-channel="output">
    <int:poller cron="*/10 * * * * MON-FRI"/>
</int:transformer>

If the input channel is a PollableChannel, then the poller configuration is required. Specifically, as mentioned above, the trigger is a required property of the PollingConsumer class. Therefore, if you omit the poller sub-element for a Polling Consumer endpoint’s configuration, an Exception may be thrown. The exception will also be thrown if you attempt to configure a poller on the element that is connected to a non-pollable channel.

It is also possible to create top-level pollers in which case only a ref is required:

<int:poller id="weekdayPoller" cron="*/10 * * * * MON-FRI"/>

<int:transformer input-channel="pollable"
    ref="transformer"
    output-channel="output">
    <int:poller ref="weekdayPoller"/>
</int:transformer>
[Note]Note

The ref attribute is only allowed on the inner-poller definitions. Defining this attribute on a top-level poller will result in a configuration exception thrown during initialization of the Application Context.

Global Default Pollers

In fact, to simplify the configuration even further, you can define a global default poller. A single top-level poller within an ApplicationContext may have the default attribute with a value of true. In that case, any endpoint with a PollableChannel for its input-channel that is defined within the same ApplicationContext and has no explicitly configured poller sub-element will use that default.

<int:poller id="defaultPoller" default="true" max-messages-per-poll="5" fixed-rate="3000"/>

<!-- No <poller/> sub-element is necessary since there is a default -->
<int:transformer input-channel="pollable"
                 ref="transformer"
                 output-channel="output"/>

Transaction Support

Spring Integration also provides transaction support for the pollers so that each receive-and-forward operation can be performed as an atomic unit-of-work. To configure transactions for a poller, simply add the_<transactional/>_ sub-element. The attributes for this element should be familiar to anyone who has experience with Spring’s Transaction management:

<int:poller fixed-delay="1000">
    <int:transactional transaction-manager="txManager"
                       propagation="REQUIRED"
                       isolation="REPEATABLE_READ"
                       timeout="10000"
                       read-only="false"/>
</int:poller>

For more information please refer to Section C.1.1, “Poller Transaction Support”.

AOP Advice chains

Since Spring transaction support depends on the Proxy mechanism  with TransactionInterceptor (AOP Advice) handling transactional behavior of the message flow initiated by the poller, some times there is a need to provide extra Advice(s) to handle other cross cutting behavior associated with the poller. For that poller defines an advice-chain element allowing you to add more advices - class that  implements MethodInterceptor interface…​

<int:service-activator id="advicedSa" input-channel="goodInputWithAdvice" ref="testBean"
		method="good" output-channel="output">
	<int:poller max-messages-per-poll="1" fixed-rate="10000">
		 <int:advice-chain>
			<ref bean="adviceA" />
			<beans:bean class="org.bar.SampleAdvice" />
			<ref bean="txAdvice" />
		</int:advice-chain>
	</int:poller>
</int:service-activator>

For more information on how to implement MethodInterceptor please refer to AOP sections of Spring reference manual (section 8 and 9). Advice chain can also be applied on the poller that does not have any transaction configuration essentially allowing you to enhance the behavior of the message flow initiated by the poller.

[Important]Important

When using an advice chain, the <transactional/> child element cannot be specified; instead, declare a <tx:advice/> bean and add it to the <advice-chain/>. See Section C.1.1, “Poller Transaction Support” for complete configuration.

TaskExecutor Support

The polling threads may be executed by any instance of Spring’s TaskExecutor abstraction. This enables concurrency for an endpoint or group of endpoints. As of Spring 3.0, there is a task namespace in the core Spring Framework, and its <executor/> element supports the creation of a simple thread pool executor. That element accepts attributes for common concurrency settings such as pool-size and queue-capacity. Configuring a thread-pooling executor can make a substantial difference in how the endpoint performs under load. These settings are available per-endpoint since the performance of an endpoint is one of the major factors to consider (the other major factor being the expected volume on the channel to which the endpoint subscribes). To enable concurrency for a polling endpoint that is configured with the XML namespace support, provide the task-executor reference on its <poller/> element and then provide one or more of the properties shown below:

<int:poller task-executor="pool" fixed-rate="1000"/>

<task:executor id="pool"
               pool-size="5-25"
               queue-capacity="20"
               keep-alive="120"/>

If no task-executor is provided, the consumer’s handler will be invoked in the caller’s thread. Note that the caller is usually the default TaskScheduler (see Section F.3, “Configuring the Task Scheduler”). Also, keep in mind that the task-executor attribute can provide a reference to any implementation of Spring’s TaskExecutor interface by specifying the bean name. The executor element above is simply provided for convenience.

As mentioned in the background section for Polling Consumers above, you can also configure a Polling Consumer in such a way as to emulate event-driven behavior. With a long receive-timeout and a short interval-trigger, you can ensure a very timely reaction to arriving messages even on a polled message source. Note that this will only apply to sources that have a blocking wait call with a timeout. For example, the File poller does not block, each receive() call returns immediately and either contains new files or not. Therefore, even if a poller contains a long receive-timeout, that value would never be usable in such a scenario. On the other hand when using Spring Integration’s own queue-based channels, the timeout value does have a chance to participate. The following example demonstrates how a Polling Consumer will receive Messages nearly instantaneously.

<int:service-activator input-channel="someQueueChannel"
    output-channel="output">
    <int:poller receive-timeout="30000" fixed-rate="10"/>

</int:service-activator>

Using this approach does not carry much overhead since internally it is nothing more then a timed-wait thread which does not require nearly as much CPU resource usage as a thrashing, infinite while loop for example.

8.1.5 Change Polling Rate at Runtime

When configuring Pollers with a fixed-delay or fixed-rate attribute, the default implementation will use a PeriodicTrigger instance. The PeriodicTrigger is part of the Core Spring Framework and it accepts the interval as a constructor argument, only. Therefore it cannot be changed at runtime.

However, you can define your own implementation of the org.springframework.scheduling.Trigger interface. You could even use the PeriodicTrigger as a starting point. Then, you can add a setter for the interval (period), or you could even embed your own throttling logic within the trigger itself if desired. The period property will be used with each call to nextExecutionTime to schedule the next poll. To use this custom trigger within pollers, declare the bean definition of the custom Trigger in your application context and inject the dependency into your Poller configuration using the trigger attribute, which references the custom Trigger bean instance. You can now obtain a reference to the Trigger bean and the polling interval can be changed between polls.

For an example, please see the Spring Integration Samples project. It contains a sample called dynamic-poller, which uses a custom Trigger and demonstrates the ability to change the polling interval at runtime.

https://github.com/SpringSource/spring-integration-samples/tree/master/intermediate

The sample provides a custom Trigger which implements the org.springframework.scheduling.Trigger interface. The sample’s Trigger is based on Spring’s PeriodicTrigger implementation. However, the fields of the custom trigger are not final and the properties have explicit getters and setters, allowing to dynamically change the polling period at runtime.

[Note]Note

It is important to note, though, that because the Trigger method is nextExecutionTime(), any changes to a dynamic trigger will not take effect until the next poll, based on the existing configuration. It is not possible to force a trigger to fire before it’s currently configured next execution time.

8.1.6 Payload Type Conversion

Throughout the reference manual, you will also see specific configuration and implementation examples of various endpoints which can accept a Message or any arbitrary Object as an input parameter. In the case of an Object, such a parameter will be mapped to a Message payload or part of the payload or header (when using the Spring Expression Language). However there are times when the type of input parameter of the endpoint method does not match the type of the payload or its part. In this scenario we need to perform type conversion. Spring Integration provides a convenient way for registering type converters (using the Spring 3.x ConversionService) within its own instance of a conversion service bean named integrationConversionService. That bean is automatically created as soon as the first converter is defined using the Spring Integration infrastructure. To register a Converter all you need is to implement org.springframework.core.convert.converter.Converter, org.springframework.core.convert.converter.GenericConverter or org.springframework.core.convert.converter.ConverterFactory.

The Converter implementation is the simplest and converts from a single type to another. For more sophistication, such as converting to a class hierarchy, you would implement a GenericConverter and possibly a ConditionalConverter. These give you complete access to the from and to type descriptors enabling complex conversions. For example, if you have an abstract class Foo that is the target of your conversion (parameter type, channel data type etc) and you have two concrete implementations Bar and Baz and you wish to convert to one or the other based on the input type, the GenericConverter would be a good fit. Refer to the JavaDocs for these interfaces for more information.

When you have implemented your converter, you can register it with convenient namespace support:

<int:converter ref="sampleConverter"/>

<bean id="sampleConverter" class="foo.bar.TestConverter"/>

or as an inner bean:

<int:converter>
    <bean class="o.s.i.config.xml.ConverterParserTests$TestConverter3"/>
</int:converter>

Starting with Spring Integration 4.0, the above configuration is available using annotations:

@Component
@IntegrationConverter
public class TestConverter implements Converter<Boolean, Number> {

	public Number convert(Boolean source) {
		return source ? 1 : 0;
	}

}

or as a @Configuration part:

@Configuration
@EnableIntegration
public class ContextConfiguration {

	@Bean
	@IntegrationConverter
	public SerializingConverter serializingConverter() {
		return new SerializingConverter();
	}

}
[Important]Important

When configuring an Application Context, the Spring Framework allows you to add a conversionService bean (see Configuring a ConversionService chapter). This service is used, when needed, to perform appropriate conversions during bean creation and configuration.

In contrast, the integrationConversionService is used for runtime conversions. These uses are quite different; converters that are intended for use when wiring bean constructor-args and properties may produce unintended results if used at runtime for Spring Integration expression evaluation against Messages within Datatype Channels, Payload Type transformers etc.

However, if you do want to use the Spring conversionService as the Spring Integration integrationConversionService, you can configure an alias in the Application Context:

<alias name="conversionService" alias="integrationConversionService"/>

In this case the conversionService's Converters will be available for Spring Integration runtime conversion.

8.1.7 Asynchronous polling

If you want the polling to be asynchronous, a Poller can optionally specify a task-executor attribute pointing to an existing instance of any TaskExecutor bean (Spring 3.0 provides a convenient namespace configuration via the task namespace). However, there are certain things you must understand when configuring a Poller with a TaskExecutor. 

The problem is that there are two configurations in place. The Poller and the TaskExecutor, and they both have to be in tune with each other otherwise you might end up creating an artificial memory leak.

Let’s look at the following configuration provided by one of the users on the Spring Integration Forum:

<int:channel id="publishChannel">
    <int:queue />
</int:channel>

<int:service-activator input-channel="publishChannel" ref="myService">
	<int:poller receive-timeout="5000" task-executor="taskExecutor" fixed-rate="50" />
</int:service-activator>

<task:executor id="taskExecutor" pool-size="20" />

The above configuration demonstrates one of those out of tune configurations.

By default, the task executor has an unbounded task queue. The poller keeps scheduling new tasks even though all the threads are blocked waiting for either a new message to arrive, or the timeout to expire. Given that there are 20 threads executing tasks with a 5 second timeout, they will be executed at a rate of 4 per second (5000/20 = 250ms). But, new tasks are being scheduled at a rate of 20 per second, so the internal queue in the task executor will grow at a rate of 16 per second (while the process is idle), so we essentially have a memory leak.

One of the ways to handle this is to set the queue-capacity attribute of the Task Executor; and even 0 is a reasonable value. You can also manage it by specifying what to do with messages that can not be queued by setting the rejection-policy attribute of the Task Executor (e.g., DISCARD). In other words, there are certain details you must understand with regard to configuring the TaskExecutor. Please refer to Task Execution and Scheduling of the Spring reference manual for more detail on the subject.

8.1.8 Endpoint Inner Beans

Many endpoints are composite beans; this includes all consumers and all polled inbound channel adapters. Consumers (polled or event- driven) delegate to a MessageHandler; polled adapters obtain messages by delegating to a MessageSource. Often, it is useful to obtain a reference to the delegate bean, perhaps to change configuration at runtime, or for testing. These beans can be obtained from the ApplicationContext with well-known names. MessageHandler s are registered with the application context with a bean id someConsumer.handler (where consumer is the endpoint’s id attribute). MessageSource s are registered with a bean id somePolledAdapter.source, again where somePolledAdapter is the id of the adapter.

The above only applies to the framework component itself. If you use an inner bean definition such as this:

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

the bean is treated like any inner bean declared that way and is not registered with the application context. If you wish to access this bean in some other manner, declare it at the top level with an id and use the ref attribute instead. See the Spring Documentation for more information.

8.2 Endpoint Roles

Starting with version 4.2, endpoints can be assigned to roles. Roles allow endpoints to be started and stopped as a group; this is particularly useful when using leadership election where a set of endpoints can be started or stopped when leadership is granted or revoked respectively.

You can assign endpoints to roles using XML, Java configuration, or programmatically:

<int:inbound-channel-adapter id="ica" channel="someChannel" expression="'foo'" role="cluster">
    <int:poller fixed-rate="60000" />
</int:inbound-channel-adapter>
@Bean
@ServiceActivator(inputChannel = "sendAsyncChannel")
@Role("cluster")
public MessageHandler sendAsyncHandler() {
    return // some MessageHandler
}
@Payload("#args[0].toLowerCase()")
@Role("cluster")
public String handle(String payload) {
    return payload.toUpperCase();
}
@Autowired
private SmartLifecycleRoleController roleController;

...

    this.roleController.addSmartLifeCycleToRole("cluster", someEndpoint);
...

Each of these adds the endpoint to the role cluster.

Invoking roleController.startLifecyclesInRole("cluster") (and the corresponding stop... method) will start/stop the endpoints.

[Note]Note

Any object implementing SmartLifecycle can be programmatically added, not just endpoints.

The SmartLifecycleRoleController implements ApplicationListener<AbstractLeaderEvent> and it will automatically start/stop its configured SmartLifecycle objects when leadership is granted/revoked (when some bean publishes OnGrantedEvent or OnRevokedEvent respectively).

[Important]Important

When using leadership election to start/stop components, it is important to set the auto-startup XML attribute (autoStartup bean property) to false so the application context does not start the components during context intialization.

Starting with _version 4.3.8, the SmartLifecycleRoleController provides several status methods:

public Collection<String> getRoles() 1

public boolean allEndpointsRunning(String role) 2

public boolean noEndpointsRunning(String role) 3

public Map<String, Boolean> getEndpointsRunningStatus(String role) 4

1

Returns a list of the roles being managed.

2

Returns true if all endpoints in the role are running.

3

Returns true if none of the endpoints in the role are running.

4

Returns a map of component name : running status - the component name is usually the bean name.

8.3 Leadership Event Handling

Groups of endpoints can be started/stopped based on leadership being granted or revoked respectively. This is useful in clustered scenarios where shared resources must only be consumed by a single instance. An example of this is a file inbound channel adapter that is polling a shared directory. (See Section 14.2, “Reading Files”).

To participate in a leader election and be notified when elected leader or when leadership is revoked, an application creates a component in the application context called a "leader initiator". Normally a leader initiator is a SmartLifecycle so it starts up (optionally) automatically when the context starts, and then publishes notifications when leadership changes. By convention the user provides a Candidate that receives the callbacks and also can revoke the leadership through a Context object provided by the framework. User code can also listen for AbstractLeaderEvents, and respond accordingly, for instance using a SmartLifecycleRoleController.

There is a basic implementation of a leader initiator based on the LockRegistry abstraction. To use it you just need to create an instance as a bean, for example:

@Bean
public LockRegistryLeaderInitiator leaderInitiator(LockRegistry locks) {
    return new LockRegistryLeaderInitiator(locks);
}

If the lock registry is implemented correctly, there will only ever be at most one leader. If the lock registry also provides locks which throw exceptions (ideally InterruptedException) when they expire or are broken, then the duration of the leaderless periods can be as short as is allowed by the inherent latency in the lock implementation. By default there is a busyWaitMillis property that adds some additional latency to prevent CPU starvation in the (more usual) case that the locks are imperfect and you only know they expired by trying to obtain one again.

See Section 37.4, “Zookeeper Leadership Event Handling” for more information about leadership election and events using Zookeeper.

8.4 Messaging Gateways

The primary purpose of a Gateway is to hide the messaging API provided by Spring Integration. It allows your application’s business logic to be completely unaware of the Spring Integration API and using a generic Gateway, your code interacts instead with a simple interface, only.

8.4.1 Enter the GatewayProxyFactoryBean

As mentioned above, it would be great to have no dependency on the Spring Integration API at all - including the gateway class. For that reason, Spring Integration provides the GatewayProxyFactoryBean that generates a proxy for any interface and internally invokes the gateway methods shown below. Using dependency injection you can then expose the interface to your business methods.

Here is an example of an interface that can be used to interact with Spring Integration:

package org.cafeteria;

public interface Cafe {

    void placeOrder(Order order);

}

8.4.2 Gateway XML Namespace Support

Namespace support is also provided which allows you to configure such an interface as a service as demonstrated by the following example.

<int:gateway id="cafeService"
         service-interface="org.cafeteria.Cafe"
         default-request-channel="requestChannel"
         default-reply-timeout="10000"
         default-reply-channel="replyChannel"/>

With this configuration defined, the "cafeService" can now be injected into other beans, and the code that invokes the methods on that proxied instance of the Cafe interface has no awareness of the Spring Integration API. The general approach is similar to that of Spring Remoting (RMI, HttpInvoker, etc.). See the "Samples" Appendix for an example that uses this "gateway" element (in the Cafe demo).

The defaults in the configuration above are applied to all methods on the gateway interface; if a reply timeout is not specified, the calling thread will wait indefinitely for a reply. See Section 8.4.10, “Gateway behavior when no response arrives”.

The defaults can be overridden for individual methods; see Section 8.4.4, “Gateway Configuration with Annotations and/or XML”.

8.4.3 Setting the Default Reply Channel

Typically you don’t have to specify the default-reply-channel, since a Gateway will auto-create a temporary, anonymous reply channel, where it will listen for the reply. However, there are some cases which may prompt you to define a default-reply-channel (or reply-channel with adapter gateways such as HTTP, JMS, etc.).

For some background, we’ll quickly discuss some of the inner-workings of the Gateway. A Gateway will create a temporary point-to-point reply channel which is anonymous and is added to the Message Headers with the name replyChannel. When providing an explicit default-reply-channel (reply-channel with remote adapter gateways), you have the option to point to a publish-subscribe channel, which is so named because you can add more than one subscriber to it. Internally Spring Integration will create a Bridge between the temporary replyChannel and the explicitly defined default-reply-channel.

So let’s say you want your reply to go not only to the gateway, but also to some other consumer. In this case you would want two things: a) a named channel you can subscribe to and b) that channel is a publish-subscribe-channel. The default strategy used by the gateway will not satisfy those needs, because the reply channel added to the header is anonymous and point-to-point. This means that no other subscriber can get a handle to it and even if it could, the channel has point-to-point behavior such that only one subscriber would get the Message. So by defining a default-reply-channel you can point to a channel of your choosing, which in this case would be a publish-subscribe-channel. The Gateway would create a bridge from it to the temporary, anonymous reply channel that is stored in the header.

Another case where you might want to provide a reply channel explicitly is for monitoring or auditing via an interceptor (e.g., wiretap). You need a named channel in order to configure a Channel Interceptor.

8.4.4 Gateway Configuration with Annotations and/or XML

public interface Cafe {

    @Gateway(requestChannel="orders")
    void placeOrder(Order order);

}

You may alternatively provide such content in method sub-elements if you prefer XML configuration (see the next paragraph).

It is also possible to pass values to be interpreted as Message headers on the Message that is created and sent to the request channel by using the @Header annotation:

public interface FileWriter {

    @Gateway(requestChannel="filesOut")
    void write(byte[] content, @Header(FileHeaders.FILENAME) String filename);

}

If you prefer the XML approach of configuring Gateway methods, you can provide method sub-elements to the gateway configuration.

<int:gateway id="myGateway" service-interface="org.foo.bar.TestGateway"
      default-request-channel="inputC">
  <int:default-header name="calledMethod" expression="#gatewayMethod.name"/>
  <int:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/>
  <int:method name="echoUpperCase" request-channel="inputB"/>
  <int:method name="echoViaDefault"/>
</int:gateway>

You can also provide individual headers per method invocation via XML. This could be very useful if the headers you want to set are static in nature and you don’t want to embed them in the gateway’s method signature via @Header annotations. For example, in the Loan Broker example we want to influence how aggregation of the Loan quotes will be done based on what type of request was initiated (single quote or all quotes). Determining the type of the request by evaluating what gateway method was invoked, although possible, would violate the separation of concerns paradigm (the method is a java artifact),  but expressing your intention (meta information) via Message headers is natural in a Messaging architecture.

<int:gateway id="loanBrokerGateway"
         service-interface="org.springframework.integration.loanbroker.LoanBrokerGateway">
  <int:method name="getLoanQuote" request-channel="loanBrokerPreProcessingChannel">
    <int:header name="RESPONSE_TYPE" value="BEST"/>
  </int:method>
  <int:method name="getAllLoanQuotes" request-channel="loanBrokerPreProcessingChannel">
    <int:header name="RESPONSE_TYPE" value="ALL"/>
  </int:method>
</int:gateway>

In the above case you can clearly see how a different value will be set for the RESPONSE_TYPE header based on the gateway’s method.

Expressions and "Global" Headers

The <header/> element supports expression as an alternative to value. The SpEL expression is evaluated to determine the value of the header. There is no #root object but the following variables are available:

#args - an Object[] containing the method arguments

#gatewayMethod - the java.reflect.Method object representing the method in the service-interface that was invoked. A header containing this variable can be used later in the flow, for example, for routing. For example, if you wish to route on the simple method name, you might add a header, with expression #gatewayMethod.name.

[Note]Note

The java.reflect.Method is not serializable; a header with expression #gatewayMethod will be lost if you later serialize the message. So, you may wish to use #gatewayMethod.name or #gatewayMethod.toString() in those cases; the toString() method provides a String representation of the method, including parameter and return types.

[Note]Note

Prior to 3.0, the #method variable was available, representing the method name only. This is still available, but deprecated; use #gatewayMethod.name instead.

Since 3.0, <default-header/> s can be defined to add headers to all messages produced by the gateway, regardless of the method invoked. Specific headers defined for a method take precedence over default headers. Specific headers defined for a method here will override any @Header annotations in the service interface. However, default headers will NOT override any @Header annotations in the service interface.

The gateway now also supports a default-payload-expression which will be applied for all methods (unless overridden).

8.4.5 Mapping Method Arguments to a Message

Using the configuration techniques in the previous section allows control of how method arguments are mapped to message elements (payload and header(s)). When no explicit configuration is used, certain conventions are used to perform the mapping. In some cases, these conventions cannot determine which argument is the payload and which should be mapped to headers.

public String send1(Object foo, Map bar);

public String send2(Map foo, Map bar);

In the first case, the convention will map the first argument to the payload (as long as it is not a Map) and the contents of the second become headers.

In the second case (or the first when the argument for parameter foo is a Map), the framework cannot determine which argument should be the payload; mapping will fail. This can generally be resolved using a payload-expression, a @Payload annotation and/or a @Headers annotation.

Alternatively, and whenever the conventions break down, you can take the entire responsibility for mapping the method calls to messages. To do this, implement an`MethodArgsMessageMapper` and provide it to the <gateway/> using the mapper attribute. The mapper maps a MethodArgsHolder, which is a simple class wrapping the java.reflect.Method instance and an Object[] containing the arguments. When providing a custom mapper, the default-payload-expression attribute and <default-header/> elements are not allowed on the gateway; similarly, the payload-expression attribute and <header/> elements are not allowed on any <method/> elements.

Mapping Method Arguments

Here are examples showing how method arguments can be mapped to the message (and some examples of invalid configuration):

public interface MyGateway {

    void payloadAndHeaderMapWithoutAnnotations(String s, Map<String, Object> map);

    void payloadAndHeaderMapWithAnnotations(@Payload String s, @Headers Map<String, Object> map);

    void headerValuesAndPayloadWithAnnotations(@Header("k1") String x, @Payload String s, @Header("k2") String y);

    void mapOnly(Map<String, Object> map); // the payload is the map and no custom headers are added

    void twoMapsAndOneAnnotatedWithPayload(@Payload Map<String, Object> payload, Map<String, Object> headers);

    @Payload("#args[0] + #args[1] + '!'")
    void payloadAnnotationAtMethodLevel(String a, String b);

    @Payload("@someBean.exclaim(#args[0])")
    void payloadAnnotationAtMethodLevelUsingBeanResolver(String s);

    void payloadAnnotationWithExpression(@Payload("toUpperCase()") String s);

    void payloadAnnotationWithExpressionUsingBeanResolver(@Payload("@someBean.sum(#this)") String s); //  1

    // invalid
    void twoMapsWithoutAnnotations(Map<String, Object> m1, Map<String, Object> m2);

    // invalid
    void twoPayloads(@Payload String s1, @Payload String s2);

    // invalid
    void payloadAndHeaderAnnotationsOnSameParameter(@Payload @Header("x") String s);

    // invalid
    void payloadAndHeadersAnnotationsOnSameParameter(@Payload @Headers Map<String, Object> map);

}

1

Note that in this example, the SpEL variable #this refers to the argument - in this case, the value of 's'.

The XML equivalent looks a little different, since there is no #this context for the method argument, but expressions can refer to method arguments using the #args variable:

<int:gateway id="myGateway" service-interface="org.foo.bar.MyGateway">
  <int:method name="send1" payload-expression="#args[0] + 'bar'"/>
  <int:method name="send2" payload-expression="@someBean.sum(#args[0])"/>
  <int:method name="send3" payload-expression="#method"/>
  <int:method name="send4">
    <int:header name="foo" expression="#args[2].toUpperCase()"/>
  </int:method>
</int:gateway>

8.4.6 @MessagingGateway Annotation

Starting with version 4.0, gateway service interfaces can be marked with a @MessagingGateway annotation instead of requiring the definition of a <gateway /> xml element for configuration. The following compares the two approaches for configuring the same gateway:

<int:gateway id="myGateway" service-interface="org.foo.bar.TestGateway"
      default-request-channel="inputC">
  <int:default-header name="calledMethod" expression="#gatewayMethod.name"/>
  <int:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/>
  <int:method name="echoUpperCase" request-channel="inputB">
  		<int:header name="foo" value="bar"/>
  </int:method>
  <int:method name="echoViaDefault"/>
</int:gateway>
@MessagingGateway(name = "myGateway", defaultRequestChannel = "inputC",
		  defaultHeaders = @GatewayHeader(name = "calledMethod",
		                           expression="#gatewayMethod.name"))
public interface TestGateway {

   @Gateway(requestChannel = "inputA", replyTimeout = 2, requestTimeout = 200)
   String echo(String payload);

   @Gateway(requestChannel = "inputB", headers = @GatewayHeader(name = "foo", value="bar"))
   String echoUpperCase(String payload);

   String echoViaDefault(String payload);

}
[Important]Important

As with the XML version, Spring Integration creates the proxy implementation with its messaging infrastructure, when discovering these annotations during a component scan. To perform this scan and register the BeanDefinition in the application context, add the @IntegrationComponentScan annotation to a @Configuration class. The standard @ComponentScan infrastructure doesn’t deal with interfaces, therefore the custom @IntegrationComponentScan logic has been introduced to determine @MessagingGateway annotation on the interfaces and register GatewayProxyFactoryBean s for them. See also Section F.6, “Annotation Support”

[Note]Note

If you have no XML configuration, the @EnableIntegration annotation is required on at least one @Configuration class. See Section 3.5, “Configuration and @EnableIntegration” for more information.

8.4.7 Invoking No-Argument Methods

When invoking methods on a Gateway interface that do not have any arguments, the default behavior is to receive a Message from a PollableChannel.

At times however, you may want to trigger no-argument methods so that you can in fact interact with other components downstream that do not require user-provided parameters, e.g. triggering no-argument SQL calls or Stored Procedures.

In order to achieve send-and-receive semantics, you must provide a payload. In order to generate a payload, method parameters on the interface are not necessary. You can either use the @Payload annotation or the payload-expression attribute in XML on the method sub-element. Below please find a few examples of what the payloads could be:

  • a literal string
  • #gatewayMethod.name
  • new java.util.Date()
  • @someBean.someMethod()'s return value

Here is an example using the @Payload annotation:

public interface Cafe {

    @Payload("new java.util.Date()")
    List<Order> retrieveOpenOrders();

}

If a method has no argument and no return value, but does contain a payload expression, it will be treated as a send-only operation.

8.4.8 Error Handling

Of course, the Gateway invocation might result in errors. By default any error that has occurred downstream will be re-thrown as a MessagingException (RuntimeException) upon the Gateway’s method invocation. However there are times when you may want to simply log the error rather than propagating it, or you may want to treat an Exception as a valid reply, by mapping it to a Message that will conform to some "error message" contract that the caller understands. To accomplish this, the Gateway provides support for a Message Channel dedicated to the errors via the error-channel attribute. In the example below, you can see that a transformer is used to create a reply Message from the Exception.

<int:gateway id="sampleGateway"
    default-request-channel="gatewayChannel"
    service-interface="foo.bar.SimpleGateway"
    error-channel="exceptionTransformationChannel"/>

<int:transformer input-channel="exceptionTransformationChannel"
        ref="exceptionTransformer" method="createErrorResponse"/>

The exceptionTransformer could be a simple POJO that knows how to create the expected error response objects. That would then be the payload that is sent back to the caller. Obviously, you could do many more elaborate things in such an "error flow" if necessary. It might involve routers (including Spring Integration’s ErrorMessageExceptionTypeRouter), filters, and so on. Most of the time, a simple transformer should be sufficient, however.

Alternatively, you might want to only log the Exception (or send it somewhere asynchronously). If you provide a one-way flow, then nothing would be sent back to the caller. In the case that you want to completely suppress Exceptions, you can provide a reference to the global "nullChannel" (essentially a /dev/null approach). Finally, as mentioned above, if no "error-channel" is defined at all, then the Exceptions will propagate as usual.

[Important]Important

Exposing the messaging system via simple POJI Gateways obviously provides benefits, but "hiding" the reality of the underlying messaging system does come at a price so there are certain things you should consider. We want our Java method to return as quickly as possible and not hang for an indefinite amount of time while the caller is waiting on it to return (void, return value, or a thrown Exception). When regular methods are used as a proxies in front of the Messaging system, we have to take into account the potentially asynchronous nature of the underlying messaging. This means that there might be a chance that a Message that was initiated by a Gateway could be dropped by a Filter, thus never reaching a component that is responsible for producing a reply. Some Service Activator method might result in an Exception, thus providing no reply (as we don’t generate Null messages). So as you can see there are multiple scenarios where a reply message might not be coming. That is perfectly natural in messaging systems. However think about the implication on the gateway method. The Gateway’s method input arguments  were incorporated into a Message and sent downstream. The reply Message would be converted to a return value of the Gateway’s method. So you might want to ensure that for each Gateway call there will always be a reply Message. Otherwise, your Gateway method might never return and will hang indefinitely. One of the ways of handling this situation is via an Asynchronous Gateway (explained later in this section). Another way of handling it is to explicitly set the reply-timeout attribute. That way, the gateway will not hang any longer than the time specified by the reply-timeout and will return null if that timeout does elapse. Finally, you might want to consider setting downstream flags such as requires-reply on a service-activator or throw-exceptions-on-rejection on a filter. These options will be discussed in more detail in the final section of this chapter.

[Note]Note

If the downstream flow returns an ErrorMessage, its payload (a Throwable) is treated as a regular downstream error: if there is an error-channel configured, it will be sent there, to the error flow; otherwise the payload is thrown to the caller of gateway. Similarly, if the error flow on the error-channel returns an ErrorMessage its payload is thrown to the caller. The same applies to any message with a Throwable payload. This can be useful in async situations when when there is a need propagate an Exception directly to the caller. To achieve this you can either return an Exception as the reply from some service, or simply throw it. Generally, even with an async flow, the framework will take care of propagating an exception thrown by the downstream flow back to the gateway. The TCP Client-Server Multiplex sample demonstrates both techniques to return the exception to the caller. It emulates a Socket IO error to the waiting thread using an aggregator with group-timeout (see the section called “Aggregator and Group Timeout”) and MessagingTimeoutException reply on the discard flow.

8.4.9 Asynchronous Gateway

Introduction

As a pattern, the Messaging Gateway is a very nice way to hide messaging-specific code while still exposing the full capabilities of the messaging system. As you’ve seen, the GatewayProxyFactoryBean provides a convenient way to expose a Proxy over a service-interface thus giving you POJO-based access to a messaging system (based on objects in your own domain, or primitives/Strings, etc).  But when a gateway is exposed via simple POJO methods which return values it does imply that for each Request message (generated when the method is invoked) there must be a Reply message (generated when the method has returned). Since Messaging systems naturally are asynchronous you may not always be able to guarantee the contract where "for each request there will always be be a reply".  With Spring Integration 2.0 we introduced support for an Asynchronous Gateway which is a convenient way to initiate flows where you may not know if a reply is expected or how long will it take for replies to arrive.

A natural way to handle these types of scenarios in Java would be relying upon java.util.concurrent.Future instances, and that is exactly what Spring Integration uses to support an Asynchronous Gateway.

From the XML configuration, there is nothing different and you still define Asynchronous Gateway the same way as a regular Gateway.

<int:gateway id="mathService" 
     service-interface="org.springframework.integration.sample.gateway.futures.MathServiceGateway"
     default-request-channel="requestChannel"/>

However the Gateway Interface (service-interface) is a little different:

public interface MathServiceGateway {

  Future<Integer> multiplyByTwo(int i);

}

As you can see from the example above, the return type for the gateway method is a Future. When GatewayProxyFactoryBean sees that the return type of the gateway method is a Future, it immediately switches to the async mode by utilizing an AsyncTaskExecutor. That is all. The call to such a method always returns immediately with a Future instance. Then, you can interact with the Future at your own pace to get the result, cancel, etc. And, as with any other use of Future instances, calling get() may reveal a timeout, an execution exception, and so on.

MathServiceGateway mathService = ac.getBean("mathService", MathServiceGateway.class);
Future<Integer> result = mathService.multiplyByTwo(number);
// do something else here since the reply might take a moment
int finalResult =  result.get(1000, TimeUnit.SECONDS);

For a more detailed example, please refer to the async-gateway sample distributed within the Spring Integration samples.

ListenableFuture

Starting with version 4.1, async gateway methods can also return ListenableFuture (introduced in Spring Framework 4.0). These return types allow you to provide a callback which is invoked when the result is available (or an exception occurs). When the gateway detects this return type, and the task executor (see below) is an AsyncListenableTaskExecutor, the executor’s submitListenable() method is invoked.

ListenableFuture<String> result = this.asyncGateway.async("foo");
result.addCallback(new ListenableFutureCallback<String>() {

    @Override
    public void onSuccess(String result) {
        ...
    }

    @Override
    public void onFailure(Throwable t) {
        ...
    }
});

AsyncTaskExecutor

By default, the GatewayProxyFactoryBean uses org.springframework.core.task.SimpleAsyncTaskExecutor when submitting internal AsyncInvocationTask instances for any gateway method whose return type is Future. However the async-executor attribute in the <gateway/> element’s configuration allows you to provide a reference to any implementation of java.util.concurrent.Executor available within the Spring application context.

The (default) SimpleAsyncTaskExecutor supports both Future and ListenableFuture return types, returning FutureTask or ListenableFutureTask respectively. Also see the section called “CompletableFuture” below. Even though there is a default executor, it is often useful to provide an external one so that you can identify its threads in logs (when using XML, the thread name is based on the executor’s bean name):

@Bean
public AsyncTaskExecutor exec() {
    SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
    simpleAsyncTaskExecutor.setThreadNamePrefix("exec-");
    return simpleAsyncTaskExecutor;
}

@MessagingGateway(asyncExecutor = "exec")
public interface ExecGateway {

    @Gateway(requestChannel = "gatewayChannel")
    Future<?> doAsync(String foo);

}

If you wish to return a different Future implementation, you can provide a custom executor, or disable the executor altogether and return the Future in the reply message payload from the downstream flow. To disable the executor, simply set it to null in the GatewayProxyFactoryBean (setAsyncTaskExecutor(null)). When configuring the gateway with XML, use async-executor=""; when configuring using the @MessagingGateway annotation, use:

@MessagingGateway(asyncExecutor = AnnotationConstants.NULL)
public interface NoExecGateway {

    @Gateway(requestChannel = "gatewayChannel")
    Future<?> doAsync(String foo);

}
[Important]Important

If the return type is a specific concrete Future implementation or some other subinterface that is not supported by the configured executor, the flow will run on the caller’s thread and the flow must return the required type in the reply message payload.

CompletableFuture

Starting with version 4.2, gateway methods can now return CompletableFuture<?>. There are several modes of operation when returning this type:

When an async executor is provided and the return type is exactly CompletableFuture (not a subclass), the framework will run the task on the executor and immediately return a CompletableFuture to the caller. CompletableFuture.supplyAsync(Supplier<U> supplier, Executor executor) is used to create the future.

When the async executor is explicitly set to null and the return type is CompletableFuture or the return type is a subclass of CompletableFuture, the flow is invoked on the caller’s thread. In this scenario, it is expected that the downstream flow will return a CompletableFuture of the appropriate type.

Usage Scenarios

CompletableFuture<Invoice> order(Order order);
<int:gateway service-interface="foo.Service" default-request-channel="orders" />

In this scenario, the caller thread returns immediately with a CompletableFuture<Invoice> which will be completed when the downstream flow replies to the gateway (with an Invoice object).

CompletableFuture<Invoice> order(Order order);
<int:gateway service-interface="foo.Service" default-request-channel="orders"
    async-executor="" />

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.

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 Promise

Starting with version 4.1, the GatewayProxyFactoryBean allows the use of a Reactor with gateway interface methods, utilizing a Promise<?> return type. The internal AsyncInvocationTask is wrapped in a reactor.function.Supplier, using a default RingBufferDispatcher for the Promise consumption. Only methods with the Promise<?> return type are run on the reactor’s dispatcher.

A Promise 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]Important

The Promise 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 Promise is flushed or via Promise.await(). Alternatively, the Promise (being a Composable) might be a part of Reactor Stream<?>, when the flush() is related to the entire Stream. For example:

@MessagingGateway
public static interface TestGateway {

	@Gateway(requestChannel = "promiseChannel")
	Promise<Integer> multiply(Integer value);

	}

	    ...

	@ServiceActivator(inputChannel = "promiseChannel")
	public Integer multiply(Integer value) {
			return value * 2;
	}

		...

	Streams.defer(Arrays.asList("1", "2", "3", "4", "5"))
			.get()
			.map(Integer::parseInt)
			.mapMany(integer -> testGateway.multiply(integer))
			.collect()
			.consume(integers -> ...)
			.flush();

Another example is a simple callback scenario:

Promise<Invoice> promise = service.process(myOrder);

promise.consume(new Consumer<Invoice>() {
	@Override
	public void accept(Invoice invoice) {
		handleInvoice(invoice);
	}
})
.flush();

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

8.4.10 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]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]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 handoff 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]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.

8.5 Service Activator

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

8.5.2 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 assumes that "exampleHandler" either contains a single method annotated with the @ServiceActivator annotation or that it contains only one public method at all. 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 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. 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 Section F.6, “Annotation Support”

[Note]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]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]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.

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

8.6 Delayer

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

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

[Tip]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]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]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: Section F.3, “Configuring the Task Scheduler”. If you want to delegate to a different scheduler, you can provide a reference through the delayer element’s scheduler attribute:

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

<task:scheduler id="exampleTaskScheduler" pool-size="3"/>
[Tip]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]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.

8.6.3 Delayer and 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 via a Control Bus command:

Message<String> delayerReschedulingMessage =
    MessageBuilder.withPayload("@'delayer.handler'.reschedulePersistedMessages()").build();
    controlBusChannel.send(delayerReschedulingMessage);
[Note]Note

For more information regarding the Message Store, JMX and the Control Bus, please read Chapter 9, System Management.

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

8.7.1 Script configuration

Depending on the complexity of your integration requirements scripts may be provided inline as CDATA in XML configuration or as a reference to a Spring resource 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]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]Important

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

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

8.8.1 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 Section 8.7.1, “Script configuration”.

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

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

8.8.2 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]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"/>

8.9 Adding Behavior to Endpoints

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

8.9.2 Provided Advice Classes

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

  • RequestHandlerRetryAdvice
  • RequestHandlerCircuitBreakerAdvice
  • ExpressionEvaluatingRequestHandlerAdvice

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

}

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

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

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

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

8.9.7 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 Section 8.9.6, “Advising Filters”. 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");
    }
}

8.9.8 Ordering Advices within an Advice Chain

Advice classes are "around" advices and are applied in a nested fashion. The first advice is the outermost, 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.

8.9.9 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();

8.9.10 Idempotent Receiver Enterprise Integration Pattern

Starting with version 4.1, Spring Integration provides an implementation of the Idempotent Receiver Enterprise Integration Pattern. It is a functional pattern and the whole idempotency logic should be implemented in the application, however to simplify the decision-making, the IdempotentReceiverInterceptor component is provided. This is an AOP Advice, 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 (Section 9.5, “Metadata Store”). 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=""  1
        endpoint=""  2
        selector=""  3
        discard-channel=""  4
        metadata-store=""  5
        key-strategy=""  6
        key-expression=""  7
        value-strategy=""  8
        value-expression=""  9
        throw-exception-on-rejection="" />  10

1

The id of the IdempotentReceiverInterceptor bean. Optional.

2

Consumer Endpoint name(s) or pattern(s) to which this interceptor will be applied. Separate names (patterns) with commas (,) e.g. endpoint="aaa, bbb*, *ccc, *ddd*, eee*fff". Endpoint bean names matching these patterns are then used to retrieve the target endpoint’s MessageHandler bean (using its .handler suffix), and the IdempotentReceiverInterceptor will be applied to those beans. Required.

3

A MessageSelector bean reference. Mutually exclusive with metadata-store and key-strategy (key-expression). When selector is not provided, one of key-strategy or key-strategy-expression is required.

4

Identifies the channel to which to send a message when the IdempotentReceiverInterceptor doesn’t accept it. When omitted, duplicate messages are forwarded to the handler with a duplicateMessage header. Optional.

5

A ConcurrentMetadataStore reference. Used by the underlying MetadataStoreSelector. Mutually exclusive with selector. Optional. The default MetadataStoreSelector uses an internal SimpleMetadataStore which does not maintain state across application executions.

6

A MessageProcessor reference. Used by the underlying MetadataStoreSelector. Evaluates an idempotentKey from the request Message. Mutually exclusive with selector and key-expression. When a selector is not provided, one of key-strategy or key-strategy-expression is required.

7

A SpEL expression to populate an ExpressionEvaluatingMessageProcessor. Used by the underlying MetadataStoreSelector. Evaluates an idempotentKey using the request Message as the evaluation context root object. Mutually exclusive with selector and key-strategy. When a selector is not provided, one of key-strategy or key-strategy-expression is required.

8

A MessageProcessor reference. Used by the underlying MetadataStoreSelector. Evaluates a value for the idempotentKey from the request Message. Mutually exclusive with selector and value-expression. By default, the MetadataStoreSelector uses the timestamp message header as the Metadata value.

9

A SpEL expression to populate an ExpressionEvaluatingMessageProcessor. Used by the underlying MetadataStoreSelector. Evaluates a value for the idempotentKey using the request Message as the evaluation context root object. Mutually exclusive with selector and value-strategy. By default, the MetadataStoreSelector uses the timestamp message header as the Metadata value.

10

Throw an exception if the IdempotentReceiverInterceptor rejects the message defaults to false. It is applied regardless of whether or not a discard-channel is provided.

For Java configuration, 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() {
    ....
}
[Note]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 Section 8.9.5, “Handle Message Advice” for more information.

8.10 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="" 1
    level="INFO" 2
    expression="" 3
    log-full-message="false" 4
    logger-name="" /> 5

1

The channel connecting the logging adapter to an upstream component.

2

The logging level at which messages sent to this adapter will be logged. Default: INFO.

3

A SpEL expression representing exactly what part(s) of the message will be logged. Default: payload - just the payload will be logged. This attribute cannot be specified if log-full-message is specified.

4

When true, the entire message will be logged (including headers). Default: false - just the payload will be logged. This attribute cannot be specified if expression is specified.

5

Specifies the name of the logger (known as category in log4j) used for log messages created by this adapter. This enables setting the log name (in the logging subsystem) for individual adapters. By default, all adapters will log under the name org.springframework.integration.handler.LoggingHandler.

8.10.1 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
    public MessageChannel logInputChannel() {
        return new DirectChannel();
    }

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

    }

}