diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategy.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategy.java new file mode 100755 index 0000000000..828601f414 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategy.java @@ -0,0 +1,81 @@ +/* + * Copyright 2002-2008 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. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.aggregator; + +import org.springframework.integration.core.Message; +import org.springframework.integration.store.MessageGroup; + +/** + * A {@link ReleaseStrategy} that releases all messages if any of the following is true: + * + * + * + * @author Dave Syer + * + * @since 2.0 + */ +public class TimeoutCountSequenceSizeReleaseStrategy implements ReleaseStrategy { + + /** + * Default timeout is one minute. + */ + public static final long DEFAULT_TIMEOUT = 60 * 1000; + + /** + * Default threshold is effectively infinite. + */ + public static final int DEFAULT_THRESHOLD = Integer.MAX_VALUE; + + private final int threshold; + + private final long timeout; + + public TimeoutCountSequenceSizeReleaseStrategy() { + this(DEFAULT_THRESHOLD, DEFAULT_TIMEOUT); + } + + /** + * @param threshold the number of messages to accept before releasing + * @param timeout the timeout for the release in milliseconds + */ + public TimeoutCountSequenceSizeReleaseStrategy(int threshold, long timeout) { + this.threshold = threshold; + this.timeout = timeout; + } + + public boolean canRelease(MessageGroup messages) { + long elapsedTime = System.currentTimeMillis() - findEarliestTimestamp(messages); + return messages.isComplete() || messages.getUnmarked().size() >= threshold || elapsedTime > timeout; + } + + /** + * @param messages the message group + * @return the earliest timestamp or Long.MAX_VALUE + */ + private long findEarliestTimestamp(MessageGroup messages) { + long result = Long.MAX_VALUE; + for (Message message : messages.getUnmarked()) { + long timestamp = message.getHeaders().getTimestamp(); + if (timestamp < result) { + result = timestamp; + } + } + return result; + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java index 690a158adf..bb54eb8a23 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java @@ -36,8 +36,8 @@ public class SequenceSizeReleaseStrategyTests { .setSequenceSize(2).build(); MessageGroup messages = new SimpleMessageGroup("FOO"); messages.add(message); - SequenceSizeReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy(); - assertFalse(ReleaseStrategy.canRelease(messages)); + SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); + assertFalse(releaseStrategy.canRelease(messages)); } @Test @@ -49,14 +49,14 @@ public class SequenceSizeReleaseStrategyTests { MessageGroup messages = new SimpleMessageGroup("FOO"); messages.add(message1); messages.add(message2); - SequenceSizeReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy(); - assertTrue(ReleaseStrategy.canRelease(messages)); + SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); + assertTrue(releaseStrategy.canRelease(messages)); } @Test public void testEmptyList() { - SequenceSizeReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy(); - assertTrue(ReleaseStrategy.canRelease(new SimpleMessageGroup("FOO"))); + SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); + assertTrue(releaseStrategy.canRelease(new SimpleMessageGroup("FOO"))); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategyTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategyTests.java new file mode 100755 index 0000000000..3b70f8596a --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategyTests.java @@ -0,0 +1,82 @@ +/* + * Copyright 2002-2008 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aggregator; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.SimpleMessageGroup; + +/** + * @author Dave Syer + */ +public class TimeoutCountSequenceSizeReleaseStrategyTests { + + @Test + public void testIncompleteList() { + Message message = MessageBuilder.withPayload("test1") + .setSequenceSize(2).build(); + MessageGroup messages = new SimpleMessageGroup("FOO"); + messages.add(message); + TimeoutCountSequenceSizeReleaseStrategy releaseStrategy = new TimeoutCountSequenceSizeReleaseStrategy(); + assertFalse(releaseStrategy.canRelease(messages)); + } + + @Test + public void testIncompleteListWithTimeout() { + Message message = MessageBuilder.withPayload("test1") + .setSequenceSize(2).build(); + MessageGroup messages = new SimpleMessageGroup("FOO"); + messages.add(message); + TimeoutCountSequenceSizeReleaseStrategy releaseStrategy = new TimeoutCountSequenceSizeReleaseStrategy(TimeoutCountSequenceSizeReleaseStrategy.DEFAULT_THRESHOLD, -100); + assertTrue(releaseStrategy.canRelease(messages)); + } + + @Test + public void testIncompleteListWithCount() { + Message message = MessageBuilder.withPayload("test1") + .setSequenceSize(2).build(); + MessageGroup messages = new SimpleMessageGroup("FOO"); + messages.add(message); + TimeoutCountSequenceSizeReleaseStrategy releaseStrategy = new TimeoutCountSequenceSizeReleaseStrategy(1, TimeoutCountSequenceSizeReleaseStrategy.DEFAULT_TIMEOUT); + assertTrue(releaseStrategy.canRelease(messages)); + } + + @Test + public void testCompleteList() { + Message message1 = MessageBuilder.withPayload("test1") + .setSequenceSize(2).build(); + Message message2 = MessageBuilder.withPayload("test2") + .setSequenceSize(2).build(); + MessageGroup messages = new SimpleMessageGroup("FOO"); + messages.add(message1); + messages.add(message2); + SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); + assertTrue(releaseStrategy.canRelease(messages)); + } + + @Test + public void testEmptyList() { + SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); + assertTrue(releaseStrategy.canRelease(new SimpleMessageGroup("FOO"))); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml index 00b78e0b28..025c6e034f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml @@ -1,67 +1,67 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +