diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java index 084420c74e..86ffacafaa 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java @@ -1208,6 +1208,7 @@ public abstract class BaseIntegrationFlowDefinition { @@ -96,13 +99,12 @@ public class DelayerEndpointSpec extends ConsumerEndpointSpec DelayerEndpointSpec delayFunction(Function, Object> delayFunction) { - this.handler.setDelayExpression(new FunctionExpression<>(delayFunction)); - return this; + return delayExpression(new FunctionExpression<>(delayFunction)); } /** * Set a group id to manage delayed messages by this handler. + * Required. * @param messageGroupId the group id for delayed messages. * @return the endpoint spec. * @since 6.2 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index a1553615da..74a619faa3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -83,6 +83,17 @@ import org.springframework.util.ObjectUtils; * Message can be released as soon as five seconds from the current time). If the value is * a Date, it will be delayed at least until that Date occurs (i.e. the delay in that case * is equivalent to {@code headerDate.getTime() - new Date().getTime()}). + *

+ * Delayed messages are stored in the {@link MessageGroupStore} as a dedicated group. + * If an external persistent store is provided, those delayed messages will be rescheduled + * after application startup. + * The {@link #messageGroupId} is required option and must be unique for each delayer + * configuration to avoid work-stealing from the store and unexpected releases. + * Different instances of the same delayer can point to the same message group in the store. + * The {@link #messageGroupId} cannot rely on a bean name which might be generated. + * After application restart the bean may get a different generated name and its delayed + * messages might be lost from reschedule since its group is not managed + * by the application anymore. * * @author Mark Fisher * @author Artem Bilan @@ -90,7 +101,6 @@ import org.springframework.util.ObjectUtils; * * @since 1.0.3 */ - @ManagedResource @IntegrationManagedResource public class DelayHandler extends AbstractReplyProducingMessageHandler implements DelayHandlerManagement, @@ -132,7 +142,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement /** * Construct an instance with default options. - * The {@link #messageGroupId}must then be provided via the setter. + * The {@link #messageGroupId} must then be provided via the setter. * @since 6.2 */ public DelayHandler() { @@ -165,6 +175,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement /** * Set a group id to manage delayed messages by this handler. + * Required. * @param messageGroupId the group id for delayed messages. * @since 6.2 */ diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt index a6d74051d5..1dab770f78 100644 --- a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt @@ -573,6 +573,7 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ /** * Populate a [DelayHandler] to the current integration flow position. + * The [DelayerEndpointSpec#messageGroupId(String)] is required option. * @since 6.2 */ fun delay(endpointConfigurer: DelayerEndpointSpec.() -> Unit) { diff --git a/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/GroovyIntegrationFlowDefinition.groovy b/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/GroovyIntegrationFlowDefinition.groovy index 423f82a2cd..a6c0a48ed8 100644 --- a/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/GroovyIntegrationFlowDefinition.groovy +++ b/spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/GroovyIntegrationFlowDefinition.groovy @@ -633,9 +633,10 @@ class GroovyIntegrationFlowDefinition { /** * Populate a {@link org.springframework.integration.handler.DelayHandler} to the current integration flow position. + * The {@link DelayerEndpointSpec#messageGroupId(String)} is required option. * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options. - * @see org.springframework.integration.dsl.DelayerEndpointSpec * @since 6.2 + * @see DelayerEndpointSpec */ GroovyIntegrationFlowDefinition delay( @DelegatesTo(value = DelayerEndpointSpec, strategy = Closure.DELEGATE_FIRST) diff --git a/src/reference/asciidoc/delayer.adoc b/src/reference/asciidoc/delayer.adoc index a9aa828866..ac0f423770 100644 --- a/src/reference/asciidoc/delayer.adoc +++ b/src/reference/asciidoc/delayer.adoc @@ -149,12 +149,16 @@ See <>. ==== 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.) +(The 'groupId' is based on the required 'id' attribute of the `` 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 `` 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.