INT-3420 Aggregator expire-groups-upon-timeout

JIRA: https://jira.spring.io/browse/INT-3420

Add option to allow the empty group to remain after timeout
so late arriving messages can be discarded.

INT-3420 Add Callout Hyperlinks

The aggregator configuration documentation had hyperlinks
from the attribute descriptions to the attribute in the
XML, but not vice-versa. For a large number of attributes
such as this, bi-directional hyperlinks are useful.

INT-3420 Doc Polishing

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

@@ -128,6 +128,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 
 	private volatile ApplicationEventPublisher applicationEventPublisher;
 
+	private volatile boolean expireGroupsUponTimeout = true;
+
 	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
 									 CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
 		Assert.notNull(processor);
@@ -300,6 +302,18 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 		this.releasePartialSequences = releasePartialSequences;
 	}
 
+	/**
+	 * Expire (completely remove) a group if it is completed due to timeout.
+	 * Subclasses setting this to false MUST handle null in the messages
+	 * argument to {@link #afterRelease(MessageGroup, Collection)}.
+	 * Default true.
+	 * @param expireGroupsUponTimeout the expireGroupsOnTimeout to set
+	 * @since 4.1
+	 */
+	protected void setExpireGroupsUponTimeout(boolean expireGroupsUponTimeout) {
+		this.expireGroupsUponTimeout = expireGroupsUponTimeout;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "aggregator";
@@ -462,7 +476,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 	 */
 	protected abstract void afterRelease(MessageGroup group, Collection<Message<?>> completedMessages);
 
-	private void forceComplete(MessageGroup group) {
+	protected void forceComplete(MessageGroup group) {
 
 		Object correlationKey = group.getGroupId();
 		// UUIDConverter is no-op if already converted
@@ -505,10 +519,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 						&& group.getTimestamp() == groupNow.getTimestamp()) {
 					if (groupSize > 0) {
 						if (releaseStrategy.canRelease(groupNow)) {
-							this.completeGroup(correlationKey, groupNow);
+							completeGroup(correlationKey, groupNow);
 						}
 						else {
-							this.expireGroup(correlationKey, groupNow);
+							expireGroup(correlationKey, groupNow);
+						}
+						if (!this.expireGroupsUponTimeout) {
+							afterRelease(groupNow, null);
+							removeGroup = false;
 						}
 					}
 					else {

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java

@@ -56,6 +56,11 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler
 		this.expireGroupsUponCompletion = expireGroupsUponCompletion;
 	}
 
+	@Override
+	public void setExpireGroupsUponTimeout(boolean expireGroupsOnTimeout) {
+		super.setExpireGroupsUponTimeout(expireGroupsOnTimeout);
+	}
+
 	@Override
 	protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages) {
 		this.messageStore.completeGroup(messageGroup.getGroupId());

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.config.xml;
 
+import org.w3c.dom.Element;
+
 import org.springframework.beans.BeanMetadataElement;
 import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.parsing.BeanComponentDefinition;
@@ -26,7 +28,6 @@ import org.springframework.integration.aggregator.DefaultAggregatingMessageGroup
 import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor;
 import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
 import org.springframework.util.StringUtils;
-import org.w3c.dom.Element;
 
 /**
  * Parser for the <em>aggregator</em> element of the integration namespace. Registers the annotation-driven
@@ -37,11 +38,14 @@ import org.w3c.dom.Element;
  * @author Oleg Zhurakousky
  * @author Dave Syer
  * @author Stefan Ferstl
+ * @author Gary Russell
  */
 public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser {
 
 	private static final String EXPIRE_GROUPS_UPON_COMPLETION = "expire-groups-upon-completion";
 
+	private static final String EXPIRE_GROUPS_UPON_TIMEOUT = "expire-groups-upon-timeout";
+
 	@Override
 	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
 		BeanComponentDefinition innerHandlerDefinition = IntegrationNamespaceUtils.parseInnerHandlerDefinition(element,
@@ -86,6 +90,7 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser {
 		this.doParse(builder, element, processor, parserContext);
 
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_COMPLETION);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_TIMEOUT);
 
 		return builder;
 	}

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

@@ -3376,6 +3376,16 @@
 						</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
+				<xsd:attribute name="expire-groups-upon-timeout" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							Boolean flag specifying, if a group is completed due to timeout (reaper or
+							'group-timeout(-expression)'), whether the group should be removed.
+							When true, late arriving messages will form a new group. When false, they
+							will be discarded. Default is 'true'
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java

@@ -13,9 +13,13 @@
 
 package org.springframework.integration.aggregator;
 
-import static org.hamcrest.CoreMatchers.*;
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -237,6 +241,38 @@ public class AggregatorTests {
 		assertEquals("ABC", this.expiryEvents.get(0).getGroupId());
 		assertEquals(2, this.expiryEvents.get(0).getMessageCount());
 		assertEquals(false, this.expiryEvents.get(0).isDiscarded());
+		Message<?> message3 = createMessage(5, "ABC", 3, 3, replyChannel, null);
+		this.aggregator.handleMessage(message3);
+		assertEquals(1, this.store.getMessageGroup("ABC").size());
+	}
+
+	@Test
+	public void testGroupRemainsAfterTimeout() throws InterruptedException {
+		this.aggregator.setSendPartialResultOnExpiry(true);
+		this.aggregator.setExpireGroupsUponTimeout(false);
+		QueueChannel replyChannel = new QueueChannel();
+		QueueChannel discardChannel = new QueueChannel();
+		this.aggregator.setDiscardChannel(discardChannel);
+		Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
+		Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
+		this.aggregator.handleMessage(message1);
+		this.aggregator.handleMessage(message2);
+		this.store.expireMessageGroups(-10000);
+		Message<?> reply = replyChannel.receive(1000);
+		assertNotNull("A reply message should have been received", reply);
+		assertEquals(15, reply.getPayload());
+		assertEquals(1, expiryEvents.size());
+		assertSame(this.aggregator, expiryEvents.get(0).getSource());
+		assertEquals("ABC", this.expiryEvents.get(0).getGroupId());
+		assertEquals(2, this.expiryEvents.get(0).getMessageCount());
+		assertEquals(false, this.expiryEvents.get(0).isDiscarded());
+		assertEquals(0, this.store.getMessageGroup("ABC").size());
+		Message<?> message3 = createMessage(5, "ABC", 3, 3, replyChannel, null);
+		this.aggregator.handleMessage(message3);
+		assertEquals(0, this.store.getMessageGroup("ABC").size());
+		Message<?> discardedMessage = discardChannel.receive(1000);
+		assertNotNull("A message should have been discarded", discardedMessage);
+		assertSame(message3, discardedMessage);
 	}
 
 	@Test

spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java

@@ -19,6 +19,7 @@ package org.springframework.integration.config;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
@@ -121,6 +122,7 @@ public class AggregatorParserTests {
 		Object handler = context.getBean("aggregatorWithExpressions.handler");
 		assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.messageBuilderFactory"));
 		assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.processor.messageBuilderFactory"));
+		assertTrue(TestUtils.getPropertyValue(handler, "expireGroupsUponTimeout", Boolean.class));
 	}
 
 	@Test
@@ -145,15 +147,16 @@ public class AggregatorParserTests {
 				releaseStrategy, accessor.getPropertyValue("releaseStrategy"));
 		assertEquals("The AggregatorEndpoint is not injected with the appropriate CorrelationStrategy instance",
 				correlationStrategy, accessor.getPropertyValue("correlationStrategy"));
-		Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate output channel",
+		assertEquals("The AggregatorEndpoint is not injected with the appropriate output channel",
 				outputChannel, accessor.getPropertyValue("outputChannel"));
-		Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate discard channel",
+		assertEquals("The AggregatorEndpoint is not injected with the appropriate discard channel",
 				discardChannel, accessor.getPropertyValue("discardChannel"));
-		Assert.assertEquals("The AggregatorEndpoint is not set with the appropriate timeout value", 86420000l,
+		assertEquals("The AggregatorEndpoint is not set with the appropriate timeout value", 86420000l,
 				TestUtils.getPropertyValue(consumer, "messagingTemplate.sendTimeout"));
-		Assert.assertEquals(
+		assertEquals(
 				"The AggregatorEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
 				true, accessor.getPropertyValue("sendPartialResultOnExpiry"));
+		assertFalse(TestUtils.getPropertyValue(consumer, "expireGroupsUponTimeout", Boolean.class));
 	}
 
 	@Test

spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml

@@ -27,7 +27,8 @@
 		release-strategy="releaseStrategy"
 		correlation-strategy="correlationStrategy"
 		send-timeout="86420000"
-		send-partial-result-on-expiry="true"/>
+		send-partial-result-on-expiry="true"
+		expire-groups-upon-timeout="false"/>
 
 	<channel id="aggregatorWithExpressionsInput"/>
 	<channel id="aggregatorWithExpressionsOutput"/>