INT-958: change meaning of timeout parameter in MessageGroupStore

org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java

@@ -66,14 +66,16 @@ public interface MessageGroupStore {
 	void registerExpiryCallback(MessageGroupCallback callback);
 
 	/**
-	 * Extract all expired groups (those whose timestamp is less than the threshold provided) and call each of the
-	 * registered callbacks on them in turn.
+	 * Extract all expired groups (whose timestamp is older than the current time less the threshold provided) and call
+	 * each of the registered callbacks on them in turn. For example: call with a timeout of 100 to expire all groups
+	 * that were created more than 100 milliseconds ago, and are not yet complete. Use a timeout of 0 (or negative to be
+	 * on the safe side) to expire all message groups.
 	 * 
-	 * @param timestamp the timestamp threshold to use
+	 * @param timeout the timeout threshold to use
 	 * @return the number of message groups expired
 	 * 
 	 * @see #registerExpiryCallback(MessageGroupCallback)
 	 */
-	int expireMessageGroups(long timestamp);
+	int expireMessageGroups(long timeout);
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java

@@ -114,10 +114,11 @@ public class SimpleMessageStore implements MessageStore, MessageGroupStore {
 		expiryCallbacks.add(callback);
 	}
 
-	public int expireMessageGroups(long timestamp) {
+	public int expireMessageGroups(long timeout) {
 		int count = 0;
+		long threshold = System.currentTimeMillis() - timeout;
 		for (MessageGroup group : correlationToMessageGroup.values()) {
-			if (group.getTimestamp() < timestamp) {
+			if (group.getTimestamp() < threshold) {
 				count++;
 				expire(group);
 				removeMessageGroup(group.getCorrelationKey());

org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java

@@ -75,7 +75,7 @@ public class AggregatorTests {
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message = createMessage(3, "ABC", 2, 1, replyChannel, null);
 		this.aggregator.handleMessage(message);
-		this.store.expireMessageGroups(System.currentTimeMillis() + 10000);
+		this.store.expireMessageGroups(-10000);
 		Message<?> reply = replyChannel.receive(100);
 		assertNull("No message should have been sent normally", reply);
 		Message<?> discardedMessage = discardChannel.receive(1000);
@@ -91,7 +91,7 @@ public class AggregatorTests {
 		Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
 		this.aggregator.handleMessage(message1);
 		this.aggregator.handleMessage(message2);
-		this.store.expireMessageGroups(System.currentTimeMillis() + 10000);
+		this.store.expireMessageGroups(-10000);
 		Message<?> reply = replyChannel.receive(0);
 		assertNotNull("A reply message should have been received", reply);
 		assertEquals(15, reply.getPayload());

org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java

@@ -120,7 +120,7 @@ public class ConcurrentAggregatorTests {
 				.getCount());
 		Message<?> reply = replyChannel.receive(100);
 		assertNull("No message should have been sent normally", reply);
-        this.store.expireMessageGroups(System.currentTimeMillis()+10000);
+        this.store.expireMessageGroups(-10000);
 		Message<?> discardedMessage = discardChannel.receive(100);
 		assertNotNull("A message should have been discarded", discardedMessage);
 		assertEquals(message, discardedMessage);
@@ -143,7 +143,7 @@ public class ConcurrentAggregatorTests {
 		latch.await(300, TimeUnit.MILLISECONDS);
 		assertEquals("handlers should have been invoked within time limit", 0,
 				latch.getCount());
-        this.store.expireMessageGroups(System.currentTimeMillis()+10000);
+        this.store.expireMessageGroups(-10000);
 		Message<?> reply = replyChannel.receive(100);
 		assertNotNull("A reply message should have been received", reply);
 		assertEquals(15, reply.getPayload());

org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java

@@ -129,7 +129,7 @@ public class CorrelatingMessageHandlerTests {
 		});
 
 		Thread.sleep(20);
-		assertEquals(0, store.expireMessageGroups(System.currentTimeMillis()+10000));
+		assertEquals(0, store.expireMessageGroups(10000));
 
 		bothMessagesHandled.await();
 

org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java

@@ -136,7 +136,7 @@ public class ResequencerTests {
 		this.resequencer.setTimeout(90000);
 		this.resequencer.handleMessage(message1);
 		this.resequencer.handleMessage(message2);
-		assertEquals(1, store.expireMessageGroups(System.currentTimeMillis()+10000));
+		assertEquals(1, store.expireMessageGroups(-10000));
 		Message<?> reply1 = discardChannel.receive(0);
 		Message<?> reply2 = discardChannel.receive(0);
 		Message<?> reply3 = discardChannel.receive(0);

org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java

@@ -93,7 +93,7 @@ public class SimpleMessageStoreTests {
 		store.addMessageToGroup("bar", testMessage1);
 		assertEquals(1, store.getMessageGroup("bar").size());
 
-		store.expireMessageGroups(System.currentTimeMillis()+10000);
+		store.expireMessageGroups(-10000);
 		assertEquals("[foo]", list.toString());
 		assertEquals(0, store.getMessageGroup("bar").size());