diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java index 9e55e2d0e0..c93ea9fd9e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java @@ -16,7 +16,6 @@ package org.springframework.integration.aggregator; import org.springframework.context.Lifecycle; -import org.springframework.integration.aggregator.*; import org.springframework.integration.channel.ChannelResolutionException; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.NullChannel; @@ -32,6 +31,7 @@ import org.springframework.util.Assert; import java.util.List; import java.util.Queue; import java.util.concurrent.*; +import java.util.concurrent.locks.ReentrantLock; /** * MessageHandler that holds a buffer of messages in a MessageStore @@ -42,8 +42,7 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L private MessageStore store = new SimpleMessageStore(100); private final CorrelationStrategy correlationStrategy; - //TODO decide if we still support tracking capacity, and if this needs to be moved into the Store instead - private final Queue trackedCorrellationIds = new LinkedBlockingQueue(); + private final IdTracker tracker = new IdTracker(); private final CompletionStrategy completionStrategy; private MessagesProcessor outputProcessor; private MessageChannel outputChannel; @@ -59,8 +58,8 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L public BufferingMessageHandler(MessageStore store, CorrelationStrategy correlationStrategy, - CompletionStrategy completionStrategy, MessagesProcessor processor - ) { + CompletionStrategy completionStrategy, + MessagesProcessor processor) { Assert.notNull(store); Assert.notNull(correlationStrategy); Assert.notNull(completionStrategy); @@ -109,12 +108,16 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L @Override protected void handleMessageInternal(Message message) throws Exception { Object correlationKey = correlationStrategy.getCorrelationKey(message); - if (!trackedCorrellationIds.contains(correlationKey)) { - store(message, correlationKey); - List> all = store.getAll(correlationKey); - complete(correlationKey, all, this.resolveReplyChannel(message)); - } else { - discardChannel.send(message); + try { + if (tracker.aquireLockFor(correlationKey)) { + store(message, correlationKey); + List> all = store.getAll(correlationKey); + complete(correlationKey, all, this.resolveReplyChannel(message)); + } else { + discardChannel.send(message); + } + } finally { + tracker.unlock(correlationKey); } } @@ -127,13 +130,6 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L return processed; } - private void pushCorrellationId(Queue trackedCorrellationIds, Object correlationKey) { - while (!trackedCorrellationIds.offer(correlationKey)) { - //make room in the queue - trackedCorrellationIds.poll(); - } - } - private BufferedMessagesCallback deleteOrTrackCallback() { return new BufferedMessagesCallback() { public void onProcessingOf(Message... processedMessage) { @@ -141,8 +137,9 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L store.delete(message.getHeaders().getId()); } } + public void onCompletionOf(Object correlationKey) { - pushCorrellationId(trackedCorrellationIds, correlationKey); + tracker.pushCorrellationId(correlationKey); } }; } @@ -224,33 +221,34 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L } } - protected final void forceComplete(Object key) { - List> all = store.getAll(key); - if (all.size() > 0) { - //last chance for normal completion - MessageChannel outputChannel = resolveReplyChannel(all.get(0)); - boolean fullyCompleted = complete(key, all, outputChannel); - if (!fullyCompleted) { - if (sendPartialResultOnTimeout) { - if (logger.isInfoEnabled()) { - logger.info("Processing partially complete messages for key [" + key + "] to: " + outputChannel); - } - outputProcessor.processAndSend(key, all, outputChannel, deleteOrTrackCallback()); - } else { - if (logger.isInfoEnabled()) { - logger.info("Discarding partially complete messages for key [" + key + "] to: " + discardChannel); - } - for (Message message : all) { - discardChannel.send(message); - store.delete(message.getHeaders().getId()); - } + + protected final void forceComplete(Object key) { + List> all = store.getAll(key); + if (all.size() > 0) { + //last chance for normal completion + MessageChannel outputChannel = resolveReplyChannel(all.get(0)); + boolean fullyCompleted = complete(key, all, outputChannel); + if (!fullyCompleted) { + if (sendPartialResultOnTimeout) { + if (logger.isInfoEnabled()) { + logger.info("Processing partially complete messages for key [" + key + "] to: " + outputChannel); + } + outputProcessor.processAndSend(key, all, outputChannel, deleteOrTrackCallback()); + } else { + if (logger.isInfoEnabled()) { + logger.info("Discarding partially complete messages for key [" + key + "] to: " + discardChannel); + } + for (Message message : all) { + discardChannel.send(message); + store.delete(message.getHeaders().getId()); } } } } + } - private class DelayedKey implements Delayed { + private final class DelayedKey implements Delayed { private Object key; private Long releaseTime; private TimeUnit unit = TimeUnit.MILLISECONDS; @@ -273,4 +271,38 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L return key; } } + + private final class IdTracker { + private ConcurrentMap trackerLocks = new ConcurrentHashMap(); + private final Queue trackedCorrellationIds = new LinkedBlockingQueue(); + + private void pushCorrellationId(Object correlationKey) { + while (!trackedCorrellationIds.offer(correlationKey)) { + //make room in the queue + trackedCorrellationIds.poll(); + } + trackerLocks.remove(correlationKey); + } + + private boolean aquireLockFor(Object correlationKey) { + ReentrantLock lock = trackerLocks.get(correlationKey); + if (lock == null) { + if (trackedCorrellationIds.contains(correlationKey)) { + return false; + } + lock = new ReentrantLock(); + ReentrantLock original = trackerLocks.putIfAbsent(correlationKey, lock); + lock = original == null ? lock : original; + } + lock.lock(); + return true; + } + + private void unlock(Object correlationKey) { + ReentrantLock lock = trackerLocks.get(correlationKey); + if (lock != null && lock.isHeldByCurrentThread()) { + lock.unlock(); + } + } + } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java index 237fb60197..8c612d21ea 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java @@ -22,16 +22,20 @@ import static org.hamcrest.CoreMatchers.*; import org.junit.Before; import org.junit.Test; import org.junit.Ignore; +import org.junit.matchers.JUnitMatchers; +import static org.junit.matchers.JUnitMatchers.*; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.aggregator.BufferingMessageHandler; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.test.util.TestUtils; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import java.util.Arrays; +import static java.util.Arrays.*; + /** * @author Marius Bogoevici * @author Alex Peters @@ -39,237 +43,236 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; */ public class NewResequencerTests { - private BufferingMessageHandler resequencer; + private BufferingMessageHandler resequencer; - private ThreadPoolTaskScheduler taskScheduler; + private ThreadPoolTaskScheduler taskScheduler; private DefaultResequencerStrategies resequencerStrategies; @Before - public void configureResequencer() { + public void configureResequencer() { this.resequencerStrategies = new DefaultResequencerStrategies(); MessageStore store = new SimpleMessageStore(30); this.resequencer = new BufferingMessageHandler(store, resequencerStrategies, resequencerStrategies, resequencerStrategies); - this.taskScheduler = TestUtils.createTaskScheduler(10); - this.resequencer.setTaskScheduler(taskScheduler); - this.taskScheduler.afterPropertiesSet(); - this.resequencer.start(); - } + this.taskScheduler = TestUtils.createTaskScheduler(10); + this.resequencer.setTaskScheduler(taskScheduler); + this.taskScheduler.afterPropertiesSet(); + this.resequencer.start(); + } - @Test - public void testBasicResequencing() throws InterruptedException { - QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage("123", "ABC", 3, 3, replyChannel); - Message message2 = createMessage("456", "ABC", 3, 1, replyChannel); - Message message3 = createMessage("789", "ABC", 3, 2, replyChannel); - this.resequencer.handleMessage(message1); - this.resequencer.handleMessage(message3); - this.resequencer.handleMessage(message2); - Message reply1 = replyChannel.receive(0); - Message reply2 = replyChannel.receive(0); - Message reply3 = replyChannel.receive(0); - assertNotNull(reply1); - assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); - assertNotNull(reply2); - assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); - assertNotNull(reply3); - assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); - } - - @Test - public void testResequencingWithDuplicateMessages() { - QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage("123", "ABC", 3, 3, replyChannel); - Message message2 = createMessage("456", "ABC", 3, 1, replyChannel); - Message message3 = createMessage("789", "ABC", 3, 2, replyChannel); - this.resequencer.handleMessage(message1); - this.resequencer.handleMessage(message3); - this.resequencer.handleMessage(message3); - this.resequencer.handleMessage(message2); - Message reply1 = replyChannel.receive(0); - Message reply2 = replyChannel.receive(0); - Message reply3 = replyChannel.receive(0); - assertNotNull(reply1); - assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); - assertNotNull(reply2); - assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); - assertNotNull(reply3); - assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); - } - - - - @Test - public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException { - this.resequencerStrategies.setReleasePartialSequences(true); - QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); - Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); - Message message3 = createMessage("789", "ABC", 4, 4, replyChannel); - Message message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel); - this.resequencer.handleMessage(message1); - this.resequencer.handleMessage(message2); - this.resequencer.handleMessage(message3); - Message reply1 = replyChannel.receive(0); - Message reply2 = replyChannel.receive(0); - Message reply3 = replyChannel.receive(0); - // only messages 1 and 2 should have been received by now - assertNotNull(reply1); - assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); - assertNotNull(reply2); - assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); - assertNull(reply3); - // when sending the last message, the whole sequence must have been sent - this.resequencer.handleMessage(message4); - reply3 = replyChannel.receive(0); - Message reply4 = replyChannel.receive(0); - assertNotNull(reply3); - assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); - assertNotNull(reply4); - assertEquals(new Integer(4), reply4.getHeaders().getSequenceNumber()); - } - - @Test - public void testResequencingWithDiscard() throws InterruptedException { + @Test + public void testBasicResequencing() throws InterruptedException { QueueChannel replyChannel = new QueueChannel(); - QueueChannel discardChannel = new QueueChannel(); - Message message1 = createMessage("123", "ABC", 3, 2, replyChannel); - Message message2 = createMessage("456", "ABC", 3, 1, replyChannel); - Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); - this.resequencer.setSendPartialResultOnTimeout(false); - this.resequencerStrategies.setReleasePartialSequences(false); - this.resequencer.setDiscardChannel(discardChannel); - this.resequencer.setTimeout(90000); - this.resequencer.handleMessage(message1); - this.resequencer.handleMessage(message2); - //this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC")); + Message message1 = createMessage("123", "ABC", 3, 3, replyChannel); + Message message2 = createMessage("456", "ABC", 3, 1, replyChannel); + Message message3 = createMessage("789", "ABC", 3, 2, replyChannel); + this.resequencer.handleMessage(message1); + this.resequencer.handleMessage(message3); + this.resequencer.handleMessage(message2); + Message reply1 = replyChannel.receive(0); + Message reply2 = replyChannel.receive(0); + Message reply3 = replyChannel.receive(0); + assertNotNull(reply1); + assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); + assertNotNull(reply2); + assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); + assertNotNull(reply3); + assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); + } + + @Test + public void testResequencingWithDuplicateMessages() { + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 3, 3, replyChannel); + Message message2 = createMessage("456", "ABC", 3, 1, replyChannel); + Message message3 = createMessage("789", "ABC", 3, 2, replyChannel); + this.resequencer.handleMessage(message1); + this.resequencer.handleMessage(message3); + this.resequencer.handleMessage(message3); + this.resequencer.handleMessage(message2); + Message reply1 = replyChannel.receive(0); + Message reply2 = replyChannel.receive(0); + Message reply3 = replyChannel.receive(0); + assertNotNull(reply1); + assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); + assertNotNull(reply2); + assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); + assertNotNull(reply3); + assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); + } + + + @Test + public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException { + this.resequencerStrategies.setReleasePartialSequences(true); + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); + Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); + Message message3 = createMessage("789", "ABC", 4, 4, replyChannel); + Message message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel); + this.resequencer.handleMessage(message1); + this.resequencer.handleMessage(message2); + this.resequencer.handleMessage(message3); + Message reply1 = replyChannel.receive(0); + Message reply2 = replyChannel.receive(0); + Message reply3 = replyChannel.receive(0); + // only messages 1 and 2 should have been received by now + assertNotNull(reply1); + assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); + assertNotNull(reply2); + assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); + assertNull(reply3); + // when sending the last message, the whole sequence must have been sent + this.resequencer.handleMessage(message4); + reply3 = replyChannel.receive(0); + Message reply4 = replyChannel.receive(0); + assertNotNull(reply3); + assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); + assertNotNull(reply4); + assertEquals(new Integer(4), reply4.getHeaders().getSequenceNumber()); + } + + @Test + public void testResequencingWithDiscard() throws InterruptedException { + QueueChannel replyChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 3, 2, replyChannel); + Message message2 = createMessage("456", "ABC", 3, 1, replyChannel); + Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); + this.resequencer.setSendPartialResultOnTimeout(false); + this.resequencerStrategies.setReleasePartialSequences(false); + this.resequencer.setDiscardChannel(discardChannel); + this.resequencer.setTimeout(90000); + this.resequencer.handleMessage(message1); + this.resequencer.handleMessage(message2); this.resequencer.forceComplete("ABC"); - Message reply1 = discardChannel.receive(0); - Message reply2 = discardChannel.receive(0); - Message reply3 = discardChannel.receive(0); - // only messages 1 and 2 should have been received by now - // messages need not be reordered - assertNotNull(reply1); - assertThat( reply1.getHeaders().getSequenceNumber(), is(new Integer(2))); - assertNotNull(reply2); - assertThat( reply2.getHeaders().getSequenceNumber(), is(new Integer(1))); - assertNull(reply3); - // when sending the last message, it waits in the buffer for retries of the other two - this.resequencer.handleMessage(message3); - reply3 = discardChannel.receive(0); - assertNull(reply3); + Message reply1 = discardChannel.receive(0); + Message reply2 = discardChannel.receive(0); + Message reply3 = discardChannel.receive(0); + // messages 1 and 2 should have been received by now in no particular order + assertNotNull(reply1); + assertNotNull(reply2); + Integer sequenceNo1 = reply1.getHeaders().getSequenceNumber(); + Integer sequenceNo2 = reply2.getHeaders().getSequenceNumber(); + assertThat(asList(sequenceNo1, sequenceNo2), hasItems(1, 2)); + assertNull(reply3); + // when sending the last message, it waits in the buffer for retries of the other two + this.resequencer.handleMessage(message3); + reply3 = discardChannel.receive(0); + assertNull(reply3); this.resequencer.handleMessage(message1); this.resequencer.handleMessage(message2); reply1 = replyChannel.receive(0); reply2 = replyChannel.receive(0); reply3 = replyChannel.receive(0); assertNotNull(reply1); - assertThat( reply1.getHeaders().getSequenceNumber(), is(new Integer(1))); + assertThat(reply1.getHeaders().getSequenceNumber(), is(new Integer(1))); assertNotNull(reply2); - assertThat( reply2.getHeaders().getSequenceNumber(), is(new Integer(2))); + assertThat(reply2.getHeaders().getSequenceNumber(), is(new Integer(2))); assertNotNull(reply3); - assertThat( reply3.getHeaders().getSequenceNumber(), is(new Integer(3))); - } + assertThat(reply3.getHeaders().getSequenceNumber(), is(new Integer(3))); + } - @Test - @Ignore //different sequence sizes are not supported - public void testResequencingWithDifferentSequenceSizes() throws InterruptedException { + @Test + @Ignore + //different sequence sizes are not supported + public void testResequencingWithDifferentSequenceSizes() throws InterruptedException { QueueChannel replyChannel = new QueueChannel(); - QueueChannel discardChannel = new QueueChannel(); - Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); - Message message2 = createMessage("456", "ABC", 5, 1, replyChannel); - this.resequencer.setSendPartialResultOnTimeout(false); - //this.resequencer.setReleasePartialSequences(false); - this.resequencer.setDiscardChannel(discardChannel); - this.resequencer.setTimeout(90000); - this.resequencer.handleMessage(message1); - this.resequencer.handleMessage(message2); - //this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC")); - Message reply1 = discardChannel.receive(0); - Message reply2 = discardChannel.receive(0); - // only messages 1 - with sequence number 2 - should have been received by now - // the other has been discarded - assertNotNull(reply1); - assertEquals(new Integer(2), reply1.getHeaders().getSequenceNumber()); - assertNull(reply2); - } + QueueChannel discardChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); + Message message2 = createMessage("456", "ABC", 5, 1, replyChannel); + this.resequencer.setSendPartialResultOnTimeout(false); + //this.resequencer.setReleasePartialSequences(false); + this.resequencer.setDiscardChannel(discardChannel); + this.resequencer.setTimeout(90000); + this.resequencer.handleMessage(message1); + this.resequencer.handleMessage(message2); + //this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC")); + Message reply1 = discardChannel.receive(0); + Message reply2 = discardChannel.receive(0); + // only messages 1 - with sequence number 2 - should have been received by now + // the other has been discarded + assertNotNull(reply1); + assertEquals(new Integer(2), reply1.getHeaders().getSequenceNumber()); + assertNull(reply2); + } - @Test - public void testResequencingWithWrongSequenceSizeAndNumber() throws InterruptedException { - QueueChannel replyChannel = new QueueChannel(); - QueueChannel discardChannel = new QueueChannel(); - Message message1 = createMessage("123", "ABC", 2, 4, replyChannel); - this.resequencer.setSendPartialResultOnTimeout(false); - //this.resequencer.setReleasePartialSequences(false); - this.resequencer.setDiscardChannel(discardChannel); - this.resequencer.setTimeout(90000); - this.resequencer.handleMessage(message1); - //this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC")); - Message reply1 = discardChannel.receive(0); - // No message has been received - the message has been rejected. - assertNull(reply1); - } + @Test + public void testResequencingWithWrongSequenceSizeAndNumber() throws InterruptedException { + QueueChannel replyChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 2, 4, replyChannel); + this.resequencer.setSendPartialResultOnTimeout(false); + //this.resequencer.setReleasePartialSequences(false); + this.resequencer.setDiscardChannel(discardChannel); + this.resequencer.setTimeout(90000); + this.resequencer.handleMessage(message1); + //this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC")); + Message reply1 = discardChannel.receive(0); + // No message has been received - the message has been rejected. + assertNull(reply1); + } - @Test - public void testResequencingWithCompleteSequenceRelease() throws InterruptedException { - //this.resequencer.setReleasePartialSequences(false); - QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); - Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); - Message message3 = createMessage("789", "ABC", 4, 4, replyChannel); - Message message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel); - this.resequencer.handleMessage(message1); - this.resequencer.handleMessage(message2); - this.resequencer.handleMessage(message3); - Message reply1 = replyChannel.receive(0); - Message reply2 = replyChannel.receive(0); - Message reply3 = replyChannel.receive(0); - // no messages should have been received yet - assertNull(reply1); - assertNull(reply2); - assertNull(reply3); - // after sending the last message, the whole sequence should have been sent - this.resequencer.handleMessage(message4); - reply1 = replyChannel.receive(0); - reply2 = replyChannel.receive(0); - reply3 = replyChannel.receive(0); - Message reply4 = replyChannel.receive(0); - assertNotNull(reply1); - assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); - assertNotNull(reply2); - assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); - assertNotNull(reply3); - assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); - assertNotNull(reply4); - assertEquals(new Integer(4), reply4.getHeaders().getSequenceNumber()); - } + @Test + public void testResequencingWithCompleteSequenceRelease() throws InterruptedException { + //this.resequencer.setReleasePartialSequences(false); + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); + Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); + Message message3 = createMessage("789", "ABC", 4, 4, replyChannel); + Message message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel); + this.resequencer.handleMessage(message1); + this.resequencer.handleMessage(message2); + this.resequencer.handleMessage(message3); + Message reply1 = replyChannel.receive(0); + Message reply2 = replyChannel.receive(0); + Message reply3 = replyChannel.receive(0); + // no messages should have been received yet + assertNull(reply1); + assertNull(reply2); + assertNull(reply3); + // after sending the last message, the whole sequence should have been sent + this.resequencer.handleMessage(message4); + reply1 = replyChannel.receive(0); + reply2 = replyChannel.receive(0); + reply3 = replyChannel.receive(0); + Message reply4 = replyChannel.receive(0); + assertNotNull(reply1); + assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); + assertNotNull(reply2); + assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); + assertNotNull(reply3); + assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); + assertNotNull(reply4); + assertEquals(new Integer(4), reply4.getHeaders().getSequenceNumber()); + } - @Test - public void testRemovalOfBarrierWhenLastMessageOfSequenceArrives() { - QueueChannel replyChannel = new QueueChannel(); - String correlationId = "ABC"; - Message message1 = createMessage("123", correlationId, 1, 1, - replyChannel); - resequencer.handleMessage(message1); - //assertThat(resequencer.barriers.containsKey(correlationId), is(false)); - } + @Test + public void testRemovalOfBarrierWhenLastMessageOfSequenceArrives() { + QueueChannel replyChannel = new QueueChannel(); + String correlationId = "ABC"; + Message message1 = createMessage("123", correlationId, 1, 1, + replyChannel); + resequencer.handleMessage(message1); + //assertThat(resequencer.barriers.containsKey(correlationId), is(false)); + } - private static Message createMessage(String payload, Object correlationId, - int sequenceSize, int sequenceNumber, MessageChannel replyChannel) { + private static Message createMessage(String payload, Object correlationId, + int sequenceSize, int sequenceNumber, MessageChannel replyChannel) { return MessageBuilder.withPayload(payload) - .setCorrelationId(correlationId) - .setSequenceSize(sequenceSize) - .setSequenceNumber(sequenceNumber) - .setReplyChannel(replyChannel) - .build(); - } + .setCorrelationId(correlationId) + .setSequenceSize(sequenceSize) + .setSequenceNumber(sequenceNumber) + .setReplyChannel(replyChannel) + .build(); + } - @After - public void stopTaskScheduler() { - this.resequencer.stop(); - this.taskScheduler.destroy(); - } + @After + public void stopTaskScheduler() { + this.resequencer.stop(); + this.taskScheduler.destroy(); + } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java index 0acb157f91..941a55123f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java @@ -36,6 +36,7 @@ import org.springframework.integration.aggregator.MessagesProcessor; import org.springframework.integration.aggregator.BufferedMessagesCallback; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.test.annotation.Repeat; import java.util.Collection; import java.util.concurrent.CountDownLatch; @@ -236,7 +237,7 @@ public class NewAggregatorEndpointTests { this.aggregator.handleMessage(message); } - @Test + @Test public void testAdditionalMessageAfterCompletion() throws InterruptedException { this.aggregator.start(); QueueChannel replyChannel = new QueueChannel(); @@ -250,7 +251,9 @@ public class NewAggregatorEndpointTests { this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch)); latch.await(1000, TimeUnit.MILLISECONDS); - Message reply = replyChannel.receive(500); + //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)); }