From 9b6d5720f80d1f9201fa60711c353241059757ca Mon Sep 17 00:00:00 2001 From: David Syer Date: Wed, 5 May 2010 11:10:54 +0000 Subject: [PATCH] INT-958: change meaning of timeout parameter in MessageGroupStore --- .../integration/store/MessageGroupStore.java | 10 ++++++---- .../integration/store/SimpleMessageStore.java | 5 +++-- .../integration/aggregator/AggregatorTests.java | 4 ++-- .../aggregator/ConcurrentAggregatorTests.java | 4 ++-- .../aggregator/CorrelatingMessageHandlerTests.java | 2 +- .../integration/aggregator/ResequencerTests.java | 2 +- .../integration/store/SimpleMessageStoreTests.java | 2 +- 7 files changed, 16 insertions(+), 13 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java index 3776edc4be..c026f7f113 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/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); } \ No newline at end of file diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 3f17dc8ef5..9b3f047dc8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/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()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index 4674906ab9..ade0854b1e 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/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()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java index 7504a92389..b4bb900bd4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java +++ b/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()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index 81e6cadbad..e7f4abba10 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/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(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index 3bbe7d617b..71f81f07eb 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/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); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java index 629bfa29c9..ea4249751e 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java +++ b/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());