diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 447279f30b..7f0a8fe94e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -27,6 +27,7 @@ import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupCallback; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; @@ -81,6 +82,11 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements Assert.notNull(store); Assert.notNull(processor); this.store = store; + store.registerExpiryCallback(new MessageGroupCallback() { + public void execute(MessageGroup group) { + forceComplete(group); + } + }); this.outputProcessor = processor; this.correlationStrategy = correlationStrategy == null ? new HeaderAttributeCorrelationStrategy( MessageHeaders.CORRELATION_ID) : correlationStrategy; @@ -198,12 +204,12 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } // TODO: INT-958 - arrange for this to be called if user desires, e.g. periodically - public final boolean forceComplete(Object correlationKey) { + private final boolean forceComplete(MessageGroup group) { + Object correlationKey = group.getCorrelationKey(); Object lock = getLock(correlationKey); synchronized (lock) { - MessageGroup group = store.getMessageGroup(correlationKey); if (group.size() > 0) { // last chance for normal completion if (releaseStrategy.canRelease(group)) { @@ -234,7 +240,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements return true; } return false; + } + } private Object getLock(Object correlationKey) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroup.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroup.java index 4ae3e36d36..b13ba286be 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroup.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroup.java @@ -38,6 +38,8 @@ public interface MessageGroup { void mark(); - public Message getOne(); + Message getOne(); + + long getTimestamp(); } \ No newline at end of file diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupCallback.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupCallback.java new file mode 100755 index 0000000000..320b0940a3 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupCallback.java @@ -0,0 +1,26 @@ +/* + * Copyright 2002-2010 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.store; + +/** + * @author Dave Syer + * + * @since 2.0 + * + */ +public interface MessageGroupCallback { + + void execute(MessageGroup group); + +} 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 c8ad2beef4..3776edc4be 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 @@ -1,17 +1,14 @@ /* * 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. + * + * 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.store; @@ -61,4 +58,22 @@ public interface MessageGroupStore { */ void removeMessageGroup(Object correlationKey); + /** + * Register a callback for when a message group is expired through {@link #expireMessageGroups(long)}. + * + * @param callback a callback to execute when a message group is cleaned up + */ + 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. + * + * @param timestamp the timestamp threshold to use + * @return the number of message groups expired + * + * @see #registerExpiryCallback(MessageGroupCallback) + */ + int expireMessageGroups(long timestamp); + } \ No newline at end of file diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index f4134b806e..8b56b1c0e7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java @@ -38,14 +38,11 @@ public class SimpleMessageGroup implements MessageGroup { public final Collection> unmarked = new HashSet>(); + private final long timestamp; + public SimpleMessageGroup(Object correlationKey) { this.correlationKey = correlationKey; - } - - public SimpleMessageGroup(MessageGroup template) { - this.correlationKey = template.getCorrelationKey(); - this.marked.addAll(template.getMarked()); - this.unmarked.addAll(template.getUnmarked()); + this.timestamp = System.currentTimeMillis(); } public SimpleMessageGroup(Collection> originalMessages, @@ -55,6 +52,17 @@ public class SimpleMessageGroup implements MessageGroup { add(message); } } + + public SimpleMessageGroup(MessageGroup template) { + this.correlationKey = template.getCorrelationKey(); + this.marked.addAll(template.getMarked()); + this.unmarked.addAll(template.getUnmarked()); + this.timestamp = template.getTimestamp(); + } + + public long getTimestamp() { + return timestamp; + } public boolean add(Message message) { if (isMember(message)) { 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 90f999003b..3f17dc8ef5 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 @@ -1,25 +1,26 @@ /* * Copyright 2002-2009 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. + * + * 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.store; +import java.util.Collection; +import java.util.LinkedHashSet; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessagingException; import org.springframework.integration.util.UpperBound; @@ -37,12 +38,16 @@ import org.springframework.util.Assert; */ public class SimpleMessageStore implements MessageStore, MessageGroupStore { + private static final Log logger = LogFactory.getLog(SimpleMessageStore.class); + private final ConcurrentMap> idToMessage; private final ConcurrentMap correlationToMessageGroup; private final UpperBound upperBound; + private Collection expiryCallbacks = new LinkedHashSet(); + /** * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given * capacity is less than 1. @@ -77,36 +82,72 @@ public class SimpleMessageStore implements MessageStore, MessageGroupStore { if (key != null) { upperBound.release(); return this.idToMessage.remove(key); - } - else + } else return null; } public MessageGroup getMessageGroup(Object correlationId) { Assert.notNull(correlationId, "'correlationKey' must not be null"); - MessageGroup collection = correlationToMessageGroup.get(correlationId); - if (collection == null) { + SimpleMessageGroup group = correlationToMessageGroup.get(correlationId); + if (group == null) { return new SimpleMessageGroup(correlationId); } - return new SimpleMessageGroup(collection); + return new SimpleMessageGroup(group); } public void addMessageToGroup(Object correlationId, Message message) { getMessageGroupInternal(correlationId).add(message); } - - public void markMessageGroup(MessageGroup group) { + + public void markMessageGroup(MessageGroup group) { Object correlationId = group.getCorrelationKey(); MessageGroup internal = getMessageGroupInternal(correlationId); internal.mark(); - group.mark(); + group.mark(); } public void removeMessageGroup(Object correlationId) { correlationToMessageGroup.remove(correlationId); } - private MessageGroup getMessageGroupInternal(Object correlationId) { + public void registerExpiryCallback(MessageGroupCallback callback) { + expiryCallbacks.add(callback); + } + + public int expireMessageGroups(long timestamp) { + int count = 0; + for (MessageGroup group : correlationToMessageGroup.values()) { + if (group.getTimestamp() < timestamp) { + count++; + expire(group); + removeMessageGroup(group.getCorrelationKey()); + } + } + return count; + } + + private void expire(MessageGroup group) { + + RuntimeException exception = null; + + for (MessageGroupCallback callback : expiryCallbacks) { + try { + callback.execute(group); + } catch (RuntimeException e) { + if (exception == null) { + exception = e; + } + logger.error("Exception in expiry callback", e); + } + } + + if (exception != null) { + throw exception; + } + + } + + private SimpleMessageGroup getMessageGroupInternal(Object correlationId) { if (!correlationToMessageGroup.containsKey(correlationId)) { correlationToMessageGroup.putIfAbsent(correlationId, new SimpleMessageGroup(correlationId)); } 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 bcb9f6f1e4..4674906ab9 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 @@ -1,17 +1,14 @@ /* * Copyright 2002-2009 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. + * + * 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; @@ -45,214 +42,211 @@ import org.springframework.integration.store.SimpleMessageStore; */ public class AggregatorTests { - private CorrelatingMessageHandler aggregator; + private CorrelatingMessageHandler aggregator; + private SimpleMessageStore store = new SimpleMessageStore(50); - @Before - public void configureAggregator() { - this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), new SimpleMessageStore(50)); - } + @Before + public void configureAggregator() { + this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), store); + } - @Test - public void testCompleteGroupWithinTimeout() throws InterruptedException { - QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); - Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); - Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); - CountDownLatch latch = new CountDownLatch(3); - this.aggregator.handleMessage(message1); - this.aggregator.handleMessage(message2); - this.aggregator.handleMessage(message3); - latch.await(1000, TimeUnit.MILLISECONDS); - Message reply = replyChannel.receive(2000); - assertNotNull(reply); - assertEquals(reply.getPayload(), 105); - } + @Test + public void testCompleteGroupWithinTimeout() throws InterruptedException { + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); + Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); + Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); + CountDownLatch latch = new CountDownLatch(3); + this.aggregator.handleMessage(message1); + this.aggregator.handleMessage(message2); + this.aggregator.handleMessage(message3); + latch.await(1000, TimeUnit.MILLISECONDS); + Message reply = replyChannel.receive(2000); + assertNotNull(reply); + assertEquals(reply.getPayload(), 105); + } - @Test - public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException { - QueueChannel discardChannel = new QueueChannel(); - this.aggregator.setTimeout(50); - this.aggregator.setReaperInterval(10); - this.aggregator.setDiscardChannel(discardChannel); - QueueChannel replyChannel = new QueueChannel(); - Message message = createMessage(3, "ABC", 2, 1, replyChannel, null); - this.aggregator.handleMessage(message); - this.aggregator.forceComplete("ABC"); - Message reply = replyChannel.receive(100); - assertNull("No message should have been sent normally", reply); - Message discardedMessage = discardChannel.receive(1000); - assertNotNull("A message should have been discarded", discardedMessage); - assertEquals(message, discardedMessage); - } + @Test + public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException { + QueueChannel discardChannel = new QueueChannel(); + this.aggregator.setTimeout(50); + this.aggregator.setReaperInterval(10); + this.aggregator.setDiscardChannel(discardChannel); + QueueChannel replyChannel = new QueueChannel(); + Message message = createMessage(3, "ABC", 2, 1, replyChannel, null); + this.aggregator.handleMessage(message); + this.store.expireMessageGroups(System.currentTimeMillis() + 10000); + Message reply = replyChannel.receive(100); + assertNull("No message should have been sent normally", reply); + Message discardedMessage = discardChannel.receive(1000); + assertNotNull("A message should have been discarded", discardedMessage); + assertEquals(message, discardedMessage); + } - @Test - public void testShouldSendPartialResultOnTimeoutTrue() throws InterruptedException { - this.aggregator.setSendPartialResultOnTimeout(true); - QueueChannel replyChannel = new QueueChannel(); - 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.aggregator.forceComplete("ABC"); - Message reply = replyChannel.receive(0); - assertNotNull("A reply message should have been received", reply); - assertEquals(15, reply.getPayload()); - } + @Test + public void testShouldSendPartialResultOnTimeoutTrue() throws InterruptedException { + this.aggregator.setSendPartialResultOnTimeout(true); + QueueChannel replyChannel = new QueueChannel(); + 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(System.currentTimeMillis() + 10000); + Message reply = replyChannel.receive(0); + assertNotNull("A reply message should have been received", reply); + assertEquals(15, reply.getPayload()); + } - @Test - public void testMultipleGroupsSimultaneously() throws InterruptedException { - QueueChannel replyChannel1 = new QueueChannel(); - QueueChannel replyChannel2 = new QueueChannel(); - Message message1 = createMessage(3, "ABC", 3, 1, replyChannel1, null); - Message message2 = createMessage(5, "ABC", 3, 2, replyChannel1, null); - Message message3 = createMessage(7, "ABC", 3, 3, replyChannel1, null); - Message message4 = createMessage(11, "XYZ", 3, 1, replyChannel2, null); - Message message5 = createMessage(13, "XYZ", 3, 2, replyChannel2, null); - Message message6 = createMessage(17, "XYZ", 3, 3, replyChannel2, null); - aggregator.handleMessage(message1); - aggregator.handleMessage(message5); - aggregator.handleMessage(message3); - aggregator.handleMessage(message6); - aggregator.handleMessage(message4); - aggregator.handleMessage(message2); - @SuppressWarnings("unchecked") - Message reply1 = (Message) replyChannel1.receive(500); - assertNotNull(reply1); - assertThat(reply1.getPayload(), is(105)); - @SuppressWarnings("unchecked") - Message reply2 = (Message) replyChannel2.receive(500); - assertNotNull(reply2); - assertThat(reply2.getPayload(), is(2431)); - } + @Test + public void testMultipleGroupsSimultaneously() throws InterruptedException { + QueueChannel replyChannel1 = new QueueChannel(); + QueueChannel replyChannel2 = new QueueChannel(); + Message message1 = createMessage(3, "ABC", 3, 1, replyChannel1, null); + Message message2 = createMessage(5, "ABC", 3, 2, replyChannel1, null); + Message message3 = createMessage(7, "ABC", 3, 3, replyChannel1, null); + Message message4 = createMessage(11, "XYZ", 3, 1, replyChannel2, null); + Message message5 = createMessage(13, "XYZ", 3, 2, replyChannel2, null); + Message message6 = createMessage(17, "XYZ", 3, 3, replyChannel2, null); + aggregator.handleMessage(message1); + aggregator.handleMessage(message5); + aggregator.handleMessage(message3); + aggregator.handleMessage(message6); + aggregator.handleMessage(message4); + aggregator.handleMessage(message2); + @SuppressWarnings("unchecked") + Message reply1 = (Message) replyChannel1.receive(500); + assertNotNull(reply1); + assertThat(reply1.getPayload(), is(105)); + @SuppressWarnings("unchecked") + Message reply2 = (Message) replyChannel2.receive(500); + assertNotNull(reply2); + assertThat(reply2.getPayload(), is(2431)); + } - @Test - @Ignore - //dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE) - public void testTrackedCorrelationIdsCapacityAtLimit() { - QueueChannel replyChannel = new QueueChannel(); - QueueChannel discardChannel = new QueueChannel(); - //this.aggregator.setTrackedCorrelationIdCapacity(3); - this.aggregator.setDiscardChannel(discardChannel); - this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); - assertEquals(1, replyChannel.receive(100).getPayload()); - this.aggregator.handleMessage(createMessage(3, 2, 1, 1, replyChannel, null)); - assertEquals(3, replyChannel.receive(100).getPayload()); - this.aggregator.handleMessage(createMessage(4, 3, 1, 1, replyChannel, null)); - assertEquals(4, replyChannel.receive(100).getPayload()); - //next message with same correllation ID is discarded - this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel, null)); - assertEquals(2, discardChannel.receive(100).getPayload()); - } + @Test + @Ignore + // dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE) + public void testTrackedCorrelationIdsCapacityAtLimit() { + QueueChannel replyChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); + // this.aggregator.setTrackedCorrelationIdCapacity(3); + this.aggregator.setDiscardChannel(discardChannel); + this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); + assertEquals(1, replyChannel.receive(100).getPayload()); + this.aggregator.handleMessage(createMessage(3, 2, 1, 1, replyChannel, null)); + assertEquals(3, replyChannel.receive(100).getPayload()); + this.aggregator.handleMessage(createMessage(4, 3, 1, 1, replyChannel, null)); + assertEquals(4, replyChannel.receive(100).getPayload()); + // next message with same correllation ID is discarded + this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel, null)); + assertEquals(2, discardChannel.receive(100).getPayload()); + } - @Test - @Ignore - //dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE) - public void testTrackedCorrelationIdsCapacityPassesLimit() { - QueueChannel replyChannel = new QueueChannel(); - QueueChannel discardChannel = new QueueChannel(); - //this.aggregator.setTrackedCorrelationIdCapacity(3); - this.aggregator.setDiscardChannel(discardChannel); - this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); - assertEquals(1, replyChannel.receive(100).getPayload()); - this.aggregator.handleMessage(createMessage(2, 2, 1, 1, replyChannel, null)); - assertEquals(2, replyChannel.receive(100).getPayload()); - this.aggregator.handleMessage(createMessage(3, 3, 1, 1, replyChannel, null)); - assertEquals(3, replyChannel.receive(100).getPayload()); - this.aggregator.handleMessage(createMessage(4, 4, 1, 1, replyChannel, null)); - assertEquals(4, replyChannel.receive(100).getPayload()); - this.aggregator.handleMessage(createMessage(5, 1, 1, 1, replyChannel, null)); - assertEquals(5, replyChannel.receive(100).getPayload()); - assertNull(discardChannel.receive(0)); - } + @Test + @Ignore + // dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE) + public void testTrackedCorrelationIdsCapacityPassesLimit() { + QueueChannel replyChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); + // this.aggregator.setTrackedCorrelationIdCapacity(3); + this.aggregator.setDiscardChannel(discardChannel); + this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); + assertEquals(1, replyChannel.receive(100).getPayload()); + this.aggregator.handleMessage(createMessage(2, 2, 1, 1, replyChannel, null)); + assertEquals(2, replyChannel.receive(100).getPayload()); + this.aggregator.handleMessage(createMessage(3, 3, 1, 1, replyChannel, null)); + assertEquals(3, replyChannel.receive(100).getPayload()); + this.aggregator.handleMessage(createMessage(4, 4, 1, 1, replyChannel, null)); + assertEquals(4, replyChannel.receive(100).getPayload()); + this.aggregator.handleMessage(createMessage(5, 1, 1, 1, replyChannel, null)); + assertEquals(5, replyChannel.receive(100).getPayload()); + assertNull(discardChannel.receive(0)); + } - @Test(expected = MessageHandlingException.class) - public void testExceptionThrownIfNoCorrelationId() throws InterruptedException { - Message message = createMessage(3, null, 2, 1, new QueueChannel(), null); - this.aggregator.handleMessage(message); - } + @Test(expected = MessageHandlingException.class) + public void testExceptionThrownIfNoCorrelationId() throws InterruptedException { + Message message = createMessage(3, null, 2, 1, new QueueChannel(), null); + this.aggregator.handleMessage(message); + } - @Test - public void testAdditionalMessageAfterCompletion() throws InterruptedException { - QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); - Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); - Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); - Message message4 = createMessage(7, "ABC", 3, 3, replyChannel, null); - CountDownLatch latch = new CountDownLatch(4); - this.aggregator.handleMessage(message1); - this.aggregator.handleMessage(message2); - this.aggregator.handleMessage(message3); - this.aggregator.handleMessage(message4); - latch.await(1000, TimeUnit.MILLISECONDS); - //small wait to make sure the fourth message is received - Thread.sleep(10); - Message reply = replyChannel.receive(0); - assertNotNull("A message should be aggregated", reply); - assertThat(((Integer) reply.getPayload()), is(105)); - } + @Test + public void testAdditionalMessageAfterCompletion() throws InterruptedException { + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); + Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); + Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); + Message message4 = createMessage(7, "ABC", 3, 3, replyChannel, null); + CountDownLatch latch = new CountDownLatch(4); + this.aggregator.handleMessage(message1); + this.aggregator.handleMessage(message2); + this.aggregator.handleMessage(message3); + this.aggregator.handleMessage(message4); + latch.await(1000, TimeUnit.MILLISECONDS); + // small wait to make sure the fourth message is received + Thread.sleep(10); + Message reply = replyChannel.receive(0); + assertNotNull("A message should be aggregated", reply); + assertThat(((Integer) reply.getPayload()), is(105)); + } - @Test - public void shouldRejectDuplicatedSequenceNumbers() throws InterruptedException { - QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); - Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); - Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); - Message message4 = createMessage(7, "ABC", 3, 3, replyChannel, null); - CountDownLatch latch = new CountDownLatch(4); - this.aggregator.handleMessage(message1); - this.aggregator.handleMessage(message3); - //duplicated sequence number, either message3 or message4 should be rejected - this.aggregator.handleMessage(message4); - this.aggregator.handleMessage(message2); - latch.await(1000, TimeUnit.MILLISECONDS); - Message reply = replyChannel.receive(0); - assertNotNull("A message should be aggregated", reply); - assertThat(((Integer) reply.getPayload()), is(105)); - } + @Test + public void shouldRejectDuplicatedSequenceNumbers() throws InterruptedException { + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); + Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); + Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); + Message message4 = createMessage(7, "ABC", 3, 3, replyChannel, null); + CountDownLatch latch = new CountDownLatch(4); + this.aggregator.handleMessage(message1); + this.aggregator.handleMessage(message3); + // duplicated sequence number, either message3 or message4 should be rejected + this.aggregator.handleMessage(message4); + this.aggregator.handleMessage(message2); + latch.await(1000, TimeUnit.MILLISECONDS); + Message reply = replyChannel.receive(0); + assertNotNull("A message should be aggregated", reply); + assertThat(((Integer) reply.getPayload()), is(105)); + } - @Test - public void testNullReturningAggregator() throws InterruptedException { - this.aggregator = new CorrelatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore(50)); - QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); - Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); - Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); - this.aggregator.handleMessage(message1); - this.aggregator.handleMessage(message2); - this.aggregator.handleMessage(message3); - Message reply = replyChannel.receive(500); - assertNull(reply); - } + @Test + public void testNullReturningAggregator() throws InterruptedException { + this.aggregator = new CorrelatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore(50)); + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); + Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); + Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); + this.aggregator.handleMessage(message1); + this.aggregator.handleMessage(message2); + this.aggregator.handleMessage(message3); + Message reply = replyChannel.receive(500); + assertNull(reply); + } + private static Message createMessage(Object payload, Object correlationId, int sequenceSize, int sequenceNumber, + MessageChannel replyChannel, String predefinedId) { + MessageBuilder builder = MessageBuilder.withPayload(payload).setCorrelationId(correlationId) + .setSequenceSize(sequenceSize).setSequenceNumber(sequenceNumber).setReplyChannel(replyChannel); + if (predefinedId != null) { + builder.setHeader(MessageHeaders.ID, predefinedId); + } + return builder.build(); + } - private static Message createMessage(Object payload, Object correlationId, - int sequenceSize, int sequenceNumber, MessageChannel replyChannel, String predefinedId) { - MessageBuilder builder = MessageBuilder.withPayload(payload) - .setCorrelationId(correlationId) - .setSequenceSize(sequenceSize) - .setSequenceNumber(sequenceNumber) - .setReplyChannel(replyChannel); - if (predefinedId != null) { - builder.setHeader(MessageHeaders.ID, predefinedId); - } - return builder.build(); - } + private class MultiplyingProcessor implements MessageGroupProcessor { + public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, + MessageChannel outputChannel) { + Integer product = 1; + for (Message message : group.getUnmarked()) { + product *= (Integer) message.getPayload(); + } + channelTemplate.send(MessageBuilder.withPayload(product).build(), outputChannel); + } + } - private class MultiplyingProcessor implements MessageGroupProcessor { - public void processAndSend(MessageGroup group, - MessageChannelTemplate channelTemplate, MessageChannel outputChannel - ) { - Integer product = 1; - for (Message message : group.getUnmarked()) { - product *= (Integer) message.getPayload(); - } - channelTemplate.send(MessageBuilder.withPayload(product).build(), outputChannel); - } - } - - private class NullReturningMessageProcessor implements MessageGroupProcessor { - public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { - //noop - } - } + private class NullReturningMessageProcessor implements MessageGroupProcessor { + public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, + MessageChannel outputChannel) { + // noop + } + } } 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 c36ade6438..7504a92389 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 @@ -39,6 +39,7 @@ import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; /** @@ -52,11 +53,12 @@ public class ConcurrentAggregatorTests { private CorrelatingMessageHandler aggregator; + private MessageGroupStore store = new SimpleMessageStore(); + @Before public void configureAggregator() { this.taskExecutor = new SimpleAsyncTaskExecutor(); - this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), new SimpleMessageStore( - 50)); + this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), store); } @Test @@ -118,7 +120,7 @@ public class ConcurrentAggregatorTests { .getCount()); Message reply = replyChannel.receive(100); assertNull("No message should have been sent normally", reply); - aggregator.forceComplete("ABC"); + this.store.expireMessageGroups(System.currentTimeMillis()+10000); Message discardedMessage = discardChannel.receive(100); assertNotNull("A message should have been discarded", discardedMessage); assertEquals(message, discardedMessage); @@ -141,7 +143,7 @@ public class ConcurrentAggregatorTests { latch.await(300, TimeUnit.MILLISECONDS); assertEquals("handlers should have been invoked within time limit", 0, latch.getCount()); - this.aggregator.forceComplete("ABC"); + this.store.expireMessageGroups(System.currentTimeMillis()+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 4e6954e414..81e6cadbad 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 @@ -17,7 +17,6 @@ package org.springframework.integration.aggregator; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; import static org.mockito.Matchers.eq; import static org.mockito.Matchers.isA; import static org.mockito.Mockito.doAnswer; @@ -40,6 +39,7 @@ import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.test.util.ReflectionTestUtils; @@ -64,9 +64,11 @@ public class CorrelatingMessageHandlerTests { @Mock private MessageChannel outputChannel; + private MessageGroupStore store = new SimpleMessageStore(); + @Before public void initializeSubject() { - handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, + handler = new CorrelatingMessageHandler(processor, store, correlationStrategy, ReleaseStrategy); handler.setOutputChannel(outputChannel); doAnswer(new DoesNothing()).when(processor).processAndSend(isA(SimpleMessageGroup.class), @@ -127,7 +129,7 @@ public class CorrelatingMessageHandlerTests { }); Thread.sleep(20); - assertFalse(handler.forceComplete("key")); + assertEquals(0, store.expireMessageGroups(System.currentTimeMillis()+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 b797f4986b..3bbe7d617b 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,8 +136,7 @@ public class ResequencerTests { this.resequencer.setTimeout(90000); this.resequencer.handleMessage(message1); this.resequencer.handleMessage(message2); - // this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC")); - this.resequencer.forceComplete("ABC"); + assertEquals(1, store.expireMessageGroups(System.currentTimeMillis()+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 5fcf49aeac..629bfa29c9 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 @@ -21,6 +21,9 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertThat; +import java.util.ArrayList; +import java.util.List; + import org.junit.Test; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessagingException; @@ -75,4 +78,25 @@ public class SimpleMessageStoreTests { assertNotSame(store.getMessageGroup("bar"), store.getMessageGroup("bar")); } + @Test + public void shouldExpireMessageGroup() throws Exception { + + SimpleMessageStore store = new SimpleMessageStore(); + final List list = new ArrayList(); + store.registerExpiryCallback(new MessageGroupCallback() { + public void execute(MessageGroup group) { + list.add(group.getOne().getPayload().toString()); + } + }); + + Message testMessage1 = MessageBuilder.withPayload("foo").build(); + store.addMessageToGroup("bar", testMessage1); + assertEquals(1, store.getMessageGroup("bar").size()); + + store.expireMessageGroups(System.currentTimeMillis()+10000); + assertEquals("[foo]", list.toString()); + assertEquals(0, store.getMessageGroup("bar").size()); + + } + }