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

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<UUID, ScheduledFuture<?>> expireGroupScheduledFutures = new HashMap<UUID, ScheduledFuture<?>>();
+
 	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<Message<?>> sorted = new ArrayList<Message<?>>(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) {

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,

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd

@@ -3433,6 +3433,66 @@
 						<xsd:documentation>A SpEL expression to evaluate against a root object that is the Collection of messages within the message group (e.g, size() > 6)</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
+				<xsd:attribute name="group-timeout" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							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.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="group-timeout-expression" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							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.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="scheduler" type="xsd:string">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type type="org.springframework.scheduling.TaskScheduler" />
+							</tool:annotation>
+						</xsd:appinfo>
+						<xsd:documentation>
+							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.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="lock-registry" type="xsd:string">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type type="org.springframework.integration.util.LockRegistry" />
+							</tool:annotation>
+						</xsd:appinfo>
+						<xsd:documentation>
+							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.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 				<xsd:attribute name="discard-channel" type="xsd:string">
 					<xsd:annotation>
 						<xsd:appinfo>

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

@@ -34,5 +34,15 @@
 	<aggregator id="nonExpiringAggregator" input-channel="nonExpiringAggregatorInput" output-channel="output"
 				expire-groups-upon-completion="false" discard-channel="discard"/>
 
+	<aggregator id="gta"
+				input-channel="groupTimeoutAggregatorInput" output-channel="output" discard-channel="discard"
+				send-partial-result-on-expiry="true"
+				group-timeout="100"/>
+
+	<aggregator input-channel="groupTimeoutExpressionAggregatorInput" output-channel="output" discard-channel="discard"
+				send-partial-result-on-expiry="true"
+				group-timeout-expression="size() ge 2 ? 100 : -1"
+			    release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
+
 
 </beans:beans>

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<String, Object> headers = stubHeaders(i, 5, 1);
 			input.send(new GenericMessage<Integer>(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<String, Object> headers = stubHeaders(i, 5, 1);
+			this.groupTimeoutAggregatorInput.send(new GenericMessage<Integer>(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<Integer>(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<Integer>(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<Integer>(3, stubHeaders(3, 6, 1)));
+		assertNull(this.output.receive(0));
+		assertNull(this.discard.receive(0));
+
+		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(4, stubHeaders(4, 6, 1)));
+		assertNull(this.output.receive(0));
+		assertNull(this.discard.receive(0));
+
+		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(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<Integer>(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<Integer> numbers) {
@@ -127,11 +199,11 @@ public class AggregatorIntegrationTests {
 		}
 	}
 
-	private Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correllationId) {
+	private Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) {
 		Map<String, Object> headers = new HashMap<String, Object>();
 		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;
 	}
 

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