Messaging Endpoints
Messaging Endpoints
Message Endpoints
The first part of this chapter covers some background theory and reveals quite a bit about the underlying API that drives Spring Integration’s various messaging components. This information can be helpful if you want to really understand what goes on behind the scenes. However, if you want to get up and running with the simplified namespace-based configuration of the various elements, feel free to skip ahead to Endpoint Namespace Support for now.
As mentioned in the overview, message endpoints are responsible for connecting the various messaging components to channels. Over the next several chapters, we cover a number of different components that consume messages. Some of these are also capable of sending reply messages. Sending messages is quite straightforward. As shown earlier in Message Channels, you can send a message to a message channel. However, receiving is a bit more complicated. The main reason is that there are two types of consumers: polling consumers and event-driven consumers.
Of the two, event-driven consumers are much simpler.
Without any need to manage and schedule a separate poller thread, they are essentially listeners with a callback method.
When connecting to one of Spring Integration’s subscribable message channels, this simple option works great.
However, when connecting to a buffering, pollable message channel, some component has to schedule and manage the polling threads.
Spring Integration provides two different endpoint implementations to accommodate these two types of consumers.
Therefore, the consumers themselves need only implement the callback interface.
When polling is required, the endpoint acts as a container for the consumer instance.
The benefit is similar to that of using a container for hosting message-driven beans, but, since these consumers are Spring-managed objects running within an ApplicationContext, it more closely resembles Spring’s own MessageListener containers.
Message Handler
Spring Integration’s MessageHandler interface is implemented by many of the components within the framework.
In other words, this is not part of the public API, and you would not typically implement MessageHandler directly.
Nevertheless, it is used by a message consumer for actually handling the consumed messages, so being aware of this strategy interface does help in terms of understanding the overall role of a consumer.
The interface is defined as follows:
public interface MessageHandler {
void handleMessage(Message<?> message);
}
Despite its simplicity, this interface provides the foundation for most of the components (routers, transformers, splitters, aggregators, service activators, and others) covered in the following chapters. Those components each perform very different functionality with the messages they handle, but the requirements for actually receiving a message are the same, and the choice between polling and event-driven behavior is also the same. Spring Integration provides two endpoint implementations that host these callback-based handlers and let them be connected to message channels.
Event-driven Consumer
Because it is the simpler of the two, we cover the event-driven consumer endpoint first.
You may recall that the SubscribableChannel interface provides a subscribe() method and that the method accepts a MessageHandler parameter (as shown in SubscribableChannel).
The following listing shows the definition of the subscribe method:
subscribableChannel.subscribe(messageHandler);
Since a handler that is subscribed to a channel does not have to actively poll that channel, this is an event-driven consumer, and the implementation provided by Spring Integration accepts a SubscribableChannel and a MessageHandler, as the following example shows:
SubscribableChannel channel = context.getBean("subscribableChannel", SubscribableChannel.class);
EventDrivenConsumer consumer = new EventDrivenConsumer(channel, exampleHandler);
Polling Consumer
Spring Integration also provides a PollingConsumer, and it can be instantiated in the same way except that the channel must implement PollableChannel, as the following example shows:
PollableChannel channel = context.getBean("pollableChannel", PollableChannel.class);
PollingConsumer consumer = new PollingConsumer(channel, exampleHandler);
| For more information regarding polling consumers, see Poller and Channel Adapter. |
There are many other configuration options for the polling consumer. For example, the trigger is a required property. The following example shows how to set the trigger:
PollingConsumer consumer = new PollingConsumer(channel, handler);
consumer.setTrigger(new PeriodicTrigger(Duration.ofSeconds(30)));
The PeriodicTrigger is typically defined with a simple interval (Duration) but also supports an initialDelay property and a boolean fixedRate property (the default is false — that is, no fixed delay).
The following example sets both properties:
PeriodicTrigger trigger = new PeriodicTrigger(Duration.ofSeconds(1));
trigger.setInitialDelay(Duration.ofSeconds(5));
trigger.setFixedRate(true);
The result of the three settings in the preceding example is a trigger that waits five seconds and then triggers every second.
The CronTrigger requires a valid cron expression.
See the Javadoc for details.
The following example sets a new CronTrigger:
CronTrigger trigger = new CronTrigger("*/10 * * * * MON-FRI");
The result of the trigger defined in the previous example is a trigger that triggers every ten seconds, Monday through Friday.
In addition to the trigger, you can specify two other polling-related configuration properties: maxMessagesPerPoll and receiveTimeout.
The following example shows how to set these two properties:
PollingConsumer consumer = new PollingConsumer(channel, handler);
consumer.setMaxMessagesPerPoll(10);
consumer.setReceiveTimeout(5000);
The maxMessagesPerPoll property specifies the maximum number of messages to receive within a given poll operation.
This means that the poller continues calling receive() without waiting, until either null is returned or the maximum value is reached.
For example, if a poller has a ten-second interval trigger and a maxMessagesPerPoll setting of 25, and it is polling a channel that has 100 messages in its queue, all 100 messages can be retrieved within 40 seconds.
It grabs 25, waits ten seconds, grabs the next 25, and so on.
If maxMessagesPerPoll is configured with a negative value, then MessageSource.receive() is called within a single polling cycle until it returns null.
Starting with version 5.5, a 0 value has a special meaning - skip the MessageSource.receive() call altogether, which may be considered as pausing for this polling endpoint until the maxMessagesPerPoll is changed to a n non-zero value at a later time, e.g. via a Control Bus.
The receiveTimeout property specifies the amount of time the poller should wait if no messages are available when it invokes the receive operation.
For example, consider two options that seem similar on the surface but are actually quite different: The first has an interval trigger of 5 seconds and a receive timeout of 50 milliseconds, while the second has an interval trigger of 50 milliseconds and a receive timeout of 5 seconds.
The first one may receive a message up to 4950 milliseconds later than it arrived on the channel (if that message arrived immediately after one of its poll calls returned).
On the other hand, the second configuration never misses a message by more than 50 milliseconds.
The difference is that the second option requires a thread to wait.
However, as a result, it can respond much more quickly to arriving messages.
This technique, known as “long polling”, can be used to emulate event-driven behavior on a polled source.
A polling consumer can also delegate to a Spring TaskExecutor, as the following example shows:
PollingConsumer consumer = new PollingConsumer(channel, handler);
TaskExecutor taskExecutor = context.getBean("exampleExecutor", TaskExecutor.class);
consumer.setTaskExecutor(taskExecutor);
Furthermore, a PollingConsumer has a property called adviceChain.
This property lets you to specify a List of AOP advices for handling additional cross-cutting concerns including transactions.
These advices are applied around the doPoll() method.
For more in-depth information, see the sections on AOP advice chains and transaction support under Endpoint Namespace Support.
The earlier examples show dependency lookups.
However, keep in mind that these consumers are most often configured as Spring bean definitions.
In fact, Spring Integration also provides a FactoryBean called ConsumerEndpointFactoryBean that creates the appropriate consumer type based on the type of channel.
Also, Spring Integration has full XML namespace support to even further hide those details.
The namespace-based configuration is in this guide featured as each component type is introduced.
Many of the MessageHandler implementations can generate reply messages.
As mentioned earlier, sending messages is trivial when compared to receiving messages.
Nevertheless, when and how many reply messages are sent depends on the handler type.
For example, an aggregator waits for a number of messages to arrive and is often configured as a downstream consumer for a splitter, which can generate multiple replies for each message it handles.
When using the namespace configuration, you do not strictly need to know all of the details.
However, it still might be worth knowing that several of these components share a common base class, the AbstractReplyProducingMessageHandler, and that it provides a setOutputChannel(..) method.
|
Endpoint Namespace Support
Throughout this reference manual, you can find specific configuration examples for endpoint elements, such as router, transformer, service-activator, and so on.
Most of these support an input-channel attribute and many support an output-channel attribute.
After being parsed, these endpoint elements produce an instance of either the PollingConsumer or the EventDrivenConsumer, depending on the type of the input-channel that is referenced: PollableChannel or SubscribableChannel, respectively.
When the channel is pollable, the polling behavior is based on the endpoint element’s poller sub-element and its attributes.
The following listing lists all of the available configuration options for a poller:
<int:poller cron="" (1)
default="false" (2)
error-channel="" (3)
fixed-delay="" (4)
fixed-rate="" (5)
id="" (6)
max-messages-per-poll="" (7)
receive-timeout="" (8)
ref="" (9)
task-executor="" (10)
time-unit="MILLISECONDS" (11)
trigger=""> (12)
<int:advice-chain /> (13)
<int:transactional /> (14)
</int:poller>| 1 | Provides the ability to configure pollers by using Cron expressions.
The underlying implementation uses an org.springframework.scheduling.support.CronTrigger.
If this attribute is set, none of the following attributes must be specified: fixed-delay, trigger, fixed-rate, and ref. |
| 2 | By setting this attribute to true, you can define exactly one global default poller.
An exception is raised if more than one default poller is defined in the application context.
Any endpoints connected to a PollableChannel (PollingConsumer) or any SourcePollingChannelAdapter that does not have an explicitly configured poller then uses the global default poller.
It defaults to false.
Optional. |
| 3 | Identifies the channel to which error messages are sent if a failure occurs in this poller’s invocation.
To completely suppress exceptions, you can provide a reference to the nullChannel.
Optional. |
| 4 | The fixed delay trigger uses a PeriodicTrigger under the covers.
If you do not use the time-unit attribute, the specified value is represented in milliseconds.
If this attribute is set, none of the following attributes must be specified: fixed-rate, trigger, cron, and ref. |
| 5 | The fixed rate trigger uses a PeriodicTrigger under the covers.
If you do not use the time-unit attribute, the specified value is represented in milliseconds.
If this attribute is set, none of the following attributes must be specified: fixed-delay, trigger, cron, and ref. |
| 6 | The ID referring to the poller’s underlying bean-definition, which is of type org.springframework.integration.scheduling.PollerMetadata.
The id attribute is required for a top-level poller element, unless it is the default poller (default="true"). |
| 7 | See Configuring An Inbound Channel Adapter for more information.
If not specified, the default value depends on the context.
If you use a PollingConsumer, this attribute defaults to -1.
However, if you use a SourcePollingChannelAdapter, the max-messages-per-poll attribute defaults to 1.
Optional. |
| 8 | Value is set on the underlying class PollerMetadata.
If not specified, it defaults to 1000 (milliseconds).
Optional. |
| 9 | Bean reference to another top-level poller.
The ref attribute must not be present on the top-level poller element.
However, if this attribute is set, none of the following attributes must be specified: fixed-rate, trigger, cron, and fixed-delay. |
| 10 | Provides the ability to reference a custom task executor. See TaskExecutor Support for further information. Optional. |
| 11 | This attribute specifies the java.util.concurrent.TimeUnit enum value on the underlying org.springframework.scheduling.support.PeriodicTrigger.
Therefore, this attribute can be used only in combination with the fixed-delay or fixed-rate attributes.
If combined with either cron or a trigger reference attribute, it causes a failure.
The minimal supported granularity for a PeriodicTrigger is milliseconds.
Therefore, the only available options are milliseconds and seconds.
If this value is not provided, any fixed-delay or fixed-rate value is interpreted as milliseconds.
Basically, this enum provides a convenience for seconds-based interval trigger values.
For hourly, daily, and monthly settings, we recommend using a cron trigger instead. |
| 12 | Reference to any Spring-configured bean that implements the org.springframework.scheduling.Trigger interface.
However, if this attribute is set, none of the following attributes must be specified: fixed-delay, fixed-rate, cron, and ref.
Optional. |
| 13 | Allows specifying extra AOP advices to handle additional cross-cutting concerns. See Transaction Support for further information. Optional. |
| 14 | Pollers can be made transactional. See AOP Advice chains for further information. Optional. |
Examples
A simple interval-based poller with a 1-second interval can be configured as follows:
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output">
<int:poller fixed-rate="1000"/>
</int:transformer>As an alternative to using the fixed-rate attribute, you can also use the fixed-delay attribute.
For a poller based on a Cron expression, use the cron attribute instead, as the following example shows:
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output">
<int:poller cron="*/10 * * * * MON-FRI"/>
</int:transformer>If the input channel is a PollableChannel, the poller configuration is required.
Specifically, as mentioned earlier, the trigger is a required property of the PollingConsumer class.
Therefore, if you omit the poller sub-element for a polling consumer endpoint’s configuration, an exception may be thrown.
The exception may also be thrown if you attempt to configure a poller on the element that is connected to a non-pollable channel.
It is also possible to create top-level pollers, in which case only a ref attribute is required, as the following example shows:
<int:poller id="weekdayPoller" cron="*/10 * * * * MON-FRI"/>
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output">
<int:poller ref="weekdayPoller"/>
</int:transformer>
The ref attribute is allowed only on the inner poller definitions.
Defining this attribute on a top-level poller results in a configuration exception being thrown during initialization of the application context.
|
Global Default Poller
To simplify the configuration even further, you can define a global default poller.
A single top-level poller component in XML DSL may have the default attribute set to true.
For Java configuration a PollerMetadata bean with the PollerMetadata.DEFAULT_POLLER name must be declared in this case.
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 uses that default.
The following example shows such a poller and a transformer that uses it:
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata defaultPoller() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(5);
pollerMetadata.setTrigger(new PeriodicTrigger(3000));
return pollerMetadata;
}
// No 'poller' attribute because there is a default global poller
@Bean
public IntegrationFlow transformFlow(MyTransformer transformer) {
return IntegrationFlow.from(MessageChannels.queue("pollable"))
.transform(transformer) // No 'poller' attribute because there is a default global poller
.channel("output")
.get();
}
@Bean(PollerMetadata.DEFAULT_POLLER)
public PollerMetadata defaultPoller() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(5);
pollerMetadata.setTrigger(new PeriodicTrigger(3000));
return pollerMetadata;
}
@Bean
public QueueChannel pollable() {
return new QueueChannel();
}
// No 'poller' attribute because there is a default global poller
@Transformer(inputChannel = "pollable", outputChannel = "output")
public Object transform(Object payload) {
...
}
@Bean(PollerMetadata.DEFAULT_POLLER)
fun defaultPoller() =
PollerMetadata()
.also {
it.maxMessagesPerPoll = 5
it.trigger = PeriodicTrigger(3000)
}
@Bean
fun convertFlow() =
integrationFlow(MessageChannels.queue("pollable")) {
transform(transformer) // No 'poller' attribute because there is a default global poller
channel("output")
}
<int:poller id="defaultPoller" default="true" max-messages-per-poll="5" fixed-delay="3000"/>
<!-- No <poller/> sub-element is necessary, because there is a default -->
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output"/>Transaction Support
Spring Integration also provides transaction support for the pollers so that each receive-and-forward operation can be performed as an atomic unit of work.
To configure transactions for a poller, add the <transactional/> sub-element.
The following example shows the available attributes:
<int:poller fixed-delay="1000">
<int:transactional transaction-manager="txManager"
propagation="REQUIRED"
isolation="REPEATABLE_READ"
timeout="10000"
read-only="false"/>
</int:poller>For more information, see Poller Transaction Support.
AOP Advice chains
Since Spring transaction support depends on the proxy mechanism with TransactionInterceptor (AOP Advice) handling transactional behavior of the message flow initiated by the poller, you must sometimes provide extra advices to handle other cross cutting behavior associated with the poller.
For that, the poller defines an advice-chain element that lets you add more advices in a class that implements the MethodInterceptor interface.
The following example shows how to define an advice-chain for a poller:
<int:service-activator id="advicedSa" input-channel="goodInputWithAdvice" ref="testBean"
method="good" output-channel="output">
<int:poller max-messages-per-poll="1" fixed-rate="10000">
<int:advice-chain>
<ref bean="adviceA" />
<beans:bean class="org.something.SampleAdvice" />
<ref bean="txAdvice" />
</int:advice-chain>
</int:poller>
</int:service-activator>For more information on how to implement the MethodInterceptor interface, see the AOP sections of the Spring Framework Reference Guide.
An advice chain can also be applied on a poller that does not have any transaction configuration, letting you enhance the behavior of the message flow initiated by the poller.
When using an advice chain, the <transactional/> child element cannot be specified.
Instead, declare a <tx:advice/> bean and add it to the <advice-chain/>.
See Poller Transaction Support for complete configuration details.
|
TaskExecutor Support
The polling threads may be executed by any instance of Spring’s TaskExecutor abstraction.
This enables concurrency for an endpoint or group of endpoints.
As of Spring 3.0, the core Spring Framework has a task namespace, and its <executor/> element supports the creation of a simple thread pool executor.
That element accepts attributes for common concurrency settings, such as pool-size and queue-capacity.
Configuring a thread-pooling executor can make a substantial difference in how the endpoint performs under load.
These settings are available for each endpoint, since the performance of an endpoint is one of the major factors to consider (the other major factor being the expected volume on the channel to which the endpoint subscribes).
To enable concurrency for a polling endpoint that is configured with the XML namespace support, provide the task-executor reference on its <poller/> element and then provide one or more of the properties shown in the following example:
<int:poller task-executor="pool" fixed-rate="1000"/>
<task:executor id="pool"
pool-size="5-25"
queue-capacity="20"
keep-alive="120"/>If you do not provide a task-executor, the consumer’s handler is invoked in the caller’s thread.
Note that the caller is usually the default TaskScheduler (see Configuring the Task Scheduler).
You should also keep in mind that the task-executor attribute can provide a reference to any implementation of Spring’s TaskExecutor interface by specifying the bean name.
The executor element shown earlier is provided for convenience.
As mentioned earlier in the background section for polling consumers, you can also configure a polling consumer in such a way as to emulate event-driven behavior.
With a long receive timeout and a short interval in the trigger, you can ensure a very timely reaction to arriving messages even on a polled message source.
Note that this applies only to sources that have a blocking wait call with a timeout.
For example, the file poller does not block.
Each receive() call returns immediately and either contains new files or not.
Therefore, even if a poller contains a long receive-timeout, that value would never be used in such a scenario.
On the other hand, when using Spring Integration’s own queue-based channels, the timeout value does have a chance to participate.
The following example shows how a polling consumer can receive messages nearly instantaneously:
<int:service-activator input-channel="someQueueChannel"
output-channel="output">
<int:poller receive-timeout="30000" fixed-rate="10"/>
</int:service-activator>Using this approach does not carry much overhead, since, internally, it is nothing more then a timed-wait thread, which does not require nearly as much CPU resource usage as (for example) a thrashing, infinite while loop.
Changing Polling Rate at Runtime
When configuring a poller with a fixed-delay or a fixed-rate attribute, the default implementation uses a PeriodicTrigger instance.
The PeriodicTrigger is part of the core Spring Framework.
It accepts the interval only as a constructor argument.
Therefore, it cannot be changed at runtime.
However, you can define your own implementation of the org.springframework.scheduling.Trigger interface.
You could even use the PeriodicTrigger as a starting point.
Then you can add a setter for the interval (period), or you can even embed your own throttling logic within the trigger itself.
The period property is used with each call to nextExecutionTime to schedule the next poll.
To use this custom trigger within pollers, declare the bean definition of the custom trigger in your application context and inject the dependency into your poller configuration by using the trigger attribute, which references the custom trigger bean instance.
You can now obtain a reference to the trigger bean and change the polling interval between polls.
For an example, see the Spring Integration Samples project.
It contains a sample called dynamic-poller, which uses a custom trigger and demonstrates the ability to change the polling interval at runtime.
The sample provides a custom trigger that implements the org.springframework.scheduling.Trigger interface.
The sample’s trigger is based on Spring’s PeriodicTrigger implementation.
However, the fields of the custom trigger are not final, and the properties have explicit getters and setters, letting you dynamically change the polling period at runtime.
It is important to note, though, that because the Trigger method is nextExecutionTime(), any changes to a dynamic trigger do not take effect until the next poll, based on the existing configuration.
It is not possible to force a trigger to fire before its currently configured next execution time.
|
Payload Type Conversion
Throughout this reference manual, you can also see specific configuration and implementation examples of various endpoints that accept a message or any arbitrary Object as an input parameter.
In the case of an Object, such a parameter is mapped to a message payload or part of the payload or header (when using the Spring Expression Language).
However, the type of input parameter of the endpoint method sometimes does not match the type of the payload or its part.
In this scenario, we need to perform type conversion.
Spring Integration provides a convenient way for registering type converters (by using the Spring ConversionService) within its own instance of a conversion service bean named integrationConversionService.
That bean is automatically created as soon as the first converter is defined by using the Spring Integration infrastructure.
To register a converter, you can implement org.springframework.core.convert.converter.Converter, org.springframework.core.convert.converter.GenericConverter, or org.springframework.core.convert.converter.ConverterFactory.
The Converter implementation is the simplest and converts from a single type to another.
For more sophistication, such as converting to a class hierarchy, you can implement a GenericConverter and possibly a ConditionalConverter.
These give you complete access to the from and to type descriptors, enabling complex conversions.
For example, if you have an abstract class called Something that is the target of your conversion (parameter type, channel data type, and so on), you have two concrete implementations called Thing1 and Thing, and you wish to convert to one or the other based on the input type, the GenericConverter would be a good fit.
For more information, see the Javadoc for these interfaces:
When you have implemented your converter, you can register it with convenient namespace support, as the following example shows:
<int:converter ref="sampleConverter"/>
<bean id="sampleConverter" class="foo.bar.TestConverter"/>Alternately, you can use an inner bean, as the following example shows:
<int:converter>
<bean class="o.s.i.config.xml.ConverterParserTests$TestConverter3"/>
</int:converter>Starting with Spring Integration 4.0, you can use annotations to create the preceding configuration, as the following example shows:
@Component
@IntegrationConverter
public class TestConverter implements Converter<Boolean, Number> {
public Number convert(Boolean source) {
return source ? 1 : 0;
}
}
Alternately, you can use the @Configuration annotation, as the following example shows:
@Configuration
@EnableIntegration
public class ContextConfiguration {
@Bean
@IntegrationConverter
public SerializingConverter serializingConverter() {
return new SerializingConverter();
}
}
|
When configuring an application context, the Spring Framework lets you add a In contrast, the However, if you do want to use the Spring In this case, the converters provided by the |
Content Type Conversion
Starting with version 5.0, by default, the method invocation mechanism is based on the org.springframework.messaging.handler.invocation.InvocableHandlerMethod infrastructure.
Its HandlerMethodArgumentResolver implementations (such as PayloadArgumentResolver and MessageMethodArgumentResolver) can use the MessageConverter abstraction to convert an incoming payload to the target method argument type.
The conversion can be based on the contentType message header.
For this purpose, Spring Integration provides the ConfigurableCompositeMessageConverter, which delegates to a list of registered converters to be invoked until one of them returns a non-null result.
By default, this converter provides (in strict order):
-
MappingJackson2MessageConverterif the Jackson processor is present on the classpath
See the Javadoc (linked in the preceding list) for more information about their purpose and appropriate contentType values for conversion.
The ConfigurableCompositeMessageConverter is used because it can be supplied with any other MessageConverter implementations, including or excluding the previously mentioned default converters.
It can also be registered as an appropriate bean in the application context, overriding the default converter, as the following example shows:
@Bean(name = IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)
public ConfigurableCompositeMessageConverter compositeMessageConverter() {
List<MessageConverter> converters =
Arrays.asList(new MarshallingMessageConverter(jaxb2Marshaller()),
new JavaSerializationMessageConverter());
return new ConfigurableCompositeMessageConverter(converters);
}
Those two new converters are registered in the composite before the defaults.
You can also not use a ConfigurableCompositeMessageConverter but provide your own MessageConverter by registering a bean with the name, integrationArgumentResolverMessageConverter (by setting the IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME property).
The MessageConverter-based (including contentType header) conversion is not available when using SpEL method invocation.
In this case, only the regular class-to-class conversion mentioned above in the Payload Type Conversion is available.
|
Asynchronous Polling
If you want the polling to be asynchronous, a poller can optionally specify a task-executor attribute that points to an existing instance of any TaskExecutor bean (Spring 3.0 provides a convenient namespace configuration through the task namespace).
However, there are certain things you must understand when configuring a poller with a TaskExecutor.
The problem is that there are two configurations in place, the poller and the TaskExecutor.
They must be in tune with each other.
Otherwise, you might end up creating an artificial memory leak.
Consider the following configuration:
<int:channel id="publishChannel">
<int:queue />
</int:channel>
<int:service-activator input-channel="publishChannel" ref="myService">
<int:poller receive-timeout="5000" task-executor="taskExecutor" fixed-rate="50" />
</int:service-activator>
<task:executor id="taskExecutor" pool-size="20" />The preceding configuration demonstrates an out-of-tune configuration.
By default, the task executor has an unbounded task queue. The poller keeps scheduling new tasks even though all the threads are blocked, waiting for either a new message to arrive or the timeout to expire. Given that there are 20 threads executing tasks with a five-second timeout, they are executed at a rate of 4 per second. However, new tasks are being scheduled at a rate of 20 per second, so the internal queue in the task executor grows at a rate of 16 per second (while the process is idle), so we have a memory leak.
One of the ways to handle this is to set the queue-capacity attribute of the task executor.
Even 0 is a reasonable value.
You can also manage it by specifying what to do with messages that can not be queued by setting the rejection-policy attribute of the Task Executor (for example, to DISCARD).
In other words, there are certain details you must understand when configuring TaskExecutor.
See “Task Execution and Scheduling” in the Spring reference manual for more detail on the subject.
Endpoint Inner Beans
Many endpoints are composite beans.
This includes all consumers and all polled inbound channel adapters.
Consumers (polled or event-driven) delegate to a MessageHandler.
Polled adapters obtain messages by delegating to a MessageSource.
Often, it is useful to obtain a reference to the delegate bean, perhaps to change configuration at runtime or for testing.
These beans can be obtained from the ApplicationContext with well known names.
MessageHandler instances are registered with the application context with bean IDs similar to someConsumer.handler (where 'consumer' is the value of the endpoint’s id attribute).
MessageSource instances are registered with bean IDs similar to somePolledAdapter.source, where 'somePolledAdapter' is the ID of the adapter.
The preceding only applies to the framework component itself. You can instead use an inner bean definition, as the following example shows:
<int:service-activator id="exampleServiceActivator" input-channel="inChannel"
output-channel = "outChannel" method="foo">
<beans:bean class="org.foo.ExampleServiceActivator"/>
</int:service-activator>The bean is treated like any inner bean declared and is not registered with the application context.
If you wish to access this bean in some other manner, declare it at the top level with an id and use the ref attribute instead.
See the Spring Documentation for more information.
Endpoint Roles
Starting with version 4.2, endpoints can be assigned to roles.
Roles let endpoints be started and stopped as a group.
This is particularly useful when using leadership election, where a set of endpoints can be started or stopped when leadership is granted or revoked, respectively.
For this purpose the framework registers a SmartLifecycleRoleController bean in the application context with the name IntegrationContextUtils.INTEGRATION_LIFECYCLE_ROLE_CONTROLLER.
Whenever it is necessary to control lifecycles, this bean can be injected or @Autowired:
<bean class="com.some.project.SomeLifecycleControl">
<property name="roleController" ref="integrationLifecycleRoleController"/>
</bean>You can assign endpoints to roles using XML, Java configuration, or programmatically. The following example shows how to configure endpoint roles with XML:
<int:inbound-channel-adapter id="ica" channel="someChannel" expression="'foo'" role="cluster"
auto-startup="false">
<int:poller fixed-rate="60000" />
</int:inbound-channel-adapter>The following example shows how to configure endpoint roles for a bean created in Java:
@Bean
@ServiceActivator(inputChannel = "sendAsyncChannel", autoStartup="false")
@Role("cluster")
public MessageHandler sendAsyncHandler() {
return // some MessageHandler
}
The following example shows how to configure endpoint roles on a method in Java:
@Payload("#args[0].toLowerCase()")
@Role("cluster")
public String handle(String payload) {
return payload.toUpperCase();
}
The following example shows how to configure endpoint roles by using the SmartLifecycleRoleController in Java:
@Autowired
private SmartLifecycleRoleController roleController;
...
this.roleController.addSmartLifeCycleToRole("cluster", someEndpoint);
...
The following example shows how to configure endpoint roles by using an IntegrationFlow in Java:
IntegrationFlow flow -> flow
.handle(..., e -> e.role("cluster"));
Each of these adds the endpoint to the cluster role.
Invoking roleController.startLifecyclesInRole("cluster") and the corresponding stop… method starts and stops the endpoints.
Any object that implements SmartLifecycle can be programmatically added — not just endpoints.
|
The SmartLifecycleRoleController implements ApplicationListener<AbstractLeaderEvent> and it automatically starts and stops its configured SmartLifecycle objects when leadership is granted or revoked (when some bean publishes OnGrantedEvent or OnRevokedEvent, respectively).
When using leadership election to start and stop components, it is important to set the auto-startup XML attribute (autoStartup bean property) to false so that the application context does not start the components during context initialization.
|
Starting with version 4.3.8, the SmartLifecycleRoleController provides several status methods:
public Collection<String> getRoles() (1)
public boolean allEndpointsRunning(String role) (2)
public boolean noEndpointsRunning(String role) (3)
public Map<String, Boolean> getEndpointsRunningStatus(String role) (4)
| 1 | Returns a list of the roles being managed. |
| 2 | Returns true if all endpoints in the role are running. |
| 3 | Returns true if none of the endpoints in the role are running. |
| 4 | Returns a map of component name : running status.
The component name is usually the bean name. |
Leadership Event Handling
Groups of endpoints can be started and stopped based on leadership being granted or revoked, respectively. This is useful in clustered scenarios where shared resources must be consumed by only a single instance. An example of this is a file inbound channel adapter that is polling a shared directory. (See Reading Files).
To participate in a leader election and be notified when elected leader, when leadership is revoked, or on failure to acquire the resources to become leader, an application creates a component in the application context called a “leader initiator”.
Normally, a leader initiator is a SmartLifecycle, so it starts (optionally) when the context starts and then publishes notifications when leadership changes.
You can also receive failure notifications by setting the publishFailedEvents to true (starting with version 5.0), for cases when you want to take a specific action if a failure occurs.
By convention, you should provide a Candidate that receives the callbacks.
You can also revoke the leadership through a Context object provided by the framework.
Your code can also listen for o.s.i.leader.event.AbstractLeaderEvent instances (the super class of OnGrantedEvent and OnRevokedEvent) and respond accordingly (for instance, by using a SmartLifecycleRoleController).
The events contain a reference to the Context object.
The following listing shows the definition of the Context interface:
public interface Context {
boolean isLeader();
void yield();
String getRole();
}
Starting with version 5.0.6, the context provides a reference to the candidate’s role.
Spring Integration provides a basic implementation of a leader initiator that is based on the LockRegistry abstraction.
To use it, you need to create an instance as a bean, as the following example shows:
@Bean
public LockRegistryLeaderInitiator leaderInitiator(LockRegistry locks) {
return new LockRegistryLeaderInitiator(locks);
}
If the lock registry is implemented correctly, there is only ever at most one leader.
If the lock registry also provides locks that throw exceptions (ideally, InterruptedException) when they expire or are broken, the duration of the leaderless periods can be as short as is allowed by the inherent latency in the lock implementation.
By default, the busyWaitMillis property adds some additional latency to prevent CPU starvation in the (more usual) case that the locks are imperfect, and you only know they expired when you try to obtain one again.
See Zookeeper Leadership Event Handling for more information about leadership election and events that use Zookeeper. See Hazelcast Leadership Event Handling for more information about leadership election and events that use Hazelcast.
Messaging Gateways
A gateway hides the messaging API provided by Spring Integration. It lets your application’s business logic be unaware of the Spring Integration API. By using a generic Gateway, your code interacts with only a simple interface.
Enter the GatewayProxyFactoryBean
As mentioned earlier, it would be great to have no dependency on the Spring Integration API — including the gateway class.
For that reason, Spring Integration provides the GatewayProxyFactoryBean, which generates a proxy for any interface and internally invokes the gateway methods shown below.
By using dependency injection, you can then expose the interface to your business methods.
The following example shows an interface that can be used to interact with Spring Integration:
package org.cafeteria;
public interface Cafe {
void placeOrder(Order order);
}
Gateway XML Namespace Support
Namespace support is also provided. It lets you configure an interface as a service, as the following example shows:
<int:gateway id="cafeService"
service-interface="org.cafeteria.Cafe"
default-request-channel="requestChannel"
default-reply-timeout="10000"
default-reply-channel="replyChannel"/>With this configuration defined, the cafeService can now be injected into other beans, and the code that invokes the methods on that proxied instance of the Cafe interface has no awareness of the Spring Integration API.
See the “Samples” Appendix for an example that uses the gateway element (in the Cafe demo).
The defaults in the preceding configuration are applied to all methods on the gateway interface. If a reply timeout is not specified, the calling thread waits indefinitely for a reply. See Gateway Behavior When No response Arrives.
The defaults can be overridden for individual methods. See Gateway Configuration with Annotations and XML.
Setting the Default Reply Channel
Typically, you need not specify the default-reply-channel, since a Gateway auto-creates a temporary, anonymous reply channel, where it listens for the reply.
However, some cases may prompt you to define a default-reply-channel (or reply-channel with adapter gateways, such as HTTP, JMS, and others).
For some background, we briefly discuss some inner workings of the gateway.
A gateway creates a temporary point-to-point reply channel.
It is anonymous and is added to the message headers with the name, replyChannel.
When providing an explicit default-reply-channel (reply-channel with remote adapter gateways), you can point to a publish-subscribe channel, which is so named because you can add more than one subscriber to it.
Internally, Spring Integration creates a bridge between the temporary replyChannel and the explicitly defined default-reply-channel.
Suppose you want your reply to go not only to the gateway but also to some other consumer. In this case, you want two things:
-
A named channel to which you can subscribe
-
That channel to be a publish-subscribe-channel
The default strategy used by the gateway does not satisfy those needs, because the reply channel added to the header is anonymous and point-to-point.
This means that no other subscriber can get a handle to it and, even if it could, the channel has point-to-point behavior such that only one subscriber would get the message.
By defining a default-reply-channel you can point to a channel of your choosing.
In this case, that is a publish-subscribe-channel.
The gateway creates a bridge from it to the temporary, anonymous reply channel that is stored in the header.
You might also want to explicitly provide a reply channel for monitoring or auditing through an interceptor (for example, wiretap). To configure a channel interceptor, you need a named channel.
Starting with version 5.4, when gateway method return type is void, the framework populates a replyChannel header as a nullChannel bean reference if such a header is not provided explicitly.
This allows any possible reply from the downstream flow to be discarded, meeting the one-way gateway contract.
|
Gateway Configuration with Annotations and XML
Consider the following example, which expands on the previous Cafe interface example by adding a @Gateway annotation:
public interface Cafe {
@Gateway(requestChannel="orders")
void placeOrder(Order order);
}
The @Header annotation lets you add values that are interpreted as message headers, as the following example shows:
public interface FileWriter {
@Gateway(requestChannel="filesOut")
void write(byte[] content, @Header(FileHeaders.FILENAME) String filename);
}
If you prefer the XML approach to configuring gateway methods, you can add method elements to the gateway configuration, as the following example shows:
<int:gateway id="myGateway" service-interface="org.foo.bar.TestGateway"
default-request-channel="inputC">
<int:default-header name="calledMethod" expression="#gatewayMethod.name"/>
<int:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/>
<int:method name="echoUpperCase" request-channel="inputB"/>
<int:method name="echoViaDefault"/>
</int:gateway>You can also use XML to provide individual headers for each method invocation.
This could be useful if the headers you want to set are static in nature, and you do not want to embed them in the gateway’s method signature by using @Header annotations.
For example, in the loan broker example, we want to influence how aggregation of the loan quotes is done, based on what type of request was initiated (single quote or all quotes).
Determining the type of the request by evaluating which gateway method was invoked, although possible, would violate the separation of concerns paradigm (the method is a Java artifact).
However, expressing your intention (meta information) in message headers is natural in a messaging architecture.
The following example shows how to add a different message header for each of two methods:
<int:gateway id="loanBrokerGateway"
service-interface="org.springframework.integration.loanbroker.LoanBrokerGateway">
<int:method name="getLoanQuote" request-channel="loanBrokerPreProcessingChannel">
<int:header name="RESPONSE_TYPE" value="BEST"/>
</int:method>
<int:method name="getAllLoanQuotes" request-channel="loanBrokerPreProcessingChannel">
<int:header name="RESPONSE_TYPE" value="ALL"/>
</int:method>
</int:gateway>In the preceding example a different value is set for the 'RESPONSE_TYPE' header, based on the gateway’s method.
If you specify, for example, the requestChannel in <int:method/> as well as in a @Gateway annotation, the annotation value wins.
|
If a no-argument gateway is specified in XML, and the interface method has both a @Payload and @Gateway annotation (with a payloadExpression or a payload-expression in an <int:method/> element), the @Payload value is ignored.
|
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.
Starting with version 5.2, the #root object of the evaluation context is a MethodArgsHolder with getMethod() and getArgs() accessors.
For example, if you wish to route on the simple method name, you might add a header with the following expression: method.name.
The java.reflect.Method is not serializable.
A header with an expression of method is lost if you later serialize the message.
Consequently, you may wish to use method.name or method.toString() in those cases.
The toString() method provides a String representation of the method, including parameter and return types.
|
Since version 3.0, <default-header/> elements can be defined to add headers to all the messages produced by the gateway, regardless of the method invoked.
Specific headers defined for a method take precedence over default headers.
Specific headers defined for a method here override any @Header annotations in the service interface.
However, default headers do NOT override any @Header annotations in the service interface.
The gateway now also supports a default-payload-expression, which is applied for all methods (unless overridden).
Mapping Method Arguments to a Message
Using the configuration techniques in the previous section allows control of how method arguments are mapped to message elements (payload and headers). When no explicit configuration is used, certain conventions are used to perform the mapping. In some cases, these conventions cannot determine which argument is the payload and which should be mapped to headers. Consider the following example:
public String send1(Object thing1, Map thing2);
public String send2(Map thing1, Map thing2);
In the first case, the convention is to map the first argument to the payload (as long as it is not a Map) and the contents of the second argument become headers.
In the second case (or the first when the argument for parameter thing1 is a Map), the framework cannot determine which argument should be the payload.
Consequently, mapping fails.
This can generally be resolved using a payload-expression, a @Payload annotation, or a @Headers annotation.
Alternatively (and whenever the conventions break down), you can take the entire responsibility for mapping the method calls to messages.
To do so, implement an MethodArgsMessageMapper and provide it to the <gateway/> by using the mapper attribute.
The mapper maps a MethodArgsHolder, which is a simple class that wraps the java.reflect.Method instance and an Object[] containing the arguments.
When providing a custom mapper, the default-payload-expression attribute and <default-header/> elements are not allowed on the gateway.
Similarly, the payload-expression attribute and <header/> elements are not allowed on any <method/> elements.
Mapping Method Arguments
The following examples show how method arguments can be mapped to the message and shows some examples of invalid configuration:
public interface MyGateway {
void payloadAndHeaderMapWithoutAnnotations(String s, Map<String, Object> map);
void payloadAndHeaderMapWithAnnotations(@Payload String s, @Headers Map<String, Object> map);
void headerValuesAndPayloadWithAnnotations(@Header("k1") String x, @Payload String s, @Header("k2") String y);
void mapOnly(Map<String, Object> map); // the payload is the map and no custom headers are added
void twoMapsAndOneAnnotatedWithPayload(@Payload Map<String, Object> payload, Map<String, Object> headers);
@Payload("args[0] + args[1] + '!'")
void payloadAnnotationAtMethodLevel(String a, String b);
@Payload("@someBean.exclaim(args[0])")
void payloadAnnotationAtMethodLevelUsingBeanResolver(String s);
void payloadAnnotationWithExpression(@Payload("toUpperCase()") String s);
void payloadAnnotationWithExpressionUsingBeanResolver(@Payload("@someBean.sum(#this)") String s); // (1)
// invalid
void twoMapsWithoutAnnotations(Map<String, Object> m1, Map<String, Object> m2);
// invalid
void twoPayloads(@Payload String s1, @Payload String s2);
// invalid
void payloadAndHeaderAnnotationsOnSameParameter(@Payload @Header("x") String s);
// invalid
void payloadAndHeadersAnnotationsOnSameParameter(@Payload @Headers Map<String, Object> map);
}
| 1 | Note that, in this example, the SpEL variable, #this, refers to the argument — in this case, the value of s. |
The XML equivalent looks a little different, since there is no #this context for the method argument.
However, expressions can refer to method arguments by using the args property for the MethodArgsHolder root object (see Expressions and “Global” Headers for more information), as the following example shows:
<int:gateway id="myGateway" service-interface="org.something.MyGateway">
<int:method name="send1" payload-expression="args[0] + 'thing2'"/>
<int:method name="send2" payload-expression="@someBean.sum(args[0])"/>
<int:method name="send3" payload-expression="method"/>
<int:method name="send4">
<int:header name="thing1" expression="args[2].toUpperCase()"/>
</int:method>
</int:gateway>@MessagingGateway Annotation
Starting with version 4.0, gateway service interfaces can be marked with a @MessagingGateway annotation instead of requiring the definition of a <gateway /> xml element for configuration.
The following pair of examples compares the two approaches for configuring the same gateway:
<int:gateway id="myGateway" service-interface="org.something.TestGateway"
default-request-channel="inputC">
<int:default-header name="calledMethod" expression="#gatewayMethod.name"/>
<int:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/>
<int:method name="echoUpperCase" request-channel="inputB">
<int:header name="thing1" value="thing2"/>
</int:method>
<int:method name="echoViaDefault"/>
</int:gateway>@MessagingGateway(name = "myGateway", defaultRequestChannel = "inputC",
defaultHeaders = @GatewayHeader(name = "calledMethod",
expression="#gatewayMethod.name"))
public interface TestGateway {
@Gateway(requestChannel = "inputA", replyTimeout = 2, requestTimeout = 200)
String echo(String payload);
@Gateway(requestChannel = "inputB", headers = @GatewayHeader(name = "thing1", value="thing2"))
String echoUpperCase(String payload);
String echoViaDefault(String payload);
}
Similarly to the XML version, when Spring Integration discovers these annotations during a component scan, it creates the proxy implementation with its messaging infrastructure.
To perform this scan and register the BeanDefinition in the application context, add the @IntegrationComponentScan annotation to a @Configuration class.
The standard @ComponentScan infrastructure does not deal with interfaces.
Consequently, we introduced the custom @IntegrationComponentScan logic to find the @MessagingGateway annotation on the interfaces and register GatewayProxyFactoryBean instances for them.
See also Annotation Support.
|
Along with the @MessagingGateway annotation you can mark a service interface with the @Profile annotation to avoid the bean creation, if such a profile is not active.
Starting with version 6.0, an interface with the @MessagingGateway can also be marked with a @Primary annotation for respective configuration logic as its possible with any Spring @Component definition.
Starting with version 6.0, @MessagingGateway interfaces can be used in the standard Spring @Import configuration.
This may be used as an alternative to the @IntegrationComponentScan or manual AnnotationGatewayProxyFactoryBean bean definitions.
The @MessagingGateway is meta-annotated with a @MessageEndpoint since version 6.0 and the name() attribute is, essentially, aliased to the @Compnent.value().
This way the bean names generating strategy for gateway proxies is realigned with the standard Spring annotation configuration for scanned and imported components.
The default AnnotationBeanNameGenerator can be overridden globally via an AnnotationConfigUtils.CONFIGURATION_BEAN_NAME_GENERATOR or as a @IntegrationComponentScan.nameGenerator() attribute.
If you have no XML configuration, the @EnableIntegration annotation is required on at least one @Configuration class.
See Configuration and @EnableIntegration for more information.
|
Invoking No-Argument Methods
When invoking methods on a Gateway interface that do not have any arguments, the default behavior is to receive a Message from a PollableChannel.
Sometimes, however, you may want to trigger no-argument methods so that you can interact with other components downstream that do not require user-provided parameters, such as triggering no-argument SQL calls or stored procedures.
To achieve send-and-receive semantics, you must provide a payload.
To generate a payload, method parameters on the interface are not necessary.
You can either use the @Payload annotation or the payload-expression attribute in XML on the method element.
The following list includes a few examples of what the payloads could be:
-
a literal string
-
#gatewayMethod.name
-
new java.util.Date()
-
@someBean.someMethod()'s return value
The following example shows how to use the @Payload annotation:
public interface Cafe {
@Payload("new java.util.Date()")
List<Order> retrieveOpenOrders();
}You can also use the @Gateway annotation.
public interface Cafe {
@Gateway(payloadExpression = "new java.util.Date()")
List<Order> retrieveOpenOrders();
}
If both annotations are present (and the payloadExpression is provided), @Gateway wins.
|
If a method has no argument and no return value but does contain a payload expression, it is treated as a send-only operation.
Invoking default Methods
An interface for gateway proxy may have default methods as well and starting with version 5.3, the framework injects a DefaultMethodInvokingMethodInterceptor into a proxy for calling default methods using a java.lang.invoke.MethodHandle approach instead of proxying.
The interfaces from JDK, such as java.util.function.Function, still can be used for gateway proxy, but their default methods cannot be called because of internal Java security reasons for a MethodHandles.Lookup instantiation against JDK classes.
These methods also can be proxied (losing their implementation logic and, at the same time, restoring previous gateway proxy behavior) using an explicit @Gateway annotation on the method, or proxyDefaultMethods on the @MessagingGateway annotation or <gateway> XML component.
Error Handling
The gateway invocation can result in errors. By default, any error that occurs downstream is re-thrown “as is” upon the gateway’s method invocation. For example, consider the following simple flow:
gateway -> service-activatorIf the service invoked by the service activator throws a MyException (for example), the framework wraps it in a MessagingException and attaches the message passed to the service activator in the failedMessage property.
Consequently, any logging performed by the framework has full the context of the failure.
By default, when the exception is caught by the gateway, the MyException is unwrapped and thrown to the caller.
You can configure a throws clause on the gateway method declaration to match the particular exception type in the cause chain.
For example, if you want to catch a whole MessagingException with all the messaging information of the reason of downstream error, you should have a gateway method similar to the following:
public interface MyGateway {
void performProcess() throws MessagingException;
}
Since we encourage POJO programming, you may not want to expose the caller to messaging infrastructure.
If your gateway method does not have a throws clause, the gateway traverses the cause tree, looking for a RuntimeException that is not a MessagingException.
If none is found, the framework throws the MessagingException.
If the MyException in the preceding discussion has a cause of SomeOtherException and your method throws SomeOtherException, the gateway further unwraps that and throws it to the caller.
When a gateway is declared with no service-interface, an internal framework interface RequestReplyExchanger is used.
Consider the following example:
public interface RequestReplyExchanger {
Message<?> exchange(Message<?> request) throws MessagingException;
}
Before version 5.0, this exchange method did not have a throws clause and, as a result, the exception was unwrapped.
If you use this interface and want to restore the previous unwrap behavior, use a custom service-interface instead or access the cause of the MessagingException yourself.
However, you may want to log the error rather than propagating it, or you may want to treat an exception as a valid reply (by mapping it to a message that conforms to some "error message" contract that the caller understands).
To accomplish this, the gateway provides support for a message channel dedicated to the errors by including support for the error-channel attribute.
In the following example, a 'transformer' creates a reply Message from the Exception:
<int:gateway id="sampleGateway"
default-request-channel="gatewayChannel"
service-interface="foo.bar.SimpleGateway"
error-channel="exceptionTransformationChannel"/>
<int:transformer input-channel="exceptionTransformationChannel"
ref="exceptionTransformer" method="createErrorResponse"/>The exceptionTransformer could be a simple POJO that knows how to create the expected error response objects.
That becomes the payload that is sent back to the caller.
You could do many more elaborate things in such an “error flow”, if necessary.
It might involve routers (including Spring Integration’s ErrorMessageExceptionTypeRouter), filters, and so on.
Most of the time, a simple 'transformer' should be sufficient, however.
Alternatively, you might want to only log the exception (or send it somewhere asynchronously).
If you provide a one-way flow, nothing would be sent back to the caller.
If you want to completely suppress exceptions, you can provide a reference to the global nullChannel (essentially a /dev/null approach).
Finally, as mentioned above, if no error-channel is defined, then the exceptions propagate as usual.
When you use the @MessagingGateway annotation (see @MessagingGateway AnnotationerrorChannel attribute.
Starting with version 5.0, when you use a gateway method with a void return type (one-way flow), the error-channel reference (if provided) is populated in the standard errorChannel header of each sent message.
This feature allows a downstream asynchronous flow, based on the standard ExecutorChannel configuration (or a QueueChannel), to override a default global errorChannel exceptions sending behavior.
Previously you had to manually specify an errorChannel header with the @GatewayHeader annotation or the <header> element.
The error-channel property was ignored for void methods with an asynchronous flow.
Instead, error messages were sent to the default errorChannel.
Exposing the messaging system through simple POJI Gateways provides benefits, but “hiding” the reality of the underlying messaging system does come at a price, so there are certain things you should consider.
We want our Java method to return as quickly as possible and not hang for an indefinite amount of time while the caller is waiting on it to return (whether void, a return value, or a thrown Exception).
When regular methods are used as a proxies in front of the messaging system, we have to take into account the potentially asynchronous nature of the underlying messaging.
This means that there might be a chance that a message that was initiated by a gateway could be dropped by a filter and never reach a component that is responsible for producing a reply.
Some service activator method might result in an exception, thus providing no reply (as we do not generate null messages).
In other words, multiple scenarios can cause a reply message to never come.
That is perfectly natural in messaging systems.
However, think about the implication on the gateway method. The gateway’s method input arguments were incorporated into a message and sent downstream.
The reply message would be converted to a return value of the gateway’s method.
So you might want to ensure that, for each gateway call, there is always a reply message.
Otherwise, your gateway method might never return and hang indefinitely.
One way to handle this situation is by using an asynchronous gateway (explained later in this section).
Another way of handling it is to explicitly set the reply-timeout attribute.
That way, the gateway does not hang any longer than the time specified by the reply-timeout and returns 'null' if that timeout does elapse.
Finally, you might want to consider setting downstream flags, such as 'requires-reply', on a service-activator or 'throw-exceptions-on-rejection' on a filter. These options are discussed in more detail in the final section of this chapter.
|
If the downstream flow returns an ErrorMessage, its payload (a Throwable) is treated as a regular downstream error.
If there is an error-channel configured, it is sent to the error flow.
Otherwise, the payload is thrown to the caller of the gateway.
Similarly, if the error flow on the error-channel returns an ErrorMessage, its payload is thrown to the caller.
The same applies to any message with a Throwable payload.
This can be useful in asynchronous situations when you need to propagate an Exception directly to the caller.
To do so, you can either return an Exception (as the reply from some service) or throw it.
Generally, even with an asynchronous flow, the framework takes care of propagating an exception thrown by the downstream flow back to the gateway.
The TCP Client-Server Multiplex sample demonstrates both techniques to return the exception to the caller.
It emulates a socket IO error to the waiting thread by using an aggregator with group-timeout (see Aggregator and Group Timeout) and a MessagingTimeoutException reply on the discard flow.
|
Gateway Timeouts
Gateways have two timeout properties: requestTimeout and replyTimeout.
The request timeout applies only if the channel can block (for example, a bounded QueueChannel that is full).
The replyTimeout value is how long the gateway waits for a reply or returns null.
It defaults to infinity.
The timeouts can be set as defaults for all methods on the gateway (defaultRequestTimeout and defaultReplyTimeout) or on the MessagingGateway interface annotation.
Individual methods can override these defaults (in <method/> child elements) or on the @Gateway annotation.
Starting with version 5.0, the timeouts can be defined as expressions, as the following example shows:
@Gateway(payloadExpression = "args[0]", requestChannel = "someChannel",
requestTimeoutExpression = "args[1]", replyTimeoutExpression = "args[2]")
String lateReply(String payload, long requestTimeout, long replyTimeout);
The evaluation context has a BeanResolver (use @someBean to reference other beans), and the args array property from the #root object is available.
See Expressions and “Global” Headers for more information about this root object.
When configuring with XML, the timeout attributes can be a long value or a SpEL expression, as the following example shows:
<method name="someMethod" request-channel="someRequestChannel"
payload-expression="args[0]"
request-timeout="1000"
reply-timeout="args[1]">
</method>Asynchronous Gateway
As a pattern, the messaging gateway offers a nice way to hide messaging-specific code while still exposing the full capabilities of the messaging system.
As described earlier, the GatewayProxyFactoryBean provides a convenient way to expose a proxy over a service-interface giving you POJO-based access to a messaging system (based on objects in your own domain, primitives/Strings, or other objects).
However, when a gateway is exposed through simple POJO methods that return values, it implies that, for each request message (generated when the method is invoked), there must be a reply message (generated when the method has returned).
Since messaging systems are naturally asynchronous, you may not always be able to guarantee the contract where “for each request, there will always be a reply”. Spring Integration 2.0 introduced support for an asynchronous gateway, which offers a convenient way to initiate flows when you may not know if a reply is expected or how long it takes for replies to arrive.
To handle these types of scenarios, Spring Integration uses java.util.concurrent.Future instances to support an asynchronous gateway.
From the XML configuration, nothing changes, and you still define asynchronous gateway the same way as you define a regular gateway, as the following example shows:
<int:gateway id="mathService"
service-interface="org.springframework.integration.sample.gateway.futures.MathServiceGateway"
default-request-channel="requestChannel"/>However, the gateway interface (a service interface) is a little different, as follows:
public interface MathServiceGateway {
Future<Integer> multiplyByTwo(int i);
}
As the preceding example shows, the return type for the gateway method is a Future.
When GatewayProxyFactoryBean sees that the return type of the gateway method is a Future, it immediately switches to the asynchronous mode by using an AsyncTaskExecutor.
That is the extent of the differences.
The call to such a method always returns immediately with a Future instance.
Then you can interact with the Future at your own pace to get the result, cancel, and so on.
Also, as with any other use of Future instances, calling get() may reveal a timeout, an execution exception, and so on.
The following example shows how to use a Future that returns from an asynchronous gateway:
MathServiceGateway mathService = ac.getBean("mathService", MathServiceGateway.class);
Future<Integer> result = mathService.multiplyByTwo(number);
// do something else here since the reply might take a moment
int finalResult = result.get(1000, TimeUnit.SECONDS);
For a more detailed example, see the async-gateway sample in the Spring Integration samples.
AsyncTaskExecutor
By default, the GatewayProxyFactoryBean uses org.springframework.core.task.SimpleAsyncTaskExecutor when submitting internal AsyncInvocationTask instances for any gateway method whose return type is a Future.
However, the async-executor attribute in the <gateway/> element’s configuration lets you provide a reference to any implementation of java.util.concurrent.Executor available within the Spring application context.
The (default) SimpleAsyncTaskExecutor supports both Future and CompletableFuture return types.
See CompletableFuture.
Even though there is a default executor, it is often useful to provide an external one so that you can identify its threads in logs (when using XML, the thread name is based on the executor’s bean name), as the following example shows:
@Bean
public AsyncTaskExecutor exec() {
SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
simpleAsyncTaskExecutor.setThreadNamePrefix("exec-");
return simpleAsyncTaskExecutor;
}
@MessagingGateway(asyncExecutor = "exec")
public interface ExecGateway {
@Gateway(requestChannel = "gatewayChannel")
Future<?> doAsync(String foo);
}
If you wish to return a different Future implementation, you can provide a custom executor or disable the executor altogether and return the Future in the reply message payload from the downstream flow.
To disable the executor, set it to null in the GatewayProxyFactoryBean (by using setAsyncTaskExecutor(null)).
When configuring the gateway with XML, use async-executor="".
When configuring by using the @MessagingGateway annotation, use code similar to the following:
@MessagingGateway(asyncExecutor = AnnotationConstants.NULL)
public interface NoExecGateway {
@Gateway(requestChannel = "gatewayChannel")
Future<?> doAsync(String foo);
}
If the return type is a specific concrete Future implementation or some other sub-interface that is not supported by the configured executor, the flow runs on the caller’s thread and the flow must return the required type in the reply message payload.
|
CompletableFuture
Starting with version 4.2, gateway methods can now return CompletableFuture<?>.
There are two modes of operation when returning this type:
-
When an async executor is provided and the return type is exactly
CompletableFuture(not a subclass), the framework runs the task on the executor and immediately returns aCompletableFutureto the caller.CompletableFuture.supplyAsync(Supplier<U> supplier, Executor executor)is used to create the future. -
When the async executor is explicitly set to
nulland the return type isCompletableFutureor the return type is a subclass ofCompletableFuture, the flow is invoked on the caller’s thread. In this scenario, the downstream flow is expected to return aCompletableFutureof the appropriate type.
The org.springframework.util.concurrent.ListenableFuture has been deprecated starting with Spring Framework 6.0.
It is recommended now to migrate to the CompletableFuture which provides similar processing functionality.
|
Usage Scenarios
In the following scenario, the caller thread returns immediately with a CompletableFuture<Invoice>, which is completed when the downstream flow replies to the gateway (with an Invoice object).
CompletableFuture<Invoice> order(Order order);
<int:gateway service-interface="something.Service" default-request-channel="orders" />In the following scenario, the caller thread returns with a CompletableFuture<Invoice> when the downstream flow provides it as the payload of the reply to the gateway.
Some other process must complete the future when the invoice is ready.
CompletableFuture<Invoice> order(Order order);
<int:gateway service-interface="foo.Service" default-request-channel="orders"
async-executor="" />In the following scenario, the caller thread returns with a CompletableFuture<Invoice> when the downstream flow provides it as the payload of the reply to the gateway.
Some other process must complete the future when the invoice is ready.
If DEBUG logging is enabled, a log entry is emitted, indicating that the async executor cannot be used for this scenario.
MyCompletableFuture<Invoice> order(Order order);
<int:gateway service-interface="foo.Service" default-request-channel="orders" />CompletableFuture instances can be used to perform additional manipulation on the reply, as the following example shows:
CompletableFuture<String> process(String data);
...
CompletableFuture result = process("foo")
.thenApply(t -> t.toUpperCase());
...
String out = result.get(10, TimeUnit.SECONDS);
Reactor Mono
Starting with version 5.0, the GatewayProxyFactoryBean allows the use of Project Reactor with gateway interface methods, using a Mono<T> return type.
The internal AsyncInvocationTask is wrapped in a Mono.fromCallable().
A Mono can be used to retrieve the result later (similar to a Future<?>), or you can consume from it with the dispatcher by invoking your Consumer when the result is returned to the gateway.
The Mono is not immediately flushed by the framework.
Consequently, the underlying message flow is not started before the gateway method returns (as it is with a Future<?> Executor task).
The flow starts when the Mono is subscribed to.
Alternatively, the Mono (being a “Composable”) might be a part of Reactor stream, when the subscribe() is related to the entire Flux.
The following example shows how to create a gateway with Project Reactor:
|
@MessagingGateway
public interface TestGateway {
@Gateway(requestChannel = "multiplyChannel")
Mono<Integer> multiply(Integer value);
}
@ServiceActivator(inputChannel = "multiplyChannel")
public Integer multiply(Integer value) {
return value * 2;
}
where such a gateway can be used in some service which deals with the Flux of data:
@Autowired
TestGateway testGateway;
public void hadnleFlux() {
Flux.just("1", "2", "3", "4", "5")
.map(Integer::parseInt)
.flatMap(this.testGateway::multiply)
.collectList()
.subscribe(System.out::println);
}
Another example that uses Project Reactor is a simple callback scenario, as the following example shows:
Mono<Invoice> mono = service.process(myOrder);
mono.subscribe(invoice -> handleInvoice(invoice));
The calling thread continues, with handleInvoice() being called when the flow completes.
Also see Kotlin Coroutines for more information.
Downstream Flows Returning an Asynchronous Type
As mentioned in the AsyncTaskExecutor section above, if you wish some downstream component to return a message with an async payload (Future, Mono, and others), you must explicitly set the async executor to null (or "" when using XML configuration).
The flow is then invoked on the caller thread and the result can be retrieved later.
Asynchronous void Return Type
The messaging gateway method can be declared like this:
@MessagingGateway
public interface MyGateway {
@Gateway(requestChannel = "sendAsyncChannel")
@Async
void sendAsync(String payload);
}
But downstream exceptions are not going to be propagated back to the caller.
To ensure asynchronous behavior for downstream flow invocation and exception propagation to the caller, starting with version 6.0, the framework provides support for the Future<Void> and Mono<Void> return types.
The use-case is similar to send-and-forget behavior described before for plain void return type, but with a difference that flow execution happens asynchronously and returned Future (or Mono) is complete with a null or exceptionally according to the send operation result.
If the Future<Void> is exact downstream flow reply, then an asyncExecutor option of the gateway must be set to null (AnnotationConstants.NULL for a @MessagingGateway configuration) and the send part is performed on a producer thread.
The reply one depends on the downstream flow configuration.
This way it is up target application to produce a Future<Void> reply correctly.
The Mono use-case is already out of the framework threading control, so setting asyncExecutor to null won’t make sense.
There Mono<Void> as a result of the request-reply gateway operation must be configured as a Mono<?> return type of the gateway method.
|
Gateway Behavior When No response Arrives
As explained earlier, the gateway provides a convenient way of interacting with a messaging system through POJO method invocations. However, a typical method invocation, which is generally expected to always return (even with an Exception), might not always map one-to-one to message exchanges (for example, a reply message might not arrive — the equivalent to a method not returning).
The rest of this section covers various scenarios and how to make the gateway behave more predictably.
Certain attributes can be configured to make synchronous gateway behavior more predictable, but some of them might not always work as you might expect.
One of them is reply-timeout (at the method level or default-reply-timeout at the gateway level).
We examine the reply-timeout attribute to see how it can and cannot influence the behavior of the synchronous gateway in various scenarios.
We examine a single-threaded scenario (all components downstream are connected through a direct channel) and multi-threaded scenarios (for example, somewhere downstream you may have a pollable or executor channel that breaks the single-thread boundary).
Long-running Process Downstream
- Sync Gateway, single-threaded
-
If a component downstream is still running (perhaps because of an infinite loop or a slow service), setting a
reply-timeouthas no effect, and the gateway method call does not return until the downstream service exits (by returning or throwing an exception). - Sync Gateway, multi-threaded
-
If a component downstream is still running (perhaps because of an infinite loop or a slow service) in a multi-threaded message flow, setting the
reply-timeouthas an effect by allowing gateway method invocation to return once the timeout has been reached, because theGatewayProxyFactoryBeanpolls on the reply channel, waiting for a message until the timeout expires. However, if the timeout has been reached before the actual reply was produced, it could result in a 'null' return from the gateway method. You should understand that the reply message (if produced) is sent to a reply channel after the gateway method invocation might have returned, so you must be aware of that and design your flow with it in mind.
Downstream Component Returns 'null'
- Sync Gateway — single-threaded
-
If a component downstream returns 'null' and no
reply-timeouthas been configured, the gateway method call hangs indefinitely, unless areply-timeouthas been configured or therequires-replyattribute has been set on the downstream component (for example, a service activator) that might return 'null'. In this case, an exception would be thrown and propagated to the gateway. - Sync Gateway — multi-threaded
-
The behavior is the same as the previous case.
Downstream Component Return Signature is 'void' While Gateway Method Signature Is Non-void
- Sync Gateway — single-threaded
-
If a component downstream returns 'void' and no
reply-timeouthas been configured, the gateway method call hangs indefinitely unless areply-timeouthas been configured. - Sync Gateway — multi-threaded
-
The behavior is the same as the previous case.
Downstream Component Results in Runtime Exception
- Sync Gateway — single-threaded
-
If a component downstream throws a runtime exception, the exception is propagated through an error message back to the gateway and re-thrown.
- Sync Gateway — multi-threaded
-
The behavior is the same as the previous case.
You should understand that, by default, reply-timeout is unbounded.
Consequently, if you do not explicitly set the reply-timeout, your gateway method invocation might hang indefinitely.
So, to make sure you analyze your flow and if there is even a remote possibility of one of these scenarios to occur, you should set the reply-timeout attribute to a "'safe'" value.
Even better, you can set the requires-reply attribute of the downstream component to 'true' to ensure a timely response, as produced by the throwing of an exception as soon as that downstream component returns null internally.
However, you should also realize that there are some scenarios (see the first one) where reply-timeout does not help.
That means it is also important to analyze your message flow and decide when to use a synchronous gateway rather than an asynchronous gateway.
As described earlier, the latter case is a matter of defining gateway methods that return Future instances.
Then you are guaranteed to receive that return value, and you have more granular control over the results of the invocation.
Also, when dealing with a router, you should remember that setting the resolution-required attribute to 'true' results in an exception thrown by the router if it can not resolve a particular channel.
Likewise, when dealing with a Filter, you can set the throw-exception-on-rejection attribute.
In both of these cases, the resulting flow behaves like it contain a service activator with the 'requires-reply' attribute.
In other words, it helps to ensure a timely response from the gateway method invocation.
|
reply-timeout is unbounded for <gateway/> elements (created by the GatewayProxyFactoryBean).
Inbound gateways for external integration (WS, HTTP, and so on) share many characteristics and attributes with these gateways.
However, for those inbound gateways, the default reply-timeout is 1000 milliseconds (one second).
If a downstream asynchronous hand-off is made to another thread, you may need to increase this attribute to allow enough time for the flow to complete before the gateway times out.
|
| You should understand that the timer starts when the thread returns to the gateway — that is, when the flow completes or a message is handed off to another thread. At that time, the calling thread starts waiting for the reply. If the flow was completely synchronous, the reply is immediately available. For asynchronous flows, the thread waits for up to this time. |
See IntegrationFlow as Gateway in the Java DSL chapter for options to define gateways through IntegrationFlow.
Service Activator
The service activator is the endpoint type for connecting any Spring-managed object to an input channel so that it may play the role of a service.
If the service produces output, it may also be connected to an output channel.
Alternatively, an output-producing service may be located at the end of a processing pipeline or message flow, in which case the inbound message’s replyChannel header can be used.
This is the default behavior if no output channel is defined.
As with most of the configuration options described here, the same behavior actually applies for most of the other components.
Configuring Service Activator
To create a service activator, use the 'service-activator' element with the 'input-channel' and 'ref' attributes, as the following example shows:
<int:service-activator input-channel="exampleChannel" ref="exampleHandler"/>The preceding configuration selects all the methods from the exampleHandler that meet one of the messaging requirements, which are as follows:
-
annotated with
@ServiceActivator -
is
public -
not return
voidifrequiresReply == true
The target method for invocation at runtime is selected for each request message by their payload type or as a fallback to the Message<?> type if such a method is present on target class.
Starting with version 5.0, one service method can be marked with the @org.springframework.integration.annotation.Default as a fallback for all non-matching cases.
This can be useful when using content-type conversion with the target method being invoked after conversion.
To delegate to an explicitly defined method of any object, you can add the method attribute, as the following example shows:
<int:service-activator input-channel="exampleChannel" ref="somePojo" method="someMethod"/>In either case, when the service method returns a non-null value, the endpoint tries to send the reply message to an appropriate reply channel.
To determine the reply channel, it first checks whether an output-channel was provided in the endpoint configuration, as the following example shows:
<int:service-activator input-channel="exampleChannel" output-channel="replyChannel"
ref="somePojo" method="someMethod"/>If the method returns a result and no output-channel is defined, the framework then checks the request message’s replyChannel header value.
If that value is available, it then checks its type.
If it is a MessageChannel, the reply message is sent to that channel.
If it is a String, the endpoint tries to resolve the channel name to a channel instance.
If the channel cannot be resolved, a DestinationResolutionException is thrown.
If it can be resolved, the message is sent there.
If the request message does not have a replyChannel header and the reply object is a Message, its replyChannel header is consulted for a target destination.
This is the technique used for request-reply messaging in Spring Integration, and it is also an example of the return address pattern.
If your method returns a result and you want to discard it and end the flow, you should configure the output-channel to send to a NullChannel.
For convenience, the framework registers one with the name, nullChannel.
See Special Channels for more information.
The service activator is one of those components that is not required to produce a reply message.
If your method returns null or has a void return type, the service activator exits after the method invocation, without any signals.
This behavior can be controlled by the AbstractReplyProducingMessageHandler.requiresReply option, which is also exposed as requires-reply when configuring with the XML namespace.
If the flag is set to true and the method returns null, a ReplyRequiredException is thrown.
The argument in the service method could be either a message or an arbitrary type.
If the latter, then it is assumed to be a message payload, which is extracted from the message and injected into the service method.
We generally recommend this approach, as it follows and promotes a POJO model when working with Spring Integration.
Arguments may also have @Header or @Headers annotations, as described in Annotation Support.
| The service method is not required to have any arguments, which means you can implement event-style service activators (where all you care about is an invocation of the service method) and not worry about the contents of the message. Think of it as a null JMS message. An example use case for such an implementation is a simple counter or monitor of messages deposited on the input channel. |
Starting with version 4.1, the framework correctly converts message properties (payload and headers) to the Java 8 Optional POJO method parameters, as the following example shows:
public class MyBean {
public String computeValue(Optional<String> payload,
@Header(value="foo", required=false) String foo1,
@Header(value="foo") Optional<String> foo2) {
if (payload.isPresent()) {
String value = payload.get();
...
}
else {
...
}
}
}
We generally recommend using a ref attribute if the custom service activator handler implementation can be reused in other <service-activator> definitions.
However, if the custom service activator handler implementation is only used within a single definition of the <service-activator>, you can provide an inner bean definition, as the following example shows:
<int:service-activator id="exampleServiceActivator" input-channel="inChannel"
output-channel = "outChannel" method="someMethod">
<beans:bean class="org.something.ExampleServiceActivator"/>
</int:service-activator>
Using both the ref attribute and an inner handler definition in the same <service-activator> configuration is not allowed, as it creates an ambiguous condition and results in an exception being thrown.
|
If the ref attribute references a bean that extends AbstractMessageProducingHandler (such as handlers provided by the framework itself), the configuration is optimized by injecting the output channel into the handler directly.
In this case, each ref must be to a separate bean instance (or a prototype-scoped bean) or use the inner <bean/> configuration type.
If you inadvertently reference the same message handler from multiple beans, you get a configuration exception.
|
Service Activators and the Spring Expression Language (SpEL)
Since Spring Integration 2.0, service activators can also benefit from SpEL.
For example, you can invoke any bean method without pointing to the bean in a ref attribute or including it as an inner bean definition, as follows:
<int:service-activator input-channel="in" output-channel="out"
expression="@accountService.processAccount(payload, headers.accountId)"/>
<bean id="accountService" class="thing1.thing2.Account"/>In the preceding configuration, instead of injecting 'accountService' by using a ref or as an inner bean, we use SpEL’s @beanId notation and invoke a method that takes a type compatible with the message payload.
We also pass a header value.
Any valid SpEL expression can be evaluated against any content in the message.
For simple scenarios, your service activators need not reference a bean if all logic can be encapsulated in such an expression, as the following example shows:
<int:service-activator input-channel="in" output-channel="out" expression="payload * 2"/>In the preceding configuration, our service logic is to multiply the payload value by two. SpEL lets us handle it relatively easily.
See Service Activators and the .handle() method in the Java DSL chapter for more information about configuring service activator.
Asynchronous Service Activator
The service activator is invoked by the calling thread.
This is an upstream thread if the input channel is a SubscribableChannel or a poller thread for a PollableChannel.
If the service returns a CompletableFuture<?>, the default action is to send that as the payload of the message sent to the output (or reply) channel.
Starting with version 4.3, you can now set the async attribute to true (by using setAsync(true) when using Java configuration).
If the service returns a CompletableFuture<?> when this the async attribute is set to true, the calling thread is released immediately and the reply message is sent on the thread (from within your service) that completes the future.
This is particularly advantageous for long-running services that use a PollableChannel, because the poller thread is released to perform other services within the framework.
If the service completes the future with an Exception, normal error processing occurs.
An ErrorMessage is sent to the errorChannel message header, if present.
Otherwise, an ErrorMessage is sent to the default errorChannel (if available).
Service Activator and Method Return Type
The service method can return any type which becomes reply message payload.
In this case a new Message<?> object is created and all the headers from a request message are copied.
This works the same way for most Spring Integration MessageHandler implementations, when interaction is based on a POJO method invocation.
A complete Message<?> object can also be returned from the method.
However, keep in mind that, unlike transformers, for a Service Activator this message will be modified by copying the headers from the request message if they are not already present in the returned message.
So, if your method parameter is a Message<?> and you copy some, but not all, existing headers in your service method, they will reappear in the reply message.
It is not a Service Activator responsibility to remove headers from a reply message and, pursuing the loosely-coupled principle, it is better to add a HeaderFilter in the integration flow.
Alternatively, a Transformer can be used instead of a Service Activator but, in that case, when returning a full Message<?> the method is completely responsible for the message, including copying request message headers (if needed).
You must ensure that important framework headers (e.g. replyChannel, errorChannel), if present, have to be preserved.
Delayer
A delayer is a simple endpoint that lets a message flow be delayed by a certain interval.
When a message is delayed, the original sender does not block.
Instead, the delayed messages are scheduled with an instance of org.springframework.scheduling.TaskScheduler to be sent to the output channel after the delay has passed.
This approach is scalable even for rather long delays, since it does not result in a large number of blocked sender threads.
On the contrary, in the typical case, a thread pool is used for the actual execution of releasing the messages.
This section contains several examples of configuring a delayer.
Configuring a Delayer
The <delayer> element is used to delay the message flow between two message channels.
As with the other endpoints, you can provide the 'input-channel' and 'output-channel' attributes, but the delayer also has 'default-delay' and 'expression' attributes (and the 'expression' element) that determine the number of milliseconds by which each message should be delayed.
The following example delays all messages by three seconds:
<int:delayer id="delayer" input-channel="input"
default-delay="3000" output-channel="output"/>If you need to determine the delay for each message, you can also provide the SpEL expression by using the 'expression' attribute, as the following expression shows:
@Bean
public IntegrationFlow flow() {
return IntegrationFlow.from("input")
.delay("delayer.messageGroupId", d -> d
.defaultDelay(3_000L)
.delayExpression("headers['delay']"))
.channel("output")
.get();
}
@Bean
fun flow() =
integrationFlow("input") {
delay("delayer.messageGroupId") {
defaultDelay(3000L)
delayExpression("headers['delay']")
}
channel("output")
}
@ServiceActivator(inputChannel = "input")
@Bean
public DelayHandler delayer() {
DelayHandler handler = new DelayHandler("delayer.messageGroupId");
handler.setDefaultDelay(3_000L);
handler.setDelayExpressionString("headers['delay']");
handler.setOutputChannelName("output");
return handler;
}
<int:delayer id="delayer" input-channel="input" output-channel="output"
default-delay="3000" expression="headers['delay']"/>In the preceding example, the three-second delay applies only when the expression evaluates to null for a given inbound message.
If you want to apply a delay only to messages that have a valid result of the expression evaluation, you can use a 'default-delay' of 0 (the default).
For any message that has a delay of 0 (or less), the message is sent immediately, on the calling thread.
The XML parser uses a message group ID of <beanName>.messageGroupId.
|
The delay handler supports expression evaluation results that represent an interval in milliseconds (any Object whose toString() method produces a value that can be parsed into a Long) as well as java.util.Date instances representing an absolute time.
In the first case, the milliseconds are counted from the current time (for example a value of 5000 would delay the message for at least five seconds from the time it is received by the delayer).
With a Date instance, the message is not released until the time represented by that Date object.
A value that equates to a non-positive delay or a Date in the past results in no delay.
Instead, it is sent directly to the output channel on the original sender’s thread.
If the expression evaluation result is not a Date and can not be parsed as a Long, the default delay (if any — the default is 0) is applied.
|
The expression evaluation may throw an evaluation exception for various reasons, including an invalid expression or other conditions.
By default, such exceptions are ignored (though logged at the DEBUG level) and the delayer falls back to the default delay (if any).
You can modify this behavior by setting the ignore-expression-failures attribute.
By default, this attribute is set to true and the delayer behavior is as described earlier.
However, if you wish to not ignore expression evaluation exceptions and throw them to the delayer’s caller, set the ignore-expression-failures attribute to false.
|
|
In the preceding example, the delay expression is specified as Consequently, if there is a possibility of the header being omitted and you want to fall back to the default delay, it is generally more efficient (and recommended) using the indexer syntax instead of dot property accessor syntax, because detecting the null is faster than catching an exception. |
The delayer delegates to an instance of Spring’s TaskScheduler abstraction.
The default scheduler used by the delayer is the ThreadPoolTaskScheduler instance provided by Spring Integration on startup.
See Configuring the Task Scheduler.
If you want to delegate to a different scheduler, you can provide a reference through the delayer element’s 'scheduler' attribute, as the following example shows:
<int:delayer id="delayer" input-channel="input" output-channel="output"
expression="headers.delay"
scheduler="exampleTaskScheduler"/>
<task:scheduler id="exampleTaskScheduler" pool-size="3"/>
If you configure an external ThreadPoolTaskScheduler, you can set waitForTasksToCompleteOnShutdown = true on this property.
It allows successful completion of 'delay' tasks that are already in the execution state (releasing the message) when the application is shutdown.
Before Spring Integration 2.2, this property was available on the <delayer> element, because DelayHandler could create its own scheduler on the background.
Since 2.2, the delayer requires an external scheduler instance and waitForTasksToCompleteOnShutdown was deleted.
You should use the scheduler’s own configuration.
|
ThreadPoolTaskScheduler has a property errorHandler, which can be injected with some implementation of org.springframework.util.ErrorHandler.
This handler allows processing an Exception from the thread of the scheduled task sending the delayed message.
By default, it uses an org.springframework.scheduling.support.TaskUtils$LoggingErrorHandler, and you can see a stack trace in the logs.
You might want to consider using an org.springframework.integration.channel.MessagePublishingErrorHandler, which sends an ErrorMessage into an error-channel, either from the failed message’s header or into the default error-channel.
This error handling is performed after a transaction rolls back (if present).
See Release Failures.
|
Delayer and a Message Store
The DelayHandler persists delayed messages into the message group in the provided MessageStore.
(The 'groupId' is based on the required 'id' attribute of the <delayer> element.)
A delayed message is removed from the MessageStore by the scheduled task immediately before the DelayHandler sends the message to the output-channel.
If the provided MessageStore is persistent (such as JdbcMessageStore), it provides the ability to not lose messages on the application shutdown.
After application startup, the DelayHandler reads messages from its message group in the MessageStore and reschedules them with a delay based on the original arrival time of the message (if the delay is numeric).
For messages where the delay header was a Date, that Date is used when rescheduling.
If a delayed message remains in the MessageStore more than its 'delay', it is sent immediately after startup.
The <delayer> can be enriched with either of two mutually exclusive elements: <transactional> and <advice-chain>.
The List of these AOP advices is applied to the proxied internal DelayHandler.ReleaseMessageHandler, which has the responsibility to release the message, after the delay, on a Thread of the scheduled task.
It might be used, for example, when the downstream message flow throws an exception and the transaction of the ReleaseMessageHandler is rolled back.
In this case, the delayed message remains in the persistent MessageStore.
You can use any custom org.aopalliance.aop.Advice implementation within the <advice-chain>.
The <transactional> element defines a simple advice chain that has only the transactional advice.
The following example shows an advice-chain within a <delayer>:
<int:delayer id="delayer" input-channel="input" output-channel="output"
expression="headers.delay"
message-store="jdbcMessageStore">
<int:advice-chain>
<beans:ref bean="customAdviceBean"/>
<tx:advice>
<tx:attributes>
<tx:method name="*" read-only="true"/>
</tx:attributes>
</tx:advice>
</int:advice-chain>
</int:delayer>The DelayHandler can be exported as a JMX MBean with managed operations (getDelayedMessageCount and reschedulePersistedMessages), which allows the rescheduling of delayed persisted messages at runtime — for example, if the TaskScheduler has previously been stopped.
These operations can be invoked through a Control Bus command, as the following example shows:
Message<String> delayerReschedulingMessage =
MessageBuilder.withPayload("@'delayer.handler'.reschedulePersistedMessages()").build();
controlBusChannel.send(delayerReschedulingMessage);
| For more information regarding the message store, JMX, and the control bus, see System Management. |
Starting with version 5.3.7, if a transaction is active when a message is stored into a MessageStore, the release task is scheduled in a TransactionSynchronization.afterCommit() callback.
This is necessary to prevent a race condition, where the scheduled release could run before the transaction has committed, and the message is not found.
In this case, the message will be released after the delay, or after the transaction commits, whichever is later.
Release Failures
Starting with version 5.0.8, there are two new properties on the delayer:
-
maxAttempts(default 5) -
retryDelay(default 1 second)
When a message is released, if the downstream flow fails, the release will be attempted after the retryDelay.
If the maxAttempts is reached, the message is discarded (unless the release is transactional, in which case the message will remain in the store, but will no longer be scheduled for release, until the application is restarted, or the reschedulePersistedMessages() method is invoked, as discussed above).
In addition, you can configure a delayedMessageErrorChannel; when a release fails, an ErrorMessage is sent to that channel with the exception as the payload and has the originalMessage property.
The ErrorMessage contains a header IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT containing the current count.
If the error flow consumes the error message and exits normally, no further action is taken; if the release is transactional, the transaction will commit and the message deleted from the store.
If the error flow throws an exception, the release will be retried up to maxAttempts as discussed above.
Scripting Support
Spring Integration 2.1 added support for the JSR223 Scripting for Java specification, introduced in Java version 6. It lets you use scripts written in any supported language (including Ruby, JRuby, Groovy and Kotlin) to provide the logic for various integration components, similar to the way the Spring Expression Language (SpEL) is used in Spring Integration. For more information about JSR223, see the documentation.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-scripting</artifactId>
<version>6.0.9</version>
</dependency>compile "org.springframework.integration:spring-integration-scripting:6.0.9"
In addition, you need to add a script engine implementation, e.g. JRuby, Jython.
Starting with version 5.2, Spring Integration provides a Kotlin Jsr223 support. You need to add these dependencies into your project to make it working:
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-script-util</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-compiler-embeddable</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-scripting-compiler-embeddable</artifactId>
<scope>runtime</scope>
</dependency>runtime 'org.jetbrains.kotlin:kotlin-script-util'
runtime 'org.jetbrains.kotlin:kotlin-compiler-embeddable'
runtime 'org.jetbrains.kotlin:kotlin-scripting-compiler-embeddable'
The KotlinScriptExecutor is selected by the provided kotlin language indicator or script file comes with the .kts extension.
In order to use a JVM scripting language, a JSR223 implementation for that language must be included in your class path. The Groovy and JRuby projects provide JSR233 support in their standard distributions.
| Various JSR223 language implementations have been developed by third parties. A particular implementation’s compatibility with Spring Integration depends on how well it conforms to the specification and the implementer’s interpretation of the specification. |
| If you plan to use Groovy as your scripting language, we recommended you use Spring-Integration’s Groovy Support as it offers additional features specific to Groovy. However, this section is relevant as well. |
Script Configuration
Depending on the complexity of your integration requirements, scripts may be provided inline as CDATA in XML configuration or as a reference to a Spring resource that contains the script.
To enable scripting support, Spring Integration defines a ScriptExecutingMessageProcessor, which binds the message payload to a variable named payload and the message headers to a headers variable, both accessible within the script execution context.
All you need to do is write a script that uses these variables.
The following pair of examples show sample configurations that create filters:
@Bean
public IntegrationFlow scriptFilter() {
return f -> f.filter(Scripts.processor("some/path/to/ruby/script/RubyFilterTests.rb"));
}
...
@Bean
public Resource scriptResource() {
return new ByteArrayResource("headers.type == 'good'".getBytes());
}
@Bean
public IntegrationFlow scriptFilter() {
return f -> f.filter(Scripts.processor(scriptResource()).lang("groovy"));
}
<int:filter input-channel="referencedScriptInput">
<int-script:script location="some/path/to/ruby/script/RubyFilterTests.rb"/>
</int:filter>
<int:filter input-channel="inlineScriptInput">
<int-script:script lang="groovy">
<![CDATA[
return payload == 'good'
]]>
</int-script:script>
</int:filter>As the preceding examples show, the script can be included inline or can be included by reference to a resource location (by using the location attribute).
Additionally, the lang attribute corresponds to the language name (or its JSR223 alias).
Other Spring Integration endpoint elements that support scripting include router, service-activator, transformer, and splitter.
The scripting configuration in each case would be identical to the above (besides the endpoint element).
Another useful feature of scripting support is the ability to update (reload) scripts without having to restart the application context.
To do so, specify the refresh-check-delay attribute on the script element, as the following example shows:
Scripts.processor(...).refreshCheckDelay(5000)
}
<int-script:script location="..." refresh-check-delay="5000"/>In the preceding example, the script location is checked for updates every 5 seconds. If the script is updated, any invocation that occurs later than 5 seconds since the update results in running the new script.
Consider the following example:
Scripts.processor(...).refreshCheckDelay(0)
}
<int-script:script location="..." refresh-check-delay="0"/>In the preceding example, the context is updated with any script modifications as soon as such modification occurs, providing a simple mechanism for 'real-time' configuration. Any negative value means the script is not reloaded after initialization of the application context. This is the default behavior. The following example shows a script that never updates:
Scripts.processor(...).refreshCheckDelay(-1)
}
<int-script:script location="..." refresh-check-delay="-1"/>| Inline scripts can not be reloaded. |
Script Variable Bindings
Variable bindings are required to enable the script to reference variables externally provided to the script’s execution context.
By default, payload and headers are used as binding variables.
You can bind additional variables to a script by using <variable> elements (or ScriptSpec.variables() option), as the following example shows:
Scripts.processor("foo/bar/MyScript.py")
.variables(Map.of("var1", "thing1", "var2", "thing2", "date", date))
}
<script:script lang="py" location="foo/bar/MyScript.py">
<script:variable name="var1" value="thing1"/>
<script:variable name="var2" value="thing2"/>
<script:variable name="date" ref="date"/>
</script:script>As shown in the preceding example, you can bind a script variable either to a scalar value or to a Spring bean reference.
Note that payload and headers are still included as binding variables.
With Spring Integration 3.0, in addition to the variable element, the variables attribute has been introduced.
This attribute and the variable elements are not mutually exclusive, and you can combine them within one script component.
However, variables must be unique, regardless of where they are defined.
Also, since Spring Integration 3.0, variable bindings are allowed for inline scripts, too, as the following example shows:
<service-activator input-channel="input">
<script:script lang="ruby" variables="thing1=THING1, date-ref=dateBean">
<script:variable name="thing2" ref="thing2Bean"/>
<script:variable name="thing3" value="thing2"/>
<![CDATA[
payload.foo = thing1
payload.date = date
payload.bar = thing2
payload.baz = thing3
payload
]]>
</script:script>
</service-activator>The preceding example shows a combination of an inline script, a variable element, and a variables attribute.
The variables attribute contains a comma-separated value, where each segment contains an '=' separated pair of the variable and its value.
The variable name can be suffixed with -ref, as in the date-ref variable in the preceding example.
That means that the binding variable has the name, date, but the value is a reference to the dateBean bean from the application context.
This may be useful when using property placeholder configuration or command-line arguments.
If you need more control over how variables are generated, you can implement your own Java class that uses the ScriptVariableGenerator strategy, which is defined by the following interface:
public interface ScriptVariableGenerator {
Map<String, Object> generateScriptVariables(Message<?> message);
}
This interface requires you to implement the generateScriptVariables(Message) method.
The message argument lets you access any data available in the message payload and headers, and the return value is the Map of bound variables.
This method is called every time the script is executed for a message.
The following example shows how to provide an implementation of ScriptVariableGenerator and reference it with the script-variable-generator attribute:
Scripts.processor("foo/bar/MyScript.groovy")
.variableGenerator(new foo.bar.MyScriptVariableGenerator())
}
<int-script:script location="foo/bar/MyScript.groovy"
script-variable-generator="variableGenerator"/>
<bean id="variableGenerator" class="foo.bar.MyScriptVariableGenerator"/>If a script-variable-generator is not provided, script components use DefaultScriptVariableGenerator, which merges any provided <variable> elements with payload and headers variables from the Message in its generateScriptVariables(Message) method.
You cannot provide both the script-variable-generator attribute and <variable> element(s).
They are mutually exclusive.
|
GraalVM Polyglot
Starting with version 6.0, the framework provides a PolyglotScriptExecutor which is based the GraalVM Polyglot API.
The JSR223 engine implementation for JavaScript, removed from Java by itself, has been replaced by using this new script executor.
See more information about enabling JavaScript support in GraalVM and what configuration options can be propagated via script variables.
By default, the framework sets allowAllAccess to true on the shared Polyglot Context which enables this interaction with host JVM:
-
The creation and use of new threads.
-
The access to public host classes.
-
The loading of new host classes by adding entries to the class path.
-
Exporting new members into the polyglot bindings.
-
Unrestricted IO operations on host system.
-
Passing experimental options.
-
The creation and use of new sub-processes.
-
The access to process environment variables.
This can be customized via overloaded PolyglotScriptExecutor constructor which accepts a org.graalvm.polyglot.Context.Builder.
To enable this JavaScript support, GraalVM with the js component installed has to be used or, when using a regular JVM, the org.graalvm.sdk:graal-sdk and org.graalvm.js:js dependencies must be included.
Groovy Support
In Spring Integration 2.0, we added Groovy support, letting you use the Groovy scripting language to provide the logic for various integration components — similar to the way the Spring Expression Language (SpEL) is supported for routing, transformation, and other integration concerns. For more information about Groovy, see the Groovy documentation, which you can find on the project website.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-groovy</artifactId>
<version>6.0.9</version>
</dependency>compile "org.springframework.integration:spring-integration-groovy:6.0.9"
In addition, starting with version 6.0, a Groovy DSL for integration flow configurations is provided.
Groovy Configuration
With Spring Integration 2.1, the configuration namespace for the Groovy support is an extension of Spring Integration’s scripting support and shares the core configuration and behavior described in detail in the Scripting Support section.
Even though Groovy scripts are well-supported by generic scripting support, the Groovy support provides the Groovy configuration namespace, which is backed by the Spring Framework’s org.springframework.scripting.groovy.GroovyScriptFactory and related components, offering extended capabilities for using Groovy.
The following listing shows two sample configurations:
<int:filter input-channel="referencedScriptInput">
<int-groovy:script location="some/path/to/groovy/file/GroovyFilterTests.groovy"/>
</int:filter>
<int:filter input-channel="inlineScriptInput">
<int-groovy:script><![CDATA[
return payload == 'good'
]]></int-groovy:script>
</int:filter>As the preceding examples show, the configuration looks identical to the general scripting support configuration.
The only difference is the use of the Groovy namespace, as indicated by the int-groovy namespace prefix.
Also note that the lang attribute on the <script> tag is not valid in this namespace.
Groovy Object Customization
If you need to customize the Groovy object itself (beyond setting variables) you can reference a bean that implements GroovyObjectCustomizer by using the customizer attribute.
For example, this might be useful if you want to implement a domain-specific language (DSL) by modifying the MetaClass and registering functions to be available within the script.
The following example shows how to do so:
<int:service-activator input-channel="groovyChannel">
<int-groovy:script location="somewhere/SomeScript.groovy" customizer="groovyCustomizer"/>
</int:service-activator>
<beans:bean id="groovyCustomizer" class="org.something.MyGroovyObjectCustomizer"/>Setting a custom GroovyObjectCustomizer is not mutually exclusive with <variable> elements or the script-variable-generator attribute.
It can also be provided when defining an inline script.
Spring Integration 3.0 introduced the variables attribute, which works in conjunction with the variable element.
Also, groovy scripts have the ability to resolve a variable to a bean in the BeanFactory, if a binding variable was not provided with the name.
The following example shows how to use a variable (entityManager):
<int-groovy:script>
<![CDATA[
entityManager.persist(payload)
payload
]]>
</int-groovy:script>entityManager must be an appropriate bean in the application context.
For more information regarding the <variable> element, the variables attribute, and the script-variable-generator attribute, see Script Variable Bindings.
Groovy Script Compiler Customization
The @CompileStatic hint is the most popular Groovy compiler customization option.
It can be used on the class or method level.
For more information, see the Groovy Reference Manual and, specifically, @CompileStatic.
To utilize this feature for short scripts (in integration scenarios), we are forced to change simple scripts to more Java-like code.
Consider the following <filter> script:
headers.type == 'good'
The preceding script becomes the following method in Spring Integration:
@groovy.transform.CompileStatic
String filter(Map headers) {
headers.type == 'good'
}
filter(headers)
With that, the filter() method is transformed and compiled to static Java code, bypassing the Groovy
dynamic phases of invocation, such as getProperty() factories and CallSite proxies.
Starting with version 4.3, you can configure the Spring Integration Groovy components with the compile-static boolean option, specifying that ASTTransformationCustomizer for @CompileStatic should be added to the internal CompilerConfiguration.
With that in place, you can omit the method declaration with @CompileStatic in our script code and still get compiled plain Java code.
In this case, the preceding script can be short but still needs to be a little more verbose than interpreted script, as the following example shows:
binding.variables.headers.type == 'good'
You must access the headers and payload (or any other) variables through the groovy.lang.Script binding property because, with @CompileStatic, we do not have the dynamic GroovyObject.getProperty() capability.
In addition, we introduced the compiler-configuration bean reference.
With this attribute, you can provide any other required Groovy compiler customizations, such as ImportCustomizer.
For more information about this feature, see the Groovy Documentation for advanced compiler configuration.
Using compilerConfiguration does not automatically add an ASTTransformationCustomizer for the @CompileStatic annotation, and it overrides the compileStatic option.
If you still need CompileStatic, you should manually add a new ASTTransformationCustomizer(CompileStatic.class) into the CompilationCustomizers of that custom compilerConfiguration.
|
The Groovy compiler customization does not have any effect on the refresh-check-delay option, and reloadable scripts can be statically compiled, too.
|
Control Bus
As described in (Enterprise Integration Patterns), the idea behind the control bus is that you can use the same messaging system for monitoring and managing the components within the framework as is used for “application-level” messaging. In Spring Integration, we build upon the adapters described earlier so that you can send Messages as a means of invoking exposed operations. One option for those operations is Groovy scripts. The following example configures a Groovy script for the control bus:
<int-groovy:control-bus input-channel="operationChannel"/>The control bus has an input channel that can be accessed to invoke operations on the beans in the application context.
The Groovy control bus runs messages on the input channel as Groovy scripts.
It takes a message, compiles the body to a script, customizes it with a GroovyObjectCustomizer, and runs it.
The control bus' MessageProcessor exposes all beans in the application context that are annotated with @ManagedResource and implement Spring’s Lifecycle interface or extend Spring’s CustomizableThreadCreator base class (for example, several of the TaskExecutor and TaskScheduler implementations).
Be careful about using managed beans with custom scopes (such as 'request') in the Control Bus' command scripts, especially inside an asynchronous message flow.
If MessageProcessor of the control bus cannot expose a bean from the application context, you may end up with some BeansException during the command script’s run.
For example, if a custom scope’s context is not established, the attempt to get a bean within that scope triggers a BeanCreationException.
|
If you need to further customize the Groovy objects, you can also provide a reference to a bean that implements GroovyObjectCustomizer through the customizer attribute, as the following example shows:
<int-groovy:control-bus input-channel="input"
output-channel="output"
customizer="groovyCustomizer"/>
<beans:bean id="groovyCustomizer" class="org.foo.MyGroovyObjectCustomizer"/>Adding Behavior to Endpoints
Prior to Spring Integration 2.2, you could add behavior to an entire Integration flow by adding an AOP Advice to a poller’s <advice-chain/> element.
However, suppose you want to retry, say, just a REST Web Service call, and not any downstream endpoints.
For example, consider the following flow:
inbound-adapter->poller->http-gateway1->http-gateway2->jdbc-outbound-adapterIf you configure some retry-logic into an advice chain on the poller and the call to http-gateway2 failed because of a network glitch, the retry causes both http-gateway1 and http-gateway2 to be called a second time.
Similarly, after a transient failure in the jdbc-outbound-adapter, both HTTP gateways are called a second time before again calling the jdbc-outbound-adapter.
Spring Integration 2.2 adds the ability to add behavior to individual endpoints.
This is achieved by the addition of the <request-handler-advice-chain/> element to many endpoints.
The following example shows how to the <request-handler-advice-chain/> element within an outbound-gateway:
<int-http:outbound-gateway id="withAdvice"
url-expression="'http://localhost/test1'"
request-channel="requests"
reply-channel="nextChannel">
<int-http:request-handler-advice-chain>
<ref bean="myRetryAdvice" />
</int-http:request-handler-advice-chain>
</int-http:outbound-gateway>In this case, myRetryAdvice is applied only locally to this gateway and does not apply to further actions taken downstream after the reply is sent to nextChannel.
The scope of the advice is limited to the endpoint itself.
|
At this time, you cannot advise an entire However, a |
Provided Advice Classes
In addition to providing the general mechanism to apply AOP advice classes, Spring Integration provides these out-of-the-box advice implementations:
-
RequestHandlerRetryAdvice(described in Retry Advice) -
RequestHandlerCircuitBreakerAdvice(described in Circuit Breaker Advice) -
ExpressionEvaluatingRequestHandlerAdvice(described in Expression Evaluating Advice) -
RateLimiterRequestHandlerAdvice(described in Rate Limiter Advice) -
CacheRequestHandlerAdvice(described in Caching Advice) -
ReactiveRequestHandlerAdvice(described in Reactive Advice)
Retry Advice
The retry advice (o.s.i.handler.advice.RequestHandlerRetryAdvice) leverages the rich retry mechanisms provided by the Spring Retry project.
The core component of spring-retry is the RetryTemplate, which allows configuration of sophisticated retry scenarios, including RetryPolicy and BackoffPolicy strategies (with a number of implementations) as well as a RecoveryCallback strategy to determine the action to take when retries are exhausted.
- Stateless Retry
-
Stateless retry is the case where the retry activity is handled entirely within the advice. The thread pauses (if configured to do so) and retries the action.
- Stateful Retry
-
Stateful retry is the case where the retry state is managed within the advice but where an exception is thrown and the caller resubmits the request. An example for stateful retry is when we want the message originator (for example,JMS) to be responsible for resubmitting, rather than performing it on the current thread. Stateful retry needs some mechanism to detect a retried submission.
For more information on spring-retry, see the project’s Javadoc and the reference documentation for Spring Batch, where spring-retry originated.
| The default back off behavior is to not back off. Retries are attempted immediately. Using a back off policy that causes threads to pause between attempts may cause performance issues, including excessive memory use and thread starvation. In high-volume environments, back off policies should be used with caution. |
Configuring the Retry Advice
The examples in this section use the following <service-activator> that always throws an exception:
public class FailingService {
public void service(String message) {
throw new RuntimeException("error");
}
}
- Simple Stateless Retry
-
The default
RetryTemplatehas aSimpleRetryPolicywhich tries three times. There is noBackOffPolicy, so the three attempts are made back-to-back-to-back with no delay between attempts. There is noRecoveryCallback, so the result is to throw the exception to the caller after the final failed retry occurs. In a Spring Integration environment, this final exception might be handled by using anerror-channelon the inbound endpoint. The following example usesRetryTemplateand shows itsDEBUGoutput:<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"/> </int:request-handler-advice-chain> </int:service-activator> DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...] DEBUG [task-scheduler-2]Retry: count=0 DEBUG [task-scheduler-2]Checking for rethrow: count=1 DEBUG [task-scheduler-2]Retry: count=1 DEBUG [task-scheduler-2]Checking for rethrow: count=2 DEBUG [task-scheduler-2]Retry: count=2 DEBUG [task-scheduler-2]Checking for rethrow: count=3 DEBUG [task-scheduler-2]Retry failed last attempt: count=3 - Simple Stateless Retry with Recovery
-
The following example adds a
RecoveryCallbackto the preceding example and uses anErrorMessageSendingRecovererto send anErrorMessageto 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> </int: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 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 theSimpleRetryPolicybut increases the attempts to four. It also adds anExponentialBackoffPolicywhere the first retry waits one second, the second waits five seconds and the third waits 25 (for four attempts in all). The following listing shows the example and itsDEBUGoutput:<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> </int:request-handler-advice-chain> </int:service-activator> <bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate"> <property name="retryPolicy"> <bean class="org.springframework.retry.policy.SimpleRetryPolicy"> <property name="maxAttempts" value="4" /> </bean> </property> <property name="backOffPolicy"> <bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy"> <property name="initialInterval" value="1000" /> <property name="multiplier" value="5.0" /> <property name="maxInterval" value="60000" /> </bean> </property> </bean> 27.058 DEBUG [task-scheduler-1]preSend on channel 'input', message: [Payload=...] 27.071 DEBUG [task-scheduler-1]Retry: count=0 27.080 DEBUG [task-scheduler-1]Sleeping for 1000 28.081 DEBUG [task-scheduler-1]Checking for rethrow: count=1 28.081 DEBUG [task-scheduler-1]Retry: count=1 28.081 DEBUG [task-scheduler-1]Sleeping for 5000 33.082 DEBUG [task-scheduler-1]Checking for rethrow: count=2 33.082 DEBUG [task-scheduler-1]Retry: count=2 33.083 DEBUG [task-scheduler-1]Sleeping for 25000 58.083 DEBUG [task-scheduler-1]Checking for rethrow: count=3 58.083 DEBUG [task-scheduler-1]Retry: count=3 58.084 DEBUG [task-scheduler-1]Checking for rethrow: count=4 58.084 DEBUG [task-scheduler-1]Retry failed last attempt: count=4 58.086 DEBUG [task-scheduler-1]Sending ErrorMessage :failedMessage:[Payload=...] - Namespace Support for Stateless Retry
-
Starting with version 4.0, the preceding configuration can be greatly simplified, thanks to the namespace support for the retry advice, as the following example shows:
<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <ref bean="retrier" /> </int:request-handler-advice-chain> </int:service-activator> <int:handler-retry-advice id="retrier" max-attempts="4" recovery-channel="myErrorChannel"> <int:exponential-back-off initial="1000" multiplier="5.0" maximum="60000" /> </int:handler-retry-advice>In the preceding example, the advice is defined as a top-level bean so that it can be used in multiple
request-handler-advice-chaininstances. You can also define the advice directly within the chain, as the following example shows:<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <int:retry-advice id="retrier" max-attempts="4" recovery-channel="myErrorChannel"> <int:exponential-back-off initial="1000" multiplier="5.0" maximum="60000" /> </int:retry-advice> </int:request-handler-advice-chain> </int:service-activator>A
<handler-retry-advice>can have a<fixed-back-off>or<exponential-back-off>child element or have no child element. A<handler-retry-advice>with no child element uses no back off. If there is norecovery-channel, the exception is thrown when retries are exhausted. The namespace can only be used with stateless retry.For more complex environments (custom policies etc.), use normal
<bean>definitions. - Simple Stateful Retry with Recovery
-
To make retry stateful, we need to provide the advice with a
RetryStateGeneratorimplementation. This class is used to identify a message as being a resubmission so that theRetryTemplatecan determine the current state of retry for this message. The framework provides aSpelExpressionRetryStateGenerator, which determines the message identifier by using a SpEL expression. This example again uses the default policies (three attempts with no back off). As with stateless retry, these policies can be customized. The following listing shows the example and itsDEBUGoutput:<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"> <property name="retryStateGenerator"> <bean class="o.s.i.handler.advice.SpelExpressionRetryStateGenerator"> <constructor-arg value="headers['jms_messageId']" /> </bean> </property> <property name="recoveryCallback"> <bean class="o.s.i.handler.advice.ErrorMessageSendingRecoverer"> <constructor-arg ref="myErrorChannel" /> </bean> </property> </bean> </int:request-handler-advice-chain> </int:service-activator> 24.351 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 24.368 DEBUG [Container#0-1]Retry: count=0 24.387 DEBUG [Container#0-1]Checking for rethrow: count=1 24.387 DEBUG [Container#0-1]Rethrow in retry for policy: count=1 24.387 WARN [Container#0-1]failure occurred in gateway sendAndReceive org.springframework.integration.MessagingException: Failed to invoke handler ... Caused by: java.lang.RuntimeException: foo ... 24.391 DEBUG [Container#0-1]Initiating transaction rollback on application exception ... 25.412 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 25.412 DEBUG [Container#0-1]Retry: count=1 25.413 DEBUG [Container#0-1]Checking for rethrow: count=2 25.413 DEBUG [Container#0-1]Rethrow in retry for policy: count=2 25.413 WARN [Container#0-1]failure occurred in gateway sendAndReceive org.springframework.integration.MessagingException: Failed to invoke handler ... Caused by: java.lang.RuntimeException: foo ... 25.414 DEBUG [Container#0-1]Initiating transaction rollback on application exception ... 26.418 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 26.418 DEBUG [Container#0-1]Retry: count=2 26.419 DEBUG [Container#0-1]Checking for rethrow: count=3 26.419 DEBUG [Container#0-1]Rethrow in retry for policy: count=3 26.419 WARN [Container#0-1]failure occurred in gateway sendAndReceive org.springframework.integration.MessagingException: Failed to invoke handler ... Caused by: java.lang.RuntimeException: foo ... 26.420 DEBUG [Container#0-1]Initiating transaction rollback on application exception ... 27.425 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 27.426 DEBUG [Container#0-1]Retry failed last attempt: count=3 27.426 DEBUG [Container#0-1]Sending ErrorMessage :failedMessage:[Payload=...]If you compare the preceding example with the stateless examples, you can see that, with stateful retry, the exception is thrown to the caller on each failure.
- Exception Classification for Retry
-
Spring Retry has a great deal of flexibility for determining which exceptions can invoke retry. The default configuration retries for all exceptions and the exception classifier looks at the top-level exception. If you configure it to, say, retry only on
MyExceptionand your application throws aSomeOtherExceptionwhere the cause is aMyException, retry does not occur.Since Spring Retry 1.0.3, the
BinaryExceptionClassifierhas a property calledtraverseCauses(the default isfalse). Whentrue, it traverses exception causes until it finds a match or runs out of causes traversing.To use this classifier for retry, use a
SimpleRetryPolicycreated with the constructor that takes the max attempts, theMapofExceptionobjects, and thetraverseCausesboolean. Then you can inject this policy into theRetryTemplate.
traverseCauses is required in this case because user exceptions may be wrapped in a MessagingException.
|
Circuit Breaker Advice
The general idea of the circuit breaker pattern is that, if a service is not currently available, do not waste time (and resources) trying to use it.
The o.s.i.handler.advice.RequestHandlerCircuitBreakerAdvice implements this pattern.
When the circuit breaker is in the closed state, the endpoint attempts to invoke the service.
The circuit breaker goes to the open state if a certain number of consecutive attempts fail.
When it is in the open state, new requests “fail fast” and no attempt is made to invoke the service until some time has expired.
When that time has expired, the circuit breaker is set to the half-open state. When in this state, if even a single attempt fails, the breaker immediately goes to the open state. If the attempt succeeds, the breaker goes to the closed state, in which case it does not go to the open state again until the configured number of consecutive failures again occur. Any successful attempt resets the state to zero failures for the purpose of determining when the breaker might go to the open state again.
Typically, this advice might be used for external services, where it might take some time to fail (such as a timeout attempting to make a network connection).
The RequestHandlerCircuitBreakerAdvice has two properties: threshold and halfOpenAfter.
The threshold property represents the number of consecutive failures that need to occur before the breaker goes open.
It defaults to 5.
The halfOpenAfter property represents the time after the last failure that the breaker waits before attempting another request.
The default is 1000 milliseconds.
The following example configures a circuit breaker and shows its DEBUG and ERROR output:
<int:service-activator input-channel="input" ref="failer" method="service">
<int:request-handler-advice-chain>
<bean class="o.s.i.handler.advice.RequestHandlerCircuitBreakerAdvice">
<property name="threshold" value="2" />
<property name="halfOpenAfter" value="12000" />
</bean>
</int:request-handler-advice-chain>
</int:service-activator>
05.617 DEBUG [task-scheduler-1]preSend on channel 'input', message: [Payload=...]
05.638 ERROR [task-scheduler-1]org.springframework.messaging.MessageHandlingException: java.lang.RuntimeException: foo
...
10.598 DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...]
10.600 ERROR [task-scheduler-2]org.springframework.messaging.MessageHandlingException: java.lang.RuntimeException: foo
...
15.598 DEBUG [task-scheduler-3]preSend on channel 'input', message: [Payload=...]
15.599 ERROR [task-scheduler-3]org.springframework.messaging.MessagingException: Circuit Breaker is Open for ServiceActivator
...
20.598 DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...]
20.598 ERROR [task-scheduler-2]org.springframework.messaging.MessagingException: Circuit Breaker is Open for ServiceActivator
...
25.598 DEBUG [task-scheduler-5]preSend on channel 'input', message: [Payload=...]
25.601 ERROR [task-scheduler-5]org.springframework.messaging.MessageHandlingException: java.lang.RuntimeException: foo
...
30.598 DEBUG [task-scheduler-1]preSend on channel 'input', message: [Payload=foo...]
30.599 ERROR [task-scheduler-1]org.springframework.messaging.MessagingException: Circuit Breaker is Open for ServiceActivatorIn the preceding example, the threshold is set to 2 and halfOpenAfter is set to 12 seconds.
A new request arrives every 5 seconds.
The first two attempts invoked the service.
The third and fourth failed with an exception indicating that the circuit breaker is open.
The fifth request was attempted because the request was 15 seconds after the last failure.
The sixth attempt fails immediately because the breaker immediately went to open.
Expression Evaluating Advice
The final supplied advice class is the o.s.i.handler.advice.ExpressionEvaluatingRequestHandlerAdvice.
This advice is more general than the other two advices.
It provides a mechanism to evaluate an expression on the original inbound message sent to the endpoint.
Separate expressions are available to be evaluated, after either success or failure.
Optionally, a message containing the evaluation result, together with the input message, can be sent to a message channel.
A typical use case for this advice might be with an <ftp:outbound-channel-adapter/>, perhaps to move the file to one directory if the transfer was successful or to another directory if it fails:
The advice has properties to set an expression when successful, an expression for failures, and corresponding channels for each.
For the successful case, the message sent to the successChannel is an AdviceMessage, with the payload being the result of the expression evaluation.
An additional property, called inputMessage, contains the original message sent to the handler.
A message sent to the failureChannel (when the handler throws an exception) is an ErrorMessage with a payload of MessageHandlingExpressionEvaluatingAdviceException.
Like all MessagingException instances, this payload has failedMessage and cause properties, as well as an additional property called evaluationResult, which contains the result of the expression evaluation.
Starting with version 5.1.3, if channels are configured, but expressions are not provided, the default expression is used to evaluate to the payload of the message.
|
When an exception is thrown in the scope of the advice, by default, that exception is thrown to the caller after any failureExpression is evaluated.
If you wish to suppress throwing the exception, set the trapException property to true.
The following advice shows how to configure an advice with Java DSL:
@SpringBootApplication
public class EerhaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context = SpringApplication.run(EerhaApplication.class, args);
MessageChannel in = context.getBean("advised.input", MessageChannel.class);
in.send(new GenericMessage<>("good"));
in.send(new GenericMessage<>("bad"));
context.close();
}
@Bean
public IntegrationFlow advised() {
return f -> f.<String>handle((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);
}
}
Rate Limiter Advice
The Rate Limiter advice (RateLimiterRequestHandlerAdvice) allows to ensure that an endpoint does not get overloaded with requests.
When the rate limit is breached the request will go in a blocked state.
A typical use case for this advice might be an external service provider not allowing more than n number of request per minute.
The RateLimiterRequestHandlerAdvice implementation is fully based on the Resilience4j project and requires either RateLimiter or RateLimiterConfig injections.
Can also be configured with defaults and/or custom name.
The following example configures a rate limiter advice with one request per 1 second:
@Bean
public RateLimiterRequestHandlerAdvice rateLimiterRequestHandlerAdvice() {
return new RateLimiterRequestHandlerAdvice(RateLimiterConfig.custom()
.limitRefreshPeriod(Duration.ofSeconds(1))
.limitForPeriod(1)
.build());
}
@ServiceActivator(inputChannel = "requestChannel", outputChannel = "resultChannel",
adviceChain = "rateLimiterRequestHandlerAdvice")
public String handleRequest(String payload) {
...
}
Caching Advice
Starting with version 5.2, the CacheRequestHandlerAdvice has been introduced.
It is based on the caching abstraction in Spring Framework and aligned with the concepts and functionality provided by the @Caching annotation family.
The logic internally is based on the CacheAspectSupport extension, where proxying for caching operations is done around the AbstractReplyProducingMessageHandler.RequestHandler.handleRequestMessage method with the request Message<?> as the argument.
This advice can be configured with a SpEL expression or a Function to evaluate a cache key.
The request Message<?> is available as the root object for the SpEL evaluation context, or as the Function input argument.
By default, the payload of the request message is used for the cache key.
The CacheRequestHandlerAdvice must be configured with cacheNames, when a default cache operation is a CacheableOperation, or with a set of any arbitrary CacheOperation s.
Every CacheOperation can be configured separately or have shared options, like a CacheManager, CacheResolver and CacheErrorHandler, can be reused from the CacheRequestHandlerAdvice configuration.
This configuration functionality is similar to Spring Framework’s @CacheConfig and @Caching annotation combination.
If a CacheManager is not provided, a single bean is resolved by default from the BeanFactory in the CacheAspectSupport.
The following example configures two advices with different set of caching operations:
@Bean
public CacheRequestHandlerAdvice cacheAdvice() {
CacheRequestHandlerAdvice cacheRequestHandlerAdvice = new CacheRequestHandlerAdvice(TEST_CACHE);
cacheRequestHandlerAdvice.setKeyExpressionString("payload");
return cacheRequestHandlerAdvice;
}
@Transformer(inputChannel = "transformerChannel", outputChannel = "nullChannel", adviceChain = "cacheAdvice")
public Object transform(Message<?> message) {
...
}
@Bean
public CacheRequestHandlerAdvice cachePutAndEvictAdvice() {
CacheRequestHandlerAdvice cacheRequestHandlerAdvice = new CacheRequestHandlerAdvice();
cacheRequestHandlerAdvice.setKeyExpressionString("payload");
CachePutOperation.Builder cachePutBuilder = new CachePutOperation.Builder();
cachePutBuilder.setCacheName(TEST_PUT_CACHE);
CacheEvictOperation.Builder cacheEvictBuilder = new CacheEvictOperation.Builder();
cacheEvictBuilder.setCacheName(TEST_CACHE);
cacheRequestHandlerAdvice.setCacheOperations(cachePutBuilder.build(), cacheEvictBuilder.build());
return cacheRequestHandlerAdvice;
}
@ServiceActivator(inputChannel = "serviceChannel", outputChannel = "nullChannel",
adviceChain = "cachePutAndEvictAdvice")
public Message<?> service(Message<?> message) {
...
}
Reactive Advice
Starting with version 5.3, a ReactiveRequestHandlerAdvice can be used for request message handlers producing a Mono replies.
A BiFunction<Message<?>, Mono<?>, Publisher<?>> has to be provided for this advice and it is called from the Mono.transform() operator on a reply produced by the intercepted handleRequestMessage() method implementation.
Typically, such a Mono customization is necessary when we would like to control network fluctuations via timeout(), retry() and similar support operators.
For example when we can an HTTP request over WebFlux client, we could use below configuration to not wait for response more than 5 seconds:
.handle(WebFlux.outboundGateway("https://somehost/"),
e -> e.customizeMonoReply((message, mono) -> mono.timeout(Duration.ofSeconds(5))));
The message argument is the request message for the message handler and can be used to determine request-scope attributes.
The mono argument is the result of this message handler’s handleRequestMessage() method implementation.
A nested Mono.transform() can also be called from this function to apply, for example, a Reactive Circuit Breaker.
Custom Advice Classes
In addition to the provided advice classes described earlier, you can implement your own advice classes.
While you can provide any implementation of org.aopalliance.aop.Advice (usually org.aopalliance.intercept.MethodInterceptor), we generally recommend that you subclass o.s.i.handler.advice.AbstractRequestHandlerAdvice.
This has the benefit of avoiding the writing of low-level aspect-oriented programming code as well as providing a starting point that is specifically tailored for use in this environment.
Subclasses need to implement the doInvoke() method, the definition of which follows:
/**
* Subclasses implement this method to apply behavior to the {@link MessageHandler} callback.execute()
* invokes the handler method and returns its result, or null).
* @param callback Subclasses invoke the execute() method on this interface to invoke the handler method.
* @param target The target handler.
* @param message The message that will be sent to the handler.
* @return the result after invoking the {@link MessageHandler}.
* @throws Exception
*/
protected abstract Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception;
The callback parameter is a convenience to avoid subclasses that deal with AOP directly.
Invoking the callback.execute() method invokes the message handler.
The target parameter is provided for those subclasses that need to maintain state for a specific handler, perhaps by maintaining that state in a Map keyed by the target.
This feature allows the same advice to be applied to multiple handlers.
The RequestHandlerCircuitBreakerAdvice uses advice this to keep circuit breaker state for each handler.
The message parameter is the message sent to the handler.
While the advice cannot modify the message before invoking the handler, it can modify the payload (if it has mutable properties).
Typically, an advice would use the message for logging or to send a copy of the message somewhere before or after invoking the handler.
The return value would normally be the value returned by callback.execute().
However, the advice does have the ability to modify the return value.
Note that only AbstractReplyProducingMessageHandler instances return values.
The following example shows a custom advice class that extends AbstractRequestHandlerAdvice:
public class MyAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
// add code before the invocation
Object result = callback.execute();
// add code after the invocation
return result;
}
}
|
In addition to the For more information, see the ReflectiveMethodInvocation Javadoc. |
Other Advice Chain Elements
While the abstract class mentioned above is a convenience, you can add any Advice, including a transaction advice, to the chain.
Handling Message Advice
As discussed in the introduction to this section, advice objects in a request handler advice chain are applied to just the current endpoint, not the downstream flow (if any).
For MessageHandler objects that produce a reply (such as those that extend AbstractReplyProducingMessageHandler), the advice is applied to an internal method: handleRequestMessage() (called from MessageHandler.handleMessage()).
For other message handlers, the advice is applied to MessageHandler.handleMessage().
There are some circumstances where, even if a message handler is an AbstractReplyProducingMessageHandler, the advice must be applied to the handleMessage method.
For example, the idempotent receiver might return null, which would cause an exception if the handler’s replyRequired property is set to true.
Another example is the BoundRabbitChannelAdvice — see Strict Message Ordering.
Starting with version 4.3.1, a new HandleMessageAdvice interface and its base implementation (AbstractHandleMessageAdvice) have been introduced.
Advice objects that implement HandleMessageAdvice are always applied to the handleMessage() method, regardless of the handler type.
It is important to understand that HandleMessageAdvice implementations (such as idempotent receiver), when applied to a handlers that return responses, are dissociated from the adviceChain and properly applied to the MessageHandler.handleMessage() method.
| Because of this disassociation, the advice chain order is not honored. |
Consider the following configuration:
<some-reply-producing-endpoint ... >
<int:request-handler-advice-chain>
<tx:advice ... />
<ref bean="myHandleMessageAdvice" />
</int:request-handler-advice-chain>
</some-reply-producing-endpoint>In the preceding example, the <tx:advice> is applied to the AbstractReplyProducingMessageHandler.handleRequestMessage().
However, myHandleMessageAdvice is applied for to MessageHandler.handleMessage().
Therefore, it is invoked before the <tx:advice>.
To retain the order, you should follow the standard Spring AOP configuration approach and use an endpoint id together with the .handler suffix to obtain the target MessageHandler bean.
Note that, in that case, the entire downstream flow is within the transaction scope.
In the case of a MessageHandler that does not return a response, the advice chain order is retained.
Starting with version 5.3, the HandleMessageAdviceAdapter is provided to apply any MethodInterceptor for the MessageHandler.handleMessage() method and, therefore, the whole sub-flow.
For example, a RetryOperationsInterceptor could be applied to the whole sub-flow starting from some endpoint; this is not possible, by default, because the consumer endpoint applies advices only to the AbstractReplyProducingMessageHandler.RequestHandler.handleRequestMessage().
Transaction Support
Starting with version 5.0, a new TransactionHandleMessageAdvice has been introduced to make the whole downstream flow transactional, thanks to the HandleMessageAdvice implementation.
When a regular TransactionInterceptor is used in the <request-handler-advice-chain> element (for example, through configuring <tx:advice>), a started transaction is applied only for an internal AbstractReplyProducingMessageHandler.handleRequestMessage() and is not propagated to the downstream flow.
To simplify XML configuration, along with the <request-handler-advice-chain>, a <transactional> element has been added to all <outbound-gateway> and <service-activator> and related components.
The following example shows <transactional> in use:
<int-jdbc:outbound-gateway query="select * from things where id=:headers[id]">
<int-jdbc:transactional/>
</int-jdbc:outbound-gateway>
<bean id="transactionManager" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.transaction.TransactionManager"/>
</bean>If you are familiar with the JPA integration components, such a configuration is not new, but now we can start a transaction from any point in our flow — not only from the <poller> or a message-driven channel adapter such as JMS.
Java configuration can be simplified by using the TransactionInterceptorBuilder, and the result bean name can be used in the messaging annotations adviceChain attribute, as the following example shows:
@Bean
public ConcurrentMetadataStore store() {
return new SimpleMetadataStore(hazelcastInstance()
.getMap("idempotentReceiverMetadataStore"));
}
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
return new IdempotentReceiverInterceptor(
new MetadataStoreSelector(
message -> message.getPayload().toString(),
message -> message.getPayload().toString().toUpperCase(), store()));
}
@Bean
public TransactionInterceptor transactionInterceptor() {
return new TransactionInterceptorBuilder(true)
.transactionManager(this.transactionManager)
.isolation(Isolation.READ_COMMITTED)
.propagation(Propagation.REQUIRES_NEW)
.build();
}
@Bean
@org.springframework.integration.annotation.Transformer(inputChannel = "input",
outputChannel = "output",
adviceChain = { "idempotentReceiverInterceptor",
"transactionInterceptor" })
public Transformer transformer() {
return message -> message;
}
Note the true parameter on the TransactionInterceptorBuilder constructor.
It causes the creation of a TransactionHandleMessageAdvice, not a regular TransactionInterceptor.
Java DSL supports an Advice through the .transactional() options on the endpoint configuration, as the following example shows:
@Bean
public IntegrationFlow updatingGatewayFlow() {
return f -> f
.handle(Jpa.updatingGateway(this.entityManagerFactory),
e -> e.transactional(true))
.channel(c -> c.queue("persistResults"));
}
Advising Filters
There is an additional consideration when advising Filter advices.
By default, any discard actions (when the filter returns false) are performed within the scope of the advice chain.
This could include all the flow downstream of the discard channel.
So, for example, if an element downstream of the discard channel throws an exception and there is a retry advice, the process is retried.
Also, if throwExceptionOnRejection is set to true (the exception is thrown within the scope of the advice).
Setting discard-within-advice to false modifies this behavior and the discard (or exception) occurs after the advice chain is called.
Advising Endpoints Using Annotations
When configuring certain endpoints by using annotations (@Filter, @ServiceActivator, @Splitter, and @Transformer), you can supply a bean name for the advice chain in the adviceChain attribute.
In addition, the @Filter annotation also has the discardWithinAdvice attribute, which can be used to configure the discard behavior, as discussed in Advising Filters.
The following example causes the discard to be performed after the advice:
@MessageEndpoint
public class MyAdvisedFilter {
@Filter(inputChannel="input", outputChannel="output",
adviceChain="adviceChain", discardWithinAdvice="false")
public boolean filter(String s) {
return s.contains("good");
}
}
Ordering Advices within Advice Chain
Advice classes are “around” advices and are applied in a nested fashion. The first advice is the outermost, while the last advice is the innermost (that is, closest to the handler being advised). It is important to put the advice classes in the correct order to achieve the functionality you desire.
For example, suppose you want to add a retry advice and a transaction advice.
You may want to place the retry advice first, followed by the transaction advice.
Consequently, each retry is performed in a new transaction.
On the other hand, if you want all the attempts and any recovery operations (in the retry RecoveryCallback) to be scoped within the transaction, you could put the transaction advice first.
Advised Handler Properties
Sometimes, it is useful to access handler properties from within the advice.
For example, most handlers implement NamedComponent to let you access the component name.
The target object can be accessed through the target argument (when subclassing AbstractRequestHandlerAdvice) or invocation.getThis() (when implementing org.aopalliance.intercept.MethodInterceptor).
When the entire handler is advised (such as when the handler does not produce replies or the advice implements HandleMessageAdvice), you can cast the target object to an interface, such as NamedComponent, as shown in the following example:
String componentName = ((NamedComponent) target).getComponentName();
When you implement MethodInterceptor directly, you could cast the target object as follows:
String componentName = ((NamedComponent) invocation.getThis()).getComponentName();
When only the handleRequestMessage() method is advised (in a reply-producing handler), you need to access the full handler, which is an AbstractReplyProducingMessageHandler.
The following example shows how to do so:
AbstractReplyProducingMessageHandler handler =
((AbstractReplyProducingMessageHandler.RequestHandler) target).getAdvisedHandler();
String componentName = handler.getComponentName();
Idempotent Receiver Enterprise Integration Pattern
Starting with version 4.1, Spring Integration provides an implementation of the Idempotent Receiver Enterprise Integration Pattern.
It is a functional pattern and the whole idempotency logic should be implemented in the application.
However, to simplify the decision-making, the IdempotentReceiverInterceptor component is provided.
This is an AOP Advice that is applied to the MessageHandler.handleMessage() method and that can filter a request message or mark it as a duplicate, according to its configuration.
Previously, you could have implemented this pattern by using a custom MessageSelector in a <filter/> (see Filter), for example.
However, since this pattern really defines the behavior of an endpoint rather than being an endpoint itself, the idempotent receiver implementation does not provide an endpoint component.
Rather, it is applied to endpoints declared in the application.
The logic of the IdempotentReceiverInterceptor is based on the provided MessageSelector and, if the message is not accepted by that selector, it is enriched with the duplicateMessage header set to true.
The target MessageHandler (or downstream flow) can consult this header to implement the correct idempotency logic.
If the IdempotentReceiverInterceptor is configured with a discardChannel or throwExceptionOnRejection = true, the duplicate message is not sent to the target MessageHandler.handleMessage().
Rather, it is discarded.
If you want to discard (do nothing with) the duplicate message, the discardChannel should be configured with a NullChannel, such as the default nullChannel bean.
To maintain state between messages and provide the ability to compare messages for the idempotency, we provide the MetadataStoreSelector.
It accepts a MessageProcessor implementation (which creates a lookup key based on the Message) and an optional ConcurrentMetadataStore (Metadata Store).
See the MetadataStoreSelector Javadoc for more information.
You can also customize the value for ConcurrentMetadataStore by using an additional MessageProcessor.
By default, MetadataStoreSelector uses the timestamp message header.
Normally, the selector selects a message for acceptance if there is no existing value for the key.
In some cases, it is useful to compare the current and new values for a key, to determine whether the message should be accepted.
Starting with version 5.3, the compareValues property is provided which references a BiPredicate<String, String>; the first parameter is the old value; return true to accept the message and replace the old value with the new value in the MetadataStore.
This can be useful to reduce the number of keys; for example, when processing lines in a file, you can store the file name in the key and the current line number in the value.
Then, after a restart, you can skip lines that have already been processed.
See Idempotent Downstream Processing a Split File for an example.
For convenience, the MetadataStoreSelector options are configurable directly on the <idempotent-receiver> component.
The following listing shows all the possible attributes:
<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)
compare-values="" (10)
throw-exception-on-rejection="" /> (11)| 1 | The ID of the IdempotentReceiverInterceptor bean.
Optional. |
| 2 | Consumer endpoint name(s) or pattern(s) to which this interceptor is applied.
Separate names (patterns) with commas (,), such as endpoint="aaa, bbb*, ccc, *ddd, eee*fff".
Endpoint bean names matching these patterns are then used to retrieve the target endpoint’s MessageHandler bean (using its .handler suffix), and the IdempotentReceiverInterceptor is applied to those beans.
Required. |
| 3 | A MessageSelector bean reference.
Mutually exclusive with metadata-store and key-strategy (key-expression).
When selector is not provided, one of key-strategy or key-strategy-expression is required. |
| 4 | Identifies the channel to which to send a message when the IdempotentReceiverInterceptor does not accept it.
When omitted, duplicate messages are forwarded to the handler with a duplicateMessage header.
Optional. |
| 5 | A ConcurrentMetadataStore reference.
Used by the underlying MetadataStoreSelector.
Mutually exclusive with selector.
Optional.
The default MetadataStoreSelector uses an internal SimpleMetadataStore that does not maintain state across application executions. |
| 6 | A MessageProcessor reference.
Used by the underlying MetadataStoreSelector.
Evaluates an idempotentKey from the request message.
Mutually exclusive with selector and key-expression.
When a selector is not provided, one of key-strategy or key-strategy-expression is required. |
| 7 | A SpEL expression to populate an ExpressionEvaluatingMessageProcessor.
Used by the underlying MetadataStoreSelector.
Evaluates an idempotentKey by using the request message as the evaluation context root object.
Mutually exclusive with selector and key-strategy.
When a selector is not provided, one of key-strategy or key-strategy-expression is required. |
| 8 | A MessageProcessor reference.
Used by the underlying MetadataStoreSelector.
Evaluates a value for the idempotentKey from the request message.
Mutually exclusive with selector and value-expression.
By default, the 'MetadataStoreSelector' uses the 'timestamp' message header as the Metadata 'value'. |
| 9 | A SpEL expression to populate an ExpressionEvaluatingMessageProcessor.
Used by the underlying MetadataStoreSelector.
Evaluates a value for the idempotentKey by using the request message as the evaluation context root object.
Mutually exclusive with selector and value-strategy.
By default, the 'MetadataStoreSelector' uses the 'timestamp' message header as the metadata 'value'. |
| 10 | A reference to a BiPredicate<String, String> bean which allows you to optionally select a message by comparing the old and new values for the key; null by default. |
| 11 | Whether to throw an exception if the IdempotentReceiverInterceptor rejects the message.
Defaults to false.
It is applied regardless of whether or not a discard-channel is provided. |
For Java configuration, Spring Integration provides the method-level @IdempotentReceiver annotation.
It is used to mark a method that has a messaging annotation (@ServiceActivator, @Router, and others) to specify which `IdempotentReceiverInterceptor objects are applied to this endpoint.
The following example shows how to use the @IdempotentReceiver annotation:
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
return new IdempotentReceiverInterceptor(new MetadataStoreSelector(m ->
m.getHeaders().get(INVOICE_NBR_HEADER)));
}
@Bean
@ServiceActivator(inputChannel = "input", outputChannel = "output")
@IdempotentReceiver("idempotentReceiverInterceptor")
public MessageHandler myService() {
....
}
When you use the Java DSL, you can add the interceptor to the endpoint’s advice chain, as the following example shows:
@Bean
public IntegrationFlow flow() {
...
.handle("someBean", "someMethod",
e -> e.advice(idempotentReceiverInterceptor()))
...
}
The IdempotentReceiverInterceptor is designed only for the MessageHandler.handleMessage(Message<?>) method.
Starting with version 4.3.1, it implements HandleMessageAdvice, with the AbstractHandleMessageAdvice as a base class, for better dissociation.
See Handling Message Advice for more information.
|
Logging Channel Adapter
The <logging-channel-adapter> is often used in conjunction with a wire tap, as discussed in Wire Tap.
However, it can also be used as the ultimate consumer of any flow.
For example, consider a flow that ends with a <service-activator> that returns a result, but you wish to discard that result.
To do that, you could send the result to NullChannel.
Alternatively, you can route it to an INFO level <logging-channel-adapter>.
That way, you can see the discarded message when logging at INFO level but not see it when logging at (for example) the WARN level.
With a NullChannel, you would see only the discarded message when logging at the DEBUG level.
The following listing shows all the possible attributes for the logging-channel-adapter element:
<int:logging-channel-adapter
channel="" (1)
level="INFO" (2)
expression="" (3)
log-full-message="false" (4)
logger-name="" /> (5)| 1 | The channel connecting the logging adapter to an upstream component. |
| 2 | The logging level at which messages sent to this adapter will be logged.
Default: INFO. |
| 3 | A SpEL expression representing exactly what parts of the message are logged.
Default: payload — only the payload is logged.
if log-full-message is specified, this attribute cannot be specified. |
| 4 | When true, the entire message (including headers) is logged.
Default: false — only the payload is logged.
This attribute cannot be specified if expression is specified. |
| 5 | Specifies the name of the logger (known as category in log4j).
Used to identify log messages created by this adapter.
This enables setting the log name (in the logging subsystem) for individual adapters.
By default, all adapters log under the following name: org.springframework.integration.handler.LoggingHandler. |
Using Java Configuration
The following Spring Boot application shows an example of configuring the LoggingHandler by using Java configuration:
@SpringBootApplication
public class LoggingJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(LoggingJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToLogger("foo");
}
@Bean
@ServiceActivator(inputChannel = "logChannel")
public LoggingHandler logging() {
LoggingHandler adapter = new LoggingHandler(LoggingHandler.Level.DEBUG);
adapter.setLoggerName("TEST_LOGGER");
adapter.setLogExpressionString("headers.id + ': ' + payload");
return adapter;
}
@MessagingGateway(defaultRequestChannel = "logChannel")
public interface MyGateway {
void sendToLogger(String data);
}
}
Configuring with the Java DSL
The following Spring Boot application shows an example of configuring the logging channel adapter by using the Java DSL:
@SpringBootApplication
public class LoggingJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(LoggingJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToLogger("foo");
}
@Bean
public IntegrationFlow loggingFlow() {
return IntegrationFlow.from(MyGateway.class)
.log(LoggingHandler.Level.DEBUG, "TEST_LOGGER",
m -> m.getHeaders().getId() + ": " + m.getPayload());
}
@MessagingGateway
public interface MyGateway {
void sendToLogger(String data);
}
}
java.util.function Interfaces Support
Starting with version 5.1, Spring Integration provides direct support for interfaces in the java.util.function package.
All messaging endpoints, (Service Activator, Transformer, Filter, etc.) can now refer to Function (or Consumer) beans.
The Messaging Annotations can be applied directly on these beans similar to regular MessageHandler definitions.
For example if you have this Function bean definition:
@Configuration
public class FunctionConfiguration {
@Bean
public Function<String, String> functionAsService() {
return String::toUpperCase;
}
}
You can use it as a simple reference in an XML configuration file:
<service-activator input-channel="processorViaFunctionChannel" ref="functionAsService"/>When we configure our flow with Messaging Annotations, the code is straightforward:
@Bean
@Transformer(inputChannel = "functionServiceChannel")
public Function<String, String> functionAsService() {
return String::toUpperCase;
}
When the function returns an array, Collection (essentially, any Iterable), Stream or Reactor Flux, @Splitter can be used on such a bean to perform iteration over the result content.
The java.util.function.Consumer interface can be used for an <int:outbound-channel-adapter> or, together with the @ServiceActivator annotation, to perform the final step of a flow:
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
public Consumer<Message<?>> messageConsumerAsService() {
// Has to be an anonymous class for proper type inference
return new Consumer<Message<?>>() {
@Override
public void accept(Message<?> e) {
collector().add(e);
}
};
}
Also, pay attention to the comment in the code snippet above: if you would like to deal with the whole message in your Function/Consumer you cannot use a lambda definition.
Because of Java type erasure we cannot determine the target type for the apply()/accept() method call.
The java.util.function.Supplier interface can simply be used together with the @InboundChannelAdapter annotation, or as a ref in an <int:inbound-channel-adapter>:
@Bean
@InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
public Supplier<String> pojoSupplier() {
return () -> "foo";
}
With the Java DSL we just need to use a reference to the function bean in the endpoint definitions.
Meanwhile, an implementation of the Supplier interface can be used as regular MessageSource definition:
@Bean
public Function<String, String> toUpperCaseFunction() {
return String::toUpperCase;
}
@Bean
public Supplier<String> stringSupplier() {
return () -> "foo";
}
@Bean
public IntegrationFlow supplierFlow() {
return IntegrationFlow.from(stringSupplier())
.transform(toUpperCaseFunction())
.channel("suppliedChannel")
.get();
}
This function support is useful when used together with the Spring Cloud Function framework, where we have a function catalog and can refer to its member functions from an integration flow definition.
Kotlin Support
The Framework also has been improved to support Kotlin lambdas for functions, so now you can use a combination of the Kotlin language and Spring Integration flow definitions:
@Bean
@Transformer(inputChannel = "functionServiceChannel")
fun kotlinFunction(): (String) -> String {
return { it.toUpperCase() }
}
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
fun kotlinConsumer(): (Message<Any>) -> Unit {
return { print(it) }
}
@Bean
@InboundChannelAdapter(value = "counterChannel",
poller = Poller(fixedRate = "10", maxMessagesPerPoll = "1"))
fun kotlinSupplier(): () -> String {
return { "baz" }
}
Kotlin Coroutines
Starting with version 6.0, Spring Integration provides support for Kotlin Coroutines.
Now suspend functions and kotlinx.coroutines.Deferred & kotlinx.coroutines.flow.Flow return types can be used for service methods:
@ServiceActivator(inputChannel = "suspendServiceChannel", outputChannel = "resultChannel")
suspend fun suspendServiceFunction(payload: String) = payload.uppercase()
@ServiceActivator(inputChannel = "flowServiceChannel", outputChannel = "resultChannel", async = "true")
fun flowServiceFunction(payload: String) =
flow {
for (i in 1..3) {
emit("$payload #$i")
}
}
The framework treats them as Reactive Streams interactions and uses ReactiveAdapterRegistry to convert to respective Mono and Flux reactor types.
Such a function reply is processed then in the reply channel, if it is a ReactiveStreamsSubscribableChannel, or as a result of CompletableFuture in the respective callback.
The functions with Flow result are not async by default on the @ServiceActivator, so Flow instance is produced as a reply message payload.
It is the target application’s responsibility to process this object as a coroutine or convert it to Flux, respectively.
|
The @MessagingGateway interface methods also can be marked with a suspend modifier when declared in Kotlin.
The framework utilizes a Mono internally to perform request-reply using the downstream flow.
Such a Mono result is processed by the MonoKt.awaitSingleOrNull() API internally to fulfil a kotlin.coroutines.Continuation argument fo the called suspend function of the gateway:
@MessagingGateway(defaultRequestChannel = "suspendRequestChannel")
interface SuspendFunGateway {
suspend fun suspendGateway(payload: String): String
}
This method has to be called as a coroutine according to Kotlin language requirements:
@Autowired
private lateinit var suspendFunGateway: SuspendFunGateway
fun someServiceMethod() {
runBlocking {
val reply = suspendFunGateway.suspendGateway("test suspend gateway")
}
}