# 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](#endpoint-namespace) 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](./channel.html#channel), 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](https://www.enterpriseintegrationpatterns.com/PollingConsumer.html) and [event-driven consumers](https://www.enterpriseintegrationpatterns.com/EventDrivenConsumer.html). 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`](./channel.html#channel-interfaces-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](./polling-consumer.html#polling-consumer) and [Channel Adapter](./channel-adapter.html#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(30, TimeUnit.SECONDS)); ``` The `PeriodicTrigger` is typically defined with a simple interval (in milliseconds) 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(1000); trigger.setInitialDelay(5000); 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](https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/scheduling/support/CronTrigger.html) 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](#endpoint-namespace). 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`: ``` (12) (13) (14) ``` |**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](./channel-adapter.html#channel-adapter-namespace-inbound) 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](#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](#transaction-support) for further information.
Optional. | |**14**| Pollers can be made transactional.
See [AOP Advice chains](#aop-advice-chains) for further information.
Optional. | ##### Examples A simple interval-based poller with a 1-second interval can be configured as follows: ``` ``` 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: ``` ``` 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: ``` ``` | |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: Java DSL ``` @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 IntegrationFlows.from(MessageChannels.queue("pollable")) .transform(transformer) // No 'poller' attribute because there is a default global poller .channel("output") .get(); } ``` Java ``` @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) { ... } ``` Kotlin DSL ``` @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") } ``` XML ``` ``` ###### 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 `` sub-element. The following example shows the available attributes: ``` ``` For more information, see [Poller Transaction Support](./transactions.html#transaction-poller). ##### 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`: ``` ``` For more information on how to implement the `MethodInterceptor` interface, see the [AOP sections of the Spring Framework Reference Guide](https://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#aop-api). 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 `` child element cannot be specified.
Instead, declare a `` bean and add it to the ``.
See [Poller Transaction Support](./transactions.html#transaction-poller) 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 `` 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 `` element and then provide one or more of the properties shown in the following example: ``` ``` 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](./configuration.html#namespace-taskscheduler)). 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](#endpoint-pollingconsumer), 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: ``` ``` 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](https://github.com/SpringSource/spring-integration-samples/tree/main/intermediate) 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`](https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/scheduling/Trigger.html) interface. The sample’s trigger is based on Spring’s [`PeriodicTrigger`](https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/scheduling/support/PeriodicTrigger.html) 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: * [org.springframework.core.convert.converter.Converter](https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/core/convert/converter/Converter.html) * [org.springframework.core.convert.converter.GenericConverter](https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/core/convert/converter/package-summary.html) * [org.springframework.core.convert.converter.ConverterFactory](https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/core/convert/converter/ConverterFactory.html) When you have implemented your converter, you can register it with convenient namespace support, as the following example shows: ``` ``` Alternately, you can use an inner bean, as the following example shows: ``` ``` 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 { 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 `conversionService` bean (see [Configuring a ConversionService](https://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#core-convert-Spring-config) chapter).
This service is used, when needed, to perform appropriate conversions during bean creation and configuration.

In contrast, the `integrationConversionService` is used for runtime conversions.
These uses are quite different.
Converters that are intended for use when wiring bean constructor arguments and properties may produce unintended results if used at runtime for Spring Integration expression evaluation against messages within data type channels, payload type transformers, and so on.

However, if you do want to use the Spring `conversionService` as the Spring Integration `integrationConversionService`, you can configure an alias in the application context, as the following example shows:

```

```

In this case, the converters provided by the `conversionService` are available for Spring Integration runtime conversion.| |---|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| #### 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): 1. [`MappingJackson2MessageConverter`](https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/jms/support/converter/MappingJackson2MessageConverter.html) if the Jackson processor is present on the classpath 2. [`ByteArrayMessageConverter`](https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/messaging/converter/ByteArrayMessageConverter.html) 3. [`ObjectStringMessageConverter`](https://docs.spring.io/spring-integration/docs/current/api//org/springframework/integration/support/converter/ObjectStringMessageConverter.html) 4. [`GenericMessageConverter`](https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/messaging/converter/GenericMessageConverter.html) 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 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 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](#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: ``` ``` 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”](https://docs.spring.io/spring/docs/current/spring-framework-reference/integration.html#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: ``` ``` 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](https://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#beans-inner-beans) 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`: ``` ``` You can assign endpoints to roles using XML, Java configuration, or programmatically. The following example shows how to configure endpoint roles with XML: ``` ``` 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` 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 getRoles() (1) public boolean allEndpointsRunning(String role) (2) public boolean noEndpointsRunning(String role) (3) public Map 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](./file.html#file-reading)). 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 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](./zookeeper.html#zk-leadership) for more information about leadership election and events that use Zookeeper. ### 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: ``` ``` With this configuration defined, the `cafeService` can now be injected into other beans, and the code that invokes the methods on that proxied instance of the `Cafe` interface has no awareness of the Spring Integration API. The general approach is similar to that of Spring Remoting (RMI, HttpInvoker, and so on). See the [“Samples”](./samples.html#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](#gateway-no-response). The defaults can be overridden for individual methods. See [Gateway Configuration with Annotations and XML](#gateway-configuration-annotations). #### 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 of the 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](./channel.html#channel-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: ``` ``` 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: ``` ``` 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 `` 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 `` element), the `@Payload` value is ignored.| |---|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| ##### Expressions and “Global” Headers The `
` 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. These two expression evaluation context variables are deprecated since version 5.2: * \#args: An `Object[]` containing the method arguments * \#gatewayMethod: The object (derived from `java.reflect.Method`) that represents the method in the `service-interface` that was invoked. A header containing this variable can be used later in the flow (for example, for routing). For example, if you wish to route on the simple method name, you might add a header with the following expression: `#gatewayMethod.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, `` 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 `` 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 `` elements are not allowed on the gateway. Similarly, the `payload-expression` attribute and `
` elements are not allowed on any `` 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 map); void payloadAndHeaderMapWithAnnotations(@Payload String s, @Headers Map map); void headerValuesAndPayloadWithAnnotations(@Header("k1") String x, @Payload String s, @Header("k2") String y); void mapOnly(Map map); // the payload is the map and no custom headers are added void twoMapsAndOneAnnotatedWithPayload(@Payload Map payload, Map 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 m1, Map m2); // invalid void twoPayloads(@Payload String s1, @Payload String s2); // invalid void payloadAndHeaderAnnotationsOnSameParameter(@Payload @Header("x") String s); // invalid void payloadAndHeadersAnnotationsOnSameParameter(@Payload @Headers Map 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` variable, as the following example shows: ``` ``` #### `@MessagingGateway` Annotation Starting with version 4.0, gateway service interfaces can be marked with a `@MessagingGateway` annotation instead of requiring the definition of a `` xml element for configuration. The following pair of examples compares the two approaches for configuring the same 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 fine the `@MessagingGateway` annotation on the interfaces and register `GatewayProxyFactoryBean` instances for them.
See also [Annotation Support](./configuration.html#annotations).| |---|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| 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. | |If you have no XML configuration, the `@EnableIntegration` annotation is required on at least one `@Configuration` class.
See [Configuration and `@EnableIntegration`](./overview.html#configuration-enable-integration) 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 retrieveOpenOrders(); } ``` You can also use the `@Gateway` annotation. ``` public interface Cafe { @Gateway(payloadExpression = "new java.util.Date()") List retrieveOpenOrders(); } ``` | |If both annotations are present (and the `payloadExpression` is provided), `@Gateway` wins.| |---|-------------------------------------------------------------------------------------------| Also see [Gateway Configuration with Annotations and XML](#gateway-configuration-annotations). 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 `` 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-activator ``` If 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`: ``` ``` 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` Annotation](#messaging-gateway-annotation)`), you can use use the `errorChannel` 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 `
` 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 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](https://github.com/spring-projects/spring-integration-samples/tree/main/intermediate/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](./aggregator.html#agg-and-group-to)) 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 `` 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 variable is available. When configuring with XML, the timeout attributes can be a long value or a SpEL expression, as the following example shows: ``` ``` #### 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](#gateway-proxy), the `GatewayProxyFactoryBean` provides a convenient way to expose a proxy over a service-interface giving you POJO-based access to a messaging system (based on objects in your own domain, primitives/Strings, or other objects). However, when a gateway is exposed through simple POJO methods that return values, it implies that, for each request message (generated when the method is invoked), there must be a reply message (generated when the method has returned). Since messaging systems are naturally asynchronous, you may not always be able to guarantee the contract where “for each request, there will always be be a reply”. Spring Integration 2.0 introduced support for an asynchronous gateway, which offers a convenient way to initiate flows when you may not know if a reply is expected or how long it takes for replies to arrive. To handle these types of scenarios, Spring Integration uses `java.util.concurrent.Future` instances to support an asynchronous gateway. From the XML configuration, nothing changes, and you still define asynchronous gateway the same way as you define a regular gateway, as the following example shows: ``` ``` However, the gateway interface (a service interface) is a little different, as follows: ``` public interface MathServiceGateway { Future 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 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](https://github.com/spring-projects/spring-integration-samples/tree/main/intermediate/async-gateway) sample in the Spring Integration samples. ##### `ListenableFuture` Starting with version 4.1, asynchronous gateway methods can also return `ListenableFuture` (introduced in Spring Framework 4.0). These return types let you provide a callback, which is invoked when the result is available (or an exception occurs). When the gateway detects this return type and the [task executor](#gateway-asynctaskexecutor) is an `AsyncListenableTaskExecutor`, the executor’s `submitListenable()` method is invoked. The following example shows how to use a `ListenableFuture`: ``` ListenableFuture result = this.asyncGateway.async("something"); result.addCallback(new ListenableFutureCallback() { @Override public void onSuccess(String result) { ... } @Override public void onFailure(Throwable t) { ... } }); ``` ##### `AsyncTaskExecutor` By default, the `GatewayProxyFactoryBean` uses `org.springframework.core.task.SimpleAsyncTaskExecutor` when submitting internal `AsyncInvocationTask` instances for any gateway method whose return type is a `Future`. However, the `async-executor` attribute in the `` element’s configuration lets you provide a reference to any implementation of `java.util.concurrent.Executor` available within the Spring application context. The (default) `SimpleAsyncTaskExecutor` supports both `Future` and `ListenableFuture` return types, returning `FutureTask` or `ListenableFutureTask` respectively. See [`CompletableFuture`](#gw-completable-future). 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 a `CompletableFuture` to the caller.`CompletableFuture.supplyAsync(Supplier supplier, Executor executor)` is used to create the future. * When the async executor is explicitly set to `null` and the return type is `CompletableFuture` or the return type is a subclass of `CompletableFuture`, the flow is invoked on the caller’s thread. In this scenario, the downstream flow is expected to return a `CompletableFuture` of the appropriate type. ###### Usage Scenarios In the following scenario, the caller thread returns immediately with a `CompletableFuture`, which is completed when the downstream flow replies to the gateway (with an `Invoice` object). ``` CompletableFuture order(Order order); ``` ``` ``` In the following scenario, the caller thread returns with a `CompletableFuture` 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 order(Order order); ``` ``` ``` In the following scenario, the caller thread returns with a `CompletableFuture` 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 order(Order order); ``` ``` ``` `CompletableFuture` instances can be used to perform additional manipulation on the reply, as the following example shows: ``` CompletableFuture 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](https://projectreactor.io/) with gateway interface methods, using a [`Mono`](https://github.com/reactor/reactor-core) return type. The internal `AsyncInvocationTask` is wrapped in a `Mono.fromCallable()`. A `Mono` can be used to retrieve the result later (similar to a `Future`), or you can consume from it with the dispatcher by invoking your `Consumer` when the result is returned to the gateway. | |The `Mono` is not immediately flushed by the framework.
Consequently, the underlying message flow is not started before the gateway method returns (as it is with a `Future` `Executor` task).
The flow starts when the `Mono` is subscribed to.
Alternatively, the `Mono` (being a “Composable”) might be a part of Reactor stream, when the `subscribe()` is related to the entire `Flux`.
The following example shows how to create a gateway with Project Reactor:| |---|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| ``` @MessagingGateway public static interface TestGateway { @Gateway(requestChannel = "promiseChannel") Mono multiply(Integer value); } ... @ServiceActivator(inputChannel = "promiseChannel") public Integer multiply(Integer value) { return value * 2; } ... Flux.just("1", "2", "3", "4", "5") .map(Integer::parseInt) .flatMap(this.testGateway::multiply) .collectList() .subscribe(integers -> ...); ``` Another example that uses Project Reactor is a simple callback scenario, as the following example shows: ``` Mono mono = service.process(myOrder); mono.subscribe(invoice -> handleInvoice(invoice)); ``` The calling thread continues, with `handleInvoice()` being called when the flow completes. ##### Downstream Flows Returning an Asynchronous Type As mentioned in the `ListenableFuture` section above, if you wish some downstream component to return a message with an async payload (`Future`, `Mono`, and others), you must explicitly set the async executor to `null` (or `""` when using XML configuration). The flow is then invoked on the caller thread and the result can be retrieved later. ##### `void` Return Type Unlike the return types mentioned earlier, when the method return type is `void`, the framework cannot implicitly determine that you wish the downstream flow to run asynchronously, with the caller thread returning immediately. In this case, you must annotate the interface method with `@Async`, as the following example shows: ``` @MessagingGateway public interface MyGateway { @Gateway(requestChannel = "sendAsyncChannel") @Async void sendAsync(String payload); } ``` Unlike the `Future` return types, there is no way to inform the caller if some exception is thrown by the flow, unless some custom `TaskExecutor` (such as an `ErrorHandlingTaskExecutor`) is associated with the `@Async` annotation. #### Gateway Behavior When No response Arrives As [explained earlier](#gateway-proxy), the gateway provides a convenient way of interacting with a messaging system through POJO method invocations. However, a typical method invocation, which is generally expected to always return (even with an Exception), might not always map one-to-one to message exchanges (for example, a reply message might not arrive — the equivalent to a method not returning). The rest of this section covers various scenarios and how to make the gateway behave more predictably. Certain attributes can be configured to make synchronous gateway behavior more predictable, but some of them might not always work as you might expect. One of them is `reply-timeout` (at the method level or `default-reply-timeout` at the gateway level). We examine the `reply-timeout` attribute to see how it can and cannot influence the behavior of the synchronous gateway in various scenarios. We examine a single-threaded scenario (all components downstream are connected through a direct channel) and multi-threaded scenarios (for example, somewhere downstream you may have a pollable or executor channel that breaks the single-thread boundary). ##### Long-running Process Downstream Sync Gateway, single-threaded If a component downstream is still running (perhaps because of an infinite loop or a slow service), setting a `reply-timeout` has no effect, and the gateway method call does not return until the downstream service exits (by returning or throwing an exception). Sync Gateway, multi-threaded If a component downstream is still running (perhaps because of an infinite loop or a slow service) in a multi-threaded message flow, setting the `reply-timeout` has an effect by allowing gateway method invocation to return once the timeout has been reached, because the `GatewayProxyFactoryBean` polls on the reply channel, waiting for a message until the timeout expires. However, if the timeout has been reached before the actual reply was produced, it could result in a 'null' return from the gateway method. You should understand that the reply message (if produced) is sent to a reply channel after the gateway method invocation might have returned, so you must be aware of that and design your flow with it in mind. ##### Downstream Component Returns 'null' Sync Gateway — single-threaded If a component downstream returns 'null' and no `reply-timeout` has been configured, the gateway method call hangs indefinitely, unless a `reply-timeout` has been configured or the `requires-reply` attribute has been set on the downstream component (for example, a service activator) that might return 'null'. In this case, an exception would be thrown and propagated to the gateway. Sync Gateway — multi-threaded The behavior is the same as the previous case. ##### Downstream Component Return Signature is 'void' While Gateway Method Signature Is Non-void Sync Gateway — single-threaded If a component downstream returns 'void' and no `reply-timeout` has been configured, the gateway method call hangs indefinitely unless a `reply-timeout` has been configured. Sync Gateway — multi-threaded The behavior is the same as the previous case. ##### Downstream Component Results in Runtime Exception Sync Gateway — single-threaded If a component downstream throws a runtime exception, the exception is propagated through an error message back to the gateway and re-thrown. Sync Gateway — multi-threaded The behavior is the same as the previous case. | |You should understand that, by default, `reply-timeout` is unbounded.
Consequently, if you do not explicitly set the `reply-timeout`, your gateway method invocation might hang indefinitely.
So, to make sure you analyze your flow and if there is even a remote possibility of one of these scenarios to occur, you should set the `reply-timeout` attribute to a "'safe'" value.
Even better, you can set the `requires-reply` attribute of the downstream component to 'true' to ensure a timely response, as produced by the throwing of an exception as soon as that downstream component returns null internally.
However you should also realize that there are some scenarios (see [the first one](#long-running-process-downstream)) where `reply-timeout` does not help.
That means it is also important to analyze your message flow and decide when to use a synchronous gateway rather than an asynchrnous gateway.
As [described earlier](#async-gateway), 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 `` 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](./dsl.html#integration-flow-as-gateway) in the Java DSL chapter for options to define gateways through `IntegrationFlows`. ### 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: ``` ``` The preceding configuration selects all the methods from the `exampleHandler` that meet one of the messaging requirements, which are as follows: * annotated with `@ServiceActivator` * is `public` * not return `void` if `requiresReply == 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](./endpoint.html#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: ``` ``` 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: ``` ``` If the method returns a result and no `output-channel` is defined, the framework then checks the request message’s `replyChannel` header value. If that value is available, it then checks its type. If it is a `MessageChannel`, the reply message is sent to that channel. If it is a `String`, the endpoint tries to resolve the channel name to a channel instance. If the channel cannot be resolved, a `DestinationResolutionException` is thrown. It it can be resolved, the message is sent there. If the request message does not have a `replyChannel` header and the `reply` object is a `Message`, its `replyChannel` header is consulted for a target destination. This is the technique used for request-reply messaging in Spring Integration, and it is also an example of the return address pattern. If your method returns a result and you want to discard it and end the flow, you should configure the `output-channel` to send to a `NullChannel`. For convenience, the framework registers one with the name, `nullChannel`. See [Special Channels](./channel.html#channel-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](./configuration.html#annotations). | |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 payload, @Header(value="foo", required=false) String foo1, @Header(value="foo") Optional 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 `` definitions. However, if the custom service activator handler implementation is only used within a single definition of the ``, you can provide an inner bean definition, as the following example shows: ``` ``` | |Using both the `ref` attribute and an inner handler definition in the same `` 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 `` configuration type.
If you inadvertently reference the same message handler from multiple beans, you get a configuration exception.| |---|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| ##### Since Spring Integration 2.0, service activators can also benefit from [SpEL](https://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#expressions). 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: ``` ``` 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: ``` ``` 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](./dsl.html#java-dsl-handle) 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 `ListenableFuture`, the default action is to send that as the payload of the message sent to the output (or reply) channel. Starting with version 4.3, you can now set the `async` attribute to `true` (by using `setAsync(true)` when using Java configuration). If the service returns a `ListenableFuture` when this the `async` attribute is set to `true`, the calling thread is released immediately and the reply message is sent on the thread (from within your service) that completes the future. This is particularly advantageous for long-running services that use a `PollableChannel`, because the poller thread is released to perform other services within the framework. If the service completes the future with an `Exception`, normal error processing occurs. An `ErrorMessage` is sent to the `errorChannel` message header, if present. Otherwise, an `ErrorMessage` is sent to the default `errorChannel` (if available). #### 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](./transformer.html#transformer), 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 `` 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: ``` ``` 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: Java DSL ``` @Bean public IntegrationFlow flow() { return IntegrationFlows.from("input") .delay("delayer.messageGroupId", d -> d .defaultDelay(3_000L) .delayExpression("headers['delay']")) .channel("output") .get(); } ``` Kotlin DSL ``` @Bean fun flow() = integrationFlow("input") { delay("delayer.messageGroupId") { defaultDelay(3000L) delayExpression("headers['delay']") } channel("output") } ``` Java ``` @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; } ``` XML ``` ``` 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 `.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 `headers['delay']`.
This is the SpEL `Indexer` syntax to access a `Map` element (`MessageHeaders` implements `Map`).
It invokes: `headers.get("delay")`.
For simple map element names (that do not contain '.') you can also use the SpEL “dot accessor” syntax, where the header expression shown earlier can be specified as `headers.delay`.
However, different results are achieved if the header is missing.
In the first case, the expression evaluates to `null`.
The second results in something similar to the following:

```
org.springframework.expression.spel.SpelEvaluationException: EL1008E:(pos 8):
Field or property 'delay' cannot be found on object of type 'org.springframework.messaging.MessageHeaders'
```

Consequently, if there is a possibility of the header being omitted and you want to fall back to the default delay, it is generally more efficient (and recommended) to use the indexer syntax instead of dot property accessor syntax, because detecting the null is faster than catching an exception.| |---|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| The delayer delegates to an instance of Spring’s `TaskScheduler` abstraction. The default scheduler used by the delayer is the `ThreadPoolTaskScheduler` instance provided by Spring Integration on startup. See [Configuring the Task Scheduler](./configuration.html#namespace-taskscheduler). 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: ``` ``` | |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 `` 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-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 `` 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 `` can be enriched with either of two mutually exclusive elements: `` and ``. 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 ``. The `` element defines a simple advice chain that has only the transactional advice. The following example shows an `advice-chain` within a ``: ``` ``` 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 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](./system-management.html#system-management-chapter).| |---|--------------------------------------------------------------------------------------------------------------------------------------------------------| 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](https://www.jcp.org/en/jsr/detail?id=223), 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](https://docs.oracle.com/javase/8/docs/technotes/guides/scripting/prog_guide/api.html). | |Starting with Java 11, the Nashorn JavaScript Engine has been deprecated with possible removal in Java 15.
It is recommended to reconsider in favor of other scripting language from now on.| |---|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| You need to include this dependency into your project: Maven ``` org.springframework.integration spring-integration-scripting 5.5.9 ``` Gradle ``` compile "org.springframework.integration:spring-integration-scripting:5.5.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: Maven ``` org.jetbrains.kotlin kotlin-script-util runtime org.jetbrains.kotlin kotlin-compiler-embeddable runtime org.jetbrains.kotlin kotlin-scripting-compiler-embeddable runtime ``` Gradle ``` 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](https://groovy-lang.org/) and [JRuby](https://www.jruby.org) 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](./groovy.html#groovy) 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: Java DSL ``` @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")); } ``` XML ``` ``` 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: Java DSL ``` Scripts.processor(...).refreshCheckDelay(5000) } ``` XML ``` ``` 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: Java DSL ``` Scripts.processor(...).refreshCheckDelay(0) } ``` XML ``` ``` 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: Java DSL ``` Scripts.processor(...).refreshCheckDelay(-1) } ``` XML ``` ``` | |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 `` elements (or `ScriptSpec.variables()` option), as the following example shows: Java DSL ``` Scripts.processor("foo/bar/MyScript.py") .variables(Map.of("var1", "thing1", "var2", "thing2", "date", date)) } ``` XML ``` ``` 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: ``` ``` 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 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: Java DSL ``` Scripts.processor("foo/bar/MyScript.groovy") .variableGenerator(new foo.bar.MyScriptVariableGenerator()) } ``` XML ``` ``` If a `script-variable-generator` is not provided, script components use `DefaultScriptVariableGenerator`, which merges any provided `` elements with `payload` and `headers` variables from the `Message` in its `generateScriptVariables(Message)` method. | |You cannot provide both the `script-variable-generator` attribute and `` element(s).
They are mutually exclusive.| |---|-------------------------------------------------------------------------------------------------------------------------------| ### 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](https://groovy-lang.org/). You need to include this dependency into your project: Maven ``` org.springframework.integration spring-integration-groovy 5.5.9 ``` Gradle ``` compile "org.springframework.integration:spring-integration-groovy:5.5.9" ``` #### 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](./scripting.html#scripting) 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: Example 1. 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 `