From eb0d1ddc845acd798c7630f09462d214d99ed6e5 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 2 Apr 2014 14:03:18 +0300 Subject: [PATCH] INT-2595: Add Time-Based `ReleaseStrategy` Option JIRA: https://jira.spring.io/browse/INT-2595 * Add `group-timeout` and `group-timeout-expression` to the Correlation Endpoint * Add logic to the `AbstractCorrelatingMessageHandler` to schedule group for `forceComplete`, when the target `ReleaseStrategy` returns `false` INT-2595: Polishing according PR comments INT-2595: Expose `lock-registry` and further docs INT-2595 Doc and Test Polishing INT-2595: Fix typos INT-2595 More Minor Doc Polish --- .../AbstractCorrelatingMessageHandler.java | 92 ++++++++++++++- ...stractCorrelatingMessageHandlerParser.java | 8 ++ .../config/xml/spring-integration-4.0.xsd | 60 ++++++++++ .../AggregatorIntegrationTests-context.xml | 10 ++ .../AggregatorIntegrationTests.java | 94 ++++++++++++++-- .../src/test/resources/log4j.properties | 2 +- src/reference/docbook/aggregator.xml | 105 +++++++++++++++++- src/reference/docbook/resequencer.xml | 27 ++++- src/reference/docbook/whats-new.xml | 11 ++ 9 files changed, 388 insertions(+), 21 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 40f99c558f..f4edc20428 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 @@ -16,7 +16,12 @@ package org.springframework.integration.aggregator; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.Date; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.locks.Lock; import org.apache.commons.logging.Log; @@ -25,10 +30,14 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.expression.IntegrationEvaluationContextAware; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; @@ -43,6 +52,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; import org.springframework.messaging.core.DestinationResolutionException; +import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -70,12 +80,15 @@ import org.springframework.util.StringUtils; * @author Artem Bilan * @since 2.0 */ -public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer { +public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageHandler + implements MessageProducer, DisposableBean, IntegrationEvaluationContextAware { private static final Log logger = LogFactory.getLog(AbstractCorrelatingMessageHandler.class); public static final long DEFAULT_SEND_TIMEOUT = 1000L; + private final Map> expireGroupScheduledFutures = new HashMap>(); + protected volatile MessageGroupStore messageStore; private final MessageGroupProcessor outputProcessor; @@ -106,6 +119,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH private volatile boolean releasePartialSequences; + private volatile Expression groupTimeoutExpression; + + private EvaluationContext evaluationContext; + public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(processor); @@ -166,6 +183,20 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH this.outputChannelName = outputChannelName; } + public void setGroupTimeoutExpression(Expression groupTimeoutExpression) { + this.groupTimeoutExpression = groupTimeoutExpression; + } + + @Override + public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + this.evaluationContext = evaluationContext; + } + + @Override + public void setTaskScheduler(TaskScheduler taskScheduler) { + super.setTaskScheduler(taskScheduler); + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -277,11 +308,18 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message); } - // TODO: INT-1117 - make the lock global? - Lock lock = this.lockRegistry.obtain(UUIDConverter.getUUID(correlationKey).toString()); + UUID groupIdUuid = UUIDConverter.getUUID(correlationKey); + Lock lock = this.lockRegistry.obtain(groupIdUuid.toString()); lock.lockInterruptibly(); try { + ScheduledFuture scheduledFuture = this.expireGroupScheduledFutures.remove(groupIdUuid); + if (scheduledFuture != null) { + boolean canceled = scheduledFuture.cancel(true); + if (canceled && logger.isDebugEnabled()) { + logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ " + correlationKey + "]."); + } + } MessageGroup messageGroup = messageStore.getMessageGroup(correlationKey); if (this.sequenceAware){ messageGroup = new SequenceAwareMessageGroup(messageGroup); @@ -304,6 +342,31 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH this.afterRelease(messageGroup, completedMessages); } } + else { + Long groupTimeout = this.obtainGroupTimeout(messageGroup); + if (groupTimeout != null && groupTimeout >= 0) { + if (groupTimeout > 0) { + final MessageGroup messageGroupToSchedule = messageGroup; + + scheduledFuture = this.getTaskScheduler() + .schedule(new Runnable() { + + @Override + public void run() { + AbstractCorrelatingMessageHandler.this.forceComplete(messageGroupToSchedule); + } + }, new Date(System.currentTimeMillis() + groupTimeout)); + + if (logger.isDebugEnabled()) { + logger.debug("Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'."); + } + this.expireGroupScheduledFutures.put(groupIdUuid, scheduledFuture); + } + else { + this.forceComplete(messageGroup); + } + } + } } else { discardChannel.send(message); @@ -330,6 +393,13 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH try { lock.lockInterruptibly(); try { + ScheduledFuture scheduledFuture = this.expireGroupScheduledFutures.remove(UUIDConverter.getUUID(correlationKey)); + if (scheduledFuture != null) { + boolean canceled = scheduledFuture.cancel(false); + if (canceled && logger.isDebugEnabled()) { + logger.debug("Cancel 'forceComplete' scheduling for MessageGroup [ " + group + "]."); + } + } MessageGroup groupNow = group; /* * If the group argument is not already complete, @@ -392,7 +462,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH } catch (InterruptedException ie) { Thread.currentThread().interrupt(); - throw new MessagingException("Thread was interrupted while trying to obtain lock"); + logger.debug("Thread was interrupted while trying to obtain lock"); } } @@ -405,7 +475,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH List> sorted = new ArrayList>(partialSequence); Collections.sort(sorted, new SequenceNumberComparator()); - Message lastReleasedMessage = sorted.get(partialSequence.size()-1); + Message lastReleasedMessage = sorted.get(partialSequence.size() - 1); return new IntegrationMessageHeaderAccessor(lastReleasedMessage).getSequenceNumber(); } @@ -512,6 +582,18 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH return false; } + private Long obtainGroupTimeout(MessageGroup group) { + return this.groupTimeoutExpression != null + ? this.groupTimeoutExpression.getValue(this.evaluationContext, group, Long.class) : null; + } + + @Override + public void destroy() throws Exception { + for (ScheduledFuture future : expireGroupScheduledFutures.values()) { + future.cancel(true); + } + } + private static class SequenceAwareMessageGroup extends SimpleMessageGroup { public SequenceAwareMessageGroup(MessageGroup messageGroup) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java index aca1bee30f..2b836420cd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java @@ -15,6 +15,7 @@ package org.springframework.integration.config.xml; import org.w3c.dom.Element; import org.springframework.beans.BeanMetadataElement; +import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; @@ -66,10 +67,17 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo element, builder, processor, parserContext); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_STORE_ATTRIBUTE); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "scheduler", "taskScheduler"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "lock-registry"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "empty-group-min-timeout", "minimumTimeoutForEmptyGroups"); + + BeanDefinition expressionDef = + IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("group-timeout", "group-timeout-expression", + parserContext, element, false); + builder.addPropertyValue("groupTimeoutExpression", expressionDef); } protected void injectPropertyWithAdapter(String beanRefAttribute, String methodRefAttribute, diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd index da67e18aef..11442a0231 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd @@ -3433,6 +3433,66 @@ A SpEL expression to evaluate against a root object that is the Collection of messages within the message group (e.g, size() > 6) + + + + A timeout in milliseconds to force the MessageGroup complete, + when the 'ReleaseStrategy' does not 'release' the group when the current Message arrives. + If 'group-timeout' is not provided or is less than '0' 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. + Mutually exclusive with the 'group-timeout-expression' attribute. + + + + + + + A SpEL expression to evaluate a 'group-timeout' with the MessageGroup as the #root evaluation + context object for scheduling the MessageGroup forced completion, + when the 'ReleaseStrategy' does not 'release' the group when the current Message arrives. + If 'group-timeout-expression' evaluates to 'null' or less than '0', + 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. + Mutually exclusive with the 'group-timeout' attribute. + + + + + + + + + + + + Provide a reference to the TaskScheduler instance to schedule 'forceComplete' on + the MessageGroup + when no new message arrives for the MessageGroup within the + 'group-timeout' or 'group-timeout-expression'. If it isn't + provided, the default scheduler 'taskScheduler', + registered in the ApplicationContext {ThreadPoolTaskScheduler} will be + used. This attribute only applies if 'group-timeout' or + 'group-timeout-expression' is specified. + + + + + + + + + + + + A reference to a 'org.springframework.integration.util.LockRegistry' bean + to obtain 'java.util.concurrent.locks.Lock' by 'groupId'. Used for concurrent operations on + MessageGroups. By default an internal 'DefaultLockRegistry' is used. + + + 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 599fc36fa2..6a94a2714f 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 @@ -34,5 +34,15 @@ + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java index f7eceb391b..09982a091d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2014 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. @@ -19,18 +19,24 @@ package org.springframework.integration.aggregator.integration; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.ApplicationContext; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; -import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -39,29 +45,35 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Iwein Fuld * @author Alex Peters * @author Oleg Zhurakousky + * @author Artem Bilan + * @author Gary Russell */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration public class AggregatorIntegrationTests { @Autowired - @Qualifier("input") + private ApplicationContext context; + + @Autowired private MessageChannel input; @Autowired - @Qualifier("expiringAggregatorInput") private MessageChannel expiringAggregatorInput; @Autowired - @Qualifier("nonExpiringAggregatorInput") private MessageChannel nonExpiringAggregatorInput; @Autowired - @Qualifier("output") + private MessageChannel groupTimeoutAggregatorInput; + + @Autowired + private MessageChannel groupTimeoutExpressionAggregatorInput; + + @Autowired private PollableChannel output; @Autowired - @Qualifier("discard") private PollableChannel discard; @Test//(timeout=5000) @@ -70,7 +82,7 @@ public class AggregatorIntegrationTests { Map headers = stubHeaders(i, 5, 1); input.send(new GenericMessage(i, headers)); } - assertEquals(0 + 1 + 2 + 3 + 4, output.receive().getPayload()); + assertEquals(0 + 1 + 2 + 3 + 4, output.receive(1000).getPayload()); } @Test @@ -116,6 +128,66 @@ public class AggregatorIntegrationTests { } + @Test + public void testGroupTimeoutScheduling() throws Exception { + for (int i = 0; i < 5; i++) { + Map headers = stubHeaders(i, 5, 1); + this.groupTimeoutAggregatorInput.send(new GenericMessage(i, headers)); + + //Wait until 'group-timeout' does its stuff. + MessageGroupStore mgs = TestUtils.getPropertyValue(this.context.getBean("gta.handler"), "messageStore", + MessageGroupStore.class); + int n = 0; + while (n++ < 100 && mgs.getMessageGroupCount() > 0) { + Thread.sleep(100); + } + assertTrue("Group did not complete", n < 100); + assertNotNull(this.output.receive(1000)); + assertNull(this.discard.receive(0)); + } + } + + + @Test + public void testGroupTimeoutExpressionScheduling() throws Exception { + // Since group-timeout-expression="size() >= 2 ? 100 : -1". The first message won't be scheduled to 'forceComplete' + this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage(1, stubHeaders(1, 6, 1))); + assertNull(this.output.receive(0)); + assertNull(this.discard.receive(0)); + + // As far as 'group.size() >= 2' it will be scheduled to 'forceComplete' + this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage(2, stubHeaders(2, 6, 1))); + assertNull(this.output.receive(0)); + Message receive = this.output.receive(500); + assertNotNull(receive); + assertEquals(2, ((Collection) receive.getPayload()).size()); + assertNull(this.discard.receive(0)); + + // The same with these three messages + this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage(3, stubHeaders(3, 6, 1))); + assertNull(this.output.receive(0)); + assertNull(this.discard.receive(0)); + + this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage(4, stubHeaders(4, 6, 1))); + assertNull(this.output.receive(0)); + assertNull(this.discard.receive(0)); + + this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage(5, stubHeaders(5, 6, 1))); + assertNull(this.output.receive(0)); + receive = this.output.receive(500); + assertNotNull(receive); + assertEquals(3, ((Collection) receive.getPayload()).size()); + assertNull(this.discard.receive(0)); + + // The last message in the sequence - normal release by provided 'ReleaseStrategy' + this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage(6, stubHeaders(6, 6, 1))); + receive = this.output.receive(0); + assertNotNull(receive); + assertEquals(1, ((Collection) receive.getPayload()).size()); + assertNull(this.discard.receive(0)); + } + + // configured in context associated with this test public static class SummingAggregator { public Integer sum(List numbers) { @@ -127,11 +199,11 @@ public class AggregatorIntegrationTests { } } - private Map stubHeaders(int sequenceNumber, int sequenceSize, int correllationId) { + private Map stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) { Map headers = new HashMap(); headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber); headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize); - headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correllationId); + headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId); return headers; } diff --git a/spring-integration-core/src/test/resources/log4j.properties b/spring-integration-core/src/test/resources/log4j.properties index 538a5c4c9e..64ab790552 100644 --- a/spring-integration-core/src/test/resources/log4j.properties +++ b/spring-integration-core/src/test/resources/log4j.properties @@ -2,6 +2,6 @@ log4j.rootCategory=WARN, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n +log4j.appender.stdout.layout.ConversionPattern=%d %c{1} [%t] : %m%n log4j.category.org.springframework.integration=WARN diff --git a/src/reference/docbook/aggregator.xml b/src/reference/docbook/aggregator.xml index 612bfcc92c..451d824fde 100644 --- a/src/reference/docbook/aggregator.xml +++ b/src/reference/docbook/aggregator.xml @@ -368,7 +368,13 @@ then you should simply provide an implementation of the ReleaseStrate release-strategy-expression="size() == 5" ]]> ]]> ]]> @@ -435,8 +441,7 @@ then you should simply provide an implementation of the ReleaseStrate or by simply invoking that method if you have a reference to the MessageGroupStore instance. Otherwise by itself this attribute has no behavior. It only serves as an indicator of what to do (discard or send to the output/reply channel) with Messages that are still in the MessageGroup that is about to be expired. - Optional. - Default - 'false'. + Optional. Default - 'false'. @@ -524,6 +529,61 @@ then you should simply provide an implementation of the ReleaseStrate property and it could be as much as this value plus the timeout. + + + A reference to a org.springframework.integration.util.LockRegistry bean; + used to obtain a Lock based on the groupId for + concurrent operations on the + MessageGroup. By default, an internal DefaultLockRegistry is used. + + + + + + A timeout in milliseconds to force the MessageGroup complete, + when the ReleaseStrategy doesn't release + the group when the current Message arrives. + This attribute provides a built-in Time-base Release Strategy for the aggregator, + when there is a need to emit a partial result (or discard the group), if a new Message does not arrive + for the MessageGroup within the timeout. + When a new Message arrives at the aggregator, any existing ScheduledFuture<?> + for its MessageGroup is canceled. If the + ReleaseStrategy + returns false (don't release) and the groupTimeout > 0 a new task will be + scheduled to expire the group. + Setting this attribute to zero is not advised because it will effectively disable the aggregator because every + message group will be immediately completed. It is possible, however to conditionally set it to zero using an + expression; see group-timeout-expression for information. + The action taken during the completion depends on the ReleaseStrategy and the + send-partial-group-on-expiry attribute. See for + more information. + Mutually exclusive with 'group-timeout-expression' attribute. + + + + + The SpEL expression that evaluates to a groupTimeout with the MessageGroup + as the #root evaluation context object. Used for scheduling the MessageGroup to + be forced complete. If the expression evaluates to null or < 0, the + completion is not scheduled. If it evaluates to zero, the group is completed immediately on + the current thread. In effect, this provides a dynamic group-timeout property. + See group-timeout for more information. + Mutually exclusive with 'group-timeout' attribute. + + + + + A TaskScheduler bean reference to schedule + the MessageGroup to be forced complete + if no new message arrives for the MessageGroup + within the groupTimeout. + If not provided, the default scheduler taskScheduler, + registered in the ApplicationContext (ThreadPoolTaskScheduler) + will be used. This attribute does not apply if group-timeout or + group-timeout-expression is not specified. + + + Using a ref attribute is generally recommended if a custom @@ -663,6 +723,45 @@ then you should simply provide an implementation of the ReleaseStrate MessageGroup itself, and you are simply stating that as soon as there are more than 5 messages in this group, it should be released. + +
+ 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 description above). There are some + cases where it is needed to emit the aggregator result (or discard the group) after a timeout + if the ReleaseStrategy + doesn't release when the current Message arrives. + For this purpose the groupTimeout option allows scheduling the MessageGroup to + be forced complete: + ]]> + With this example, the normal release will be possible if the aggregator receives the last message + in sequence as defined by the release-strategy-expression. If that specific message does not arrive, + the groupTimeout will force the group complete after 10 seconds as long as the group contains at least 2 Messages. + + + The results of forcing the group complete depends on the ReleaseStrategy and + the send-partial-result-on-expiry. + First, the release strategy is again consulted to see if a normal release is to be made - while the + group won't have changed, the ReleaseStrategy can decide to release the group at this time. If the + release strategy still does not release the group, it will be expired. + If send-partial-result-on-expiry is true, + existing messages in the (partial) MessageGroup will be released as a normal aggregator reply Message to the + output-channel, otherwise it will be discarded. + + + There is a difference between groupTimeout behavior and MessageGroupStoreReaper + (see ). The reaper initiates forced completion for all MessageGroups + in the MessageGroupStore periodically. The groupTimeout does it for each MessageGroup + individually, if a new Message doesn't 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). + +
diff --git a/src/reference/docbook/resequencer.xml b/src/reference/docbook/resequencer.xml index 9c2aa2adb4..7c249b04af 100644 --- a/src/reference/docbook/resequencer.xml +++ b/src/reference/docbook/resequencer.xml @@ -52,8 +52,13 @@ release-strategy="releaseStrategyBean" ]]>]]> + empty-group-min-timeout="60000" ]]> ]]> The id of the resequencer is @@ -160,7 +165,27 @@ empty group will also be affected by the reaper's timeout property and it could be as much as this value plus the timeout. + + + See . + + + + + See . + + + + + See . + + + + + See . + + diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 4c8aabc186..8901015c8e 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -209,5 +209,16 @@ For more information see .
+
+ Correlation Endpoint: Time-based Release Strategy + + The mutually exclusive group-timeout and group-timeout-expression + attributes have been added to the <int:aggregator> and <int:resequencer>. + These attributes allow forced completion of a partial MessageGroup, + if the ReleaseStrategy does not release a group and no + further messages arrive within the time specified. + For more information see . + +