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

[[delayer-namespace]]
== Configuring a Delayer

The `<delayer>` element is used to delay the message flow between two message channels.
As with the other endpoints, you can provide the 'input-channel' and 'output-channel' attributes, but the delayer also has 'default-delay' and 'expression' attributes (and the 'expression' element) that determine the number of milliseconds by which each message should be delayed.
The following example delays all messages by three seconds:

[source,xml]
----
<int:delayer id="delayer" input-channel="input"
             default-delay="3000" output-channel="output"/>
----

If you need to determine the delay for each message, you can also provide the SpEL expression by using the 'expression' attribute, as the following expression shows:


[tabs]
======
Java DSL::
+
[source, java, role="primary"]
----
@Bean
public IntegrationFlow flow() {
    return IntegrationFlow.from("input")
            .delay(d -> d
                    .messageGroupId("delayer.messageGroupId")
                    .defaultDelay(3_000L)
                    .delayExpression("headers['delay']"))
            .channel("output")
            .get();
}
----

Kotlin DSL::
+
[source, kotlin, role="secondary"]
----
@Bean
fun flow() =
    integrationFlow("input") {
        delay {
            messageGroupId("delayer.messageGroupId")
            defaultDelay(3000L)
            delayExpression("headers['delay']")
        }
        channel("output")
    }
----

Java::
+
[source, java, role="secondary"]
----
@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::
+
[source, xml, role="secondary"]
----
<int:delayer id="delayer" input-channel="input" output-channel="output"
             default-delay="3000" expression="headers['delay']"/>
----
======

In the preceding example, the three-second delay applies only when the expression evaluates to null for a given inbound message.
If you want to apply a delay only to messages that have a valid result of the expression evaluation, you can use a 'default-delay' of `0` (the default).
For any message that has a delay of `0` (or less), the message is sent immediately, on the calling thread.

NOTE: The XML parser uses a message group ID of `<beanName>.messageGroupId`.

TIP: The delay handler supports expression evaluation results that represent an interval in milliseconds (any `Object` whose `toString()` method produces a value that can be parsed into a `Long`) as well as `java.util.Date` instances representing an absolute time.
In the first case, the milliseconds 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.

IMPORTANT: The expression evaluation may throw an evaluation exception for various reasons, including an invalid expression or other conditions.
By default, such exceptions are ignored (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`.

[TIP]
=====
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:

[source,java]
----
 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) using the indexer syntax instead of dot property accessor syntax, because detecting the null is faster than catching an exception.
=====

The delayer delegates to an instance of Spring's `TaskScheduler` abstraction.
The default scheduler used by the delayer is the `ThreadPoolTaskScheduler` instance provided by Spring Integration on startup.
See xref:configuration/namespace-taskscheduler.adoc[Configuring the Task Scheduler].
If you want to delegate to a different scheduler, you can provide a reference through the delayer element's 'scheduler' attribute, as the following example shows:

[source,xml]
----
<int:delayer id="delayer" input-channel="input" output-channel="output"
    expression="headers.delay"
    scheduler="exampleTaskScheduler"/>

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

TIP: If you configure an external `ThreadPoolTaskScheduler`, you can set `waitForTasksToCompleteOnShutdown = true` on this property.
It allows successful completion of 'delay' tasks that are already in the execution state (releasing the message) when the application is shutdown.
Before Spring Integration 2.2, this property was available on the `<delayer>` element, because `DelayHandler` could create its own scheduler on the background.
Since 2.2, the delayer requires an external scheduler instance and `waitForTasksToCompleteOnShutdown` was deleted.
You should use the scheduler's own configuration.

TIP: `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 xref:delayer.adoc#delayer-release-failures[Release Failures].

[[delayer-message-store]]
== Delayer and a Message Store

The `DelayHandler` persists delayed messages into the message group in the provided `MessageStore`.
(The 'groupId' is based on the required 'id' attribute of the `<delayer>` element.
See also `DelayHandler.setMessageGroupId(String)`.)
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 `messageGroupId` is required and cannot rely on a `DelayHandler` bean name which can be generated.
That way, after application restart, a `DelayHandler` may get a new generated bean name.
Therefore, delayed messages might be lost from rescheduling since their group is not managed by the application anymore.

The `<delayer>` can be enriched with either of two mutually exclusive elements: `<transactional>` and `<advice-chain>`.
The `List` of these AOP advices is applied to the proxied internal `DelayHandler.ReleaseMessageHandler`, which has the responsibility to release the message, after the delay, on a `Thread` of the scheduled task.
It might be used, for example, when the downstream message flow throws an exception and the transaction of the `ReleaseMessageHandler` is rolled back.
In this case, the delayed message remains in the persistent `MessageStore`.
You can use any custom `org.aopalliance.aop.Advice` implementation within the `<advice-chain>`.
The `<transactional>` element defines a simple advice chain that has only the transactional advice.
The following example shows an `advice-chain` within a `<delayer>`:

[source,xml]
----
<int:delayer id="delayer" input-channel="input" output-channel="output"
    expression="headers.delay"
    message-store="jdbcMessageStore">
    <int:advice-chain>
        <beans:ref bean="customAdviceBean"/>
        <tx:advice>
            <tx:attributes>
                <tx:method name="*" read-only="true"/>
            </tx:attributes>
        </tx:advice>
    </int:advice-chain>
</int:delayer>
----

The `DelayHandler` can be exported as a JMX `MBean` with managed operations (`getDelayedMessageCount` and `reschedulePersistedMessages`), which allows the rescheduling of delayed persisted messages at runtime -- for example, if the `TaskScheduler` has previously been stopped.
These operations can be invoked through a `Control Bus` command, as the following example shows:

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

NOTE: For more information regarding the message store, JMX, and the control bus, see xref:system-management.adoc[System Management].

Starting with version 5.3.7, if a transaction is active when a message is stored into a `MessageStore`, the release task is scheduled in a `TransactionSynchronization.afterCommit()` callback.
This is necessary to prevent a race condition, where the scheduled release could run before the transaction has committed, and the message is not found.
In this case, the message will be released after the delay, or after the transaction commits, whichever is later.

[[delayer-release-failures]]
== 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.
