From b2a4e67db75378359e9c306ff70d57cf3beaf796 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 8 Mar 2021 16:24:29 -0500 Subject: [PATCH] INT-4377: aggregator groupTimeout as Date (#3505) * INT-4377: aggregator groupTimeout as Date JIRA: https://jira.spring.io/browse/INT-4377 Change the `groupTimeoutExpression` logic to let it to be evaluated to `Date` instance for some fine-grained scheduling use-case, e.g. to determine a scheduling moment from the group creation time (`timestamp`) instead of a current message arrival * Fix language in docs accoridng PR review Co-authored-by: Gary Russell Co-authored-by: Gary Russell --- .../AbstractCorrelatingMessageHandler.java | 49 +++++++++++++------ .../dsl/CorrelationHandlerSpec.java | 5 +- .../integration/config/spring-integration.xsd | 1 + .../AggregatorIntegrationTests-context.xml | 2 +- .../AggregatorIntegrationTests.java | 10 ++-- src/reference/asciidoc/aggregator.adoc | 12 ++++- src/reference/asciidoc/whats-new.adoc | 3 ++ 7 files changed, 58 insertions(+), 24 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index f098c1ae37..1bdd4c8a6b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -626,16 +626,24 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } private void scheduleGroupToForceComplete(MessageGroup messageGroup) { - final Long groupTimeout = obtainGroupTimeout(messageGroup); + Object groupTimeout = obtainGroupTimeout(messageGroup); /* * When 'groupTimeout' is evaluated to 'null' we do nothing. * The 'MessageGroupStoreReaper' can be used to 'forceComplete' message groups. */ if (groupTimeout != null) { - if (groupTimeout > 0) { - final Object groupId = messageGroup.getGroupId(); - final long timestamp = messageGroup.getTimestamp(); - final long lastModified = messageGroup.getLastModified(); + Date startTime = null; + if (groupTimeout instanceof Date) { + startTime = (Date) groupTimeout; + } + else if ((Long) groupTimeout > 0) { + startTime = new Date(System.currentTimeMillis() + (Long) groupTimeout); + } + + if (startTime != null) { + Object groupId = messageGroup.getGroupId(); + long timestamp = messageGroup.getTimestamp(); + long lastModified = messageGroup.getLastModified(); ScheduledFuture scheduledFuture = getTaskScheduler() .schedule(() -> { @@ -643,14 +651,12 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP processForceRelease(groupId, timestamp, lastModified); } catch (MessageDeliveryException ex) { - if (AbstractCorrelatingMessageHandler.this.logger.isWarnEnabled()) { - AbstractCorrelatingMessageHandler.this.logger.warn(ex, - () -> "The MessageGroup [" + groupId - + "] is rescheduled by the reason of:"); - } + logger.warn(ex, () -> + "The MessageGroup [" + groupId + + "] is rescheduled by the reason of: "); scheduleGroupToForceComplete(groupId); } - }, new Date(System.currentTimeMillis() + groupTimeout)); + }, startTime); this.logger.debug(() -> "Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'."); this.expireGroupScheduledFutures.put(UUIDConverter.getUUID(groupId), scheduledFuture); @@ -905,9 +911,22 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP "The expected collection of Messages contains non-Message element: " + commonElementType); } - protected Long obtainGroupTimeout(MessageGroup group) { - return this.groupTimeoutExpression != null - ? this.groupTimeoutExpression.getValue(this.evaluationContext, group, Long.class) : null; + protected Object obtainGroupTimeout(MessageGroup group) { + if (this.groupTimeoutExpression != null) { + Object timeout = this.groupTimeoutExpression.getValue(this.evaluationContext, group); + if (timeout instanceof Date) { + return timeout; + } + else if (timeout != null) { + try { + return Long.parseLong(timeout.toString()); + } + catch (NumberFormatException ex) { + throw new IllegalStateException("Error evaluating 'groupTimeoutExpression'", ex); + } + } + } + return null; } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java index 0c4a259e21..fe7d46b0b7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java @@ -107,6 +107,8 @@ public abstract class CorrelationHandlerSpec{@code .groupTimeout(g -> g.size() * 2000L)}. + * Must return {@link java.util.Date}, {@link java.lang.Long} or {@link String} a long. * @param groupTimeoutFunction a function invoked to resolve the group timeout in milliseconds. * @return the handler spec. * @see AbstractCorrelatingMessageHandler#setGroupTimeoutExpression */ - public S groupTimeout(Function groupTimeoutFunction) { + public S groupTimeout(Function groupTimeoutFunction) { this.handler.setGroupTimeoutExpression(new FunctionExpression<>(groupTimeoutFunction)); return _this(); } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd index 97d8d22932..e1e67d69c1 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd @@ -3913,6 +3913,7 @@ the MessageGroup won't be scheduled to be forced complete. The action taken when the group is forced complete depends on the 'send-partial-result-on-expiry' attribute. + Can be evaluated directly to 'java.util.Date' instance for a scheduled task. Mutually exclusive with the 'group-timeout' attribute. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml index 7f17173b68..944c7906ca 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml @@ -47,7 +47,7 @@ >. In some cases, you may need to emit the aggregator result (or discard the group) after a timeout if the `ReleaseStrategy` does not release when the current message arrives. For this purpose, the `groupTimeout` option lets scheduling the `MessageGroup` be forced to complete, as the following example shows: @@ -721,6 +721,16 @@ The reaper initiates forced completion for all `MessageGroup` s in the `MessageG The `groupTimeout` does it for each `MessageGroup` individually if a new message does not arrive during the `groupTimeout`. Also, the reaper can be used to remove empty groups (empty groups are retained in order to discard late messages if `expire-groups-upon-completion` is false). +Starting with version 5.5, the `groupTimeoutExpression` can be evaluated to a `java.util.Date` instance. +This can be useful in cases like determining a scheduled task moment based on the group creation time (`MessageGroup.getTimestamp()`) instead of a current message arrival as it is calculated when `groupTimeoutExpression` is evaluated to `long`: + +==== +[source,xml] +---- +group-timeout-expression="size() ge 2 ? new java.util.Date(timestamp + 200) : null" +---- +==== + [[aggregator-annotations]] ===== Configuring an Aggregator with Annotations diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 6e4d32b389..2209a78249 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -34,6 +34,9 @@ The `ConsumerEndpointFactoryBean` now accept a `reactiveCustomizer` `Function` t This is covered as a `ConsumerEndpointSpec.reactive()` option in Java DSL and as a `@Reactive` nested annotation for the messaging annotations. See <<./reactive-streams.adoc#reactive-streams,Reactive Streams Support>> for more information. +The `groupTimeoutExpression` for a correlation message handler (an `Aggregator` and `Resequencer`) can now be evaluated to a `java.util.Date` for some fine-grained scheduling use-cases. +See <<./aggregator.adoc#aggregator,Aggregator>> for more information. + [[x5.5-amqp]] ==== AMQP Changes