Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4377: aggregator groupTimeout as Date (#3505)

Browse Source 
* 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 <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2021-03-08 16:24:29 -05:00
committed by GitHub
parent e9f234683e
commit b2a4e67db7
7 changed files with 58 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -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

5
spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java
View File

@@ -107,6 +107,8 @@ public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,
}
/**
* Specify a SpEL expression to evaluate the group timeout for scheduled expiration.
* Must return {@link java.util.Date}, {@link java.lang.Long} or {@link String} as a long.
* @param groupTimeoutExpression the group timeout expression string.
* @return the handler spec.
* @see AbstractCorrelatingMessageHandler#setGroupTimeoutExpression
@@ -122,11 +124,12 @@ public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,
* based on the message group.
* Usually used with a JDK8 lambda:
* <p>{@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<MessageGroup, Long> groupTimeoutFunction) {
public S groupTimeout(Function<MessageGroup, ?> groupTimeoutFunction) {
this.handler.setGroupTimeoutExpression(new FunctionExpression<>(groupTimeoutFunction));
return _this();
}

1
spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd
View File

@@ -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.
</xsd:documentation>
</xsd:annotation>

2
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml
View File

@@ -47,7 +47,7 @@
<aggregator input-channel="groupTimeoutExpressionAggregatorInput" output-channel="output" discard-channel="discard"
send-partial-result-on-expiry="true"
group-timeout-expression="size() ge 2 ? 100 : null"
group-timeout-expression="size() ge 2 ? new java.util.Date(timestamp + 200) : null"
release-strategy-expression="messages[0].headers.sequenceNumber == messages[0].headers.sequenceSize"/>
<aggregator input-channel="zeroGroupTimeoutExpressionAggregatorInput" output-channel="output" discard-channel="discard"

10
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 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.
@@ -26,8 +26,7 @@ import java.util.Map;
import java.util.Queue;
import java.util.function.Function;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
@@ -42,7 +41,7 @@ import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Iwein Fuld
@@ -51,7 +50,7 @@ import org.springframework.test.context.junit4.SpringRunner;
* @author Artem Bilan
* @author Gary Russell
*/
@RunWith(SpringRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class AggregatorIntegrationTests {
@@ -274,5 +273,4 @@ public class AggregatorIntegrationTests {
}
}

12
src/reference/asciidoc/aggregator.adoc
View File

@@ -691,7 +691,7 @@ In the preceding example, the root object of the SpEL evaluation context is the
[[agg-and-group-to]]
====== Aggregator and Group Timeout
Starting with version 4.0, two new mutually exclusive attributes have been introduced: `group-timeout` and `group-timeout-expression` (see the earlier description).
Starting with version 4.0, two new mutually exclusive attributes have been introduced: `group-timeout` and `group-timeout-expression`.
See <<aggregator-xml>>.
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

3
src/reference/asciidoc/whats-new.adoc
View File

@@ -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