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 e4672bce70..082492ee17 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 @@ -110,9 +110,14 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L Object correlationKey = correlationStrategy.getCorrelationKey(message); try { if (tracker.aquireLockFor(correlationKey)) { - store(message, correlationKey); - List> all = store.getAll(correlationKey); - complete(correlationKey, all, this.resolveReplyChannel(message, this.outputChannel, this.channelResolver)); + List> group = store.getAll(correlationKey); + if (noSupersedingMessage(message, group)) { + store(message, correlationKey); + group.add(message); + complete(correlationKey, group, this.resolveReplyChannel(message, this.outputChannel, this.channelResolver)); + } else { + discardChannel.send(message); + } } else { discardChannel.send(message); } @@ -121,6 +126,15 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L } } + private boolean noSupersedingMessage(Message message, List> group) { + for (Message member : group) { + if (member.getHeaders().getSequenceNumber() == message.getHeaders().getSequenceNumber()) { + return false; + } + } + return true; + } + private boolean complete(Object correlationKey, List> correlatedMessages, MessageChannel messageChannel) { boolean processed = false; if (completionStrategy.isComplete(correlatedMessages)) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java index 1fd601a9aa..8ba28e31f3 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java @@ -16,11 +16,6 @@ package org.springframework.integration.aggregator; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; - import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -30,6 +25,8 @@ import java.util.concurrent.TimeUnit; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.junit.Ignore; +import static org.junit.Assert.*; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.channel.QueueChannel; @@ -40,6 +37,7 @@ import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.StringMessage; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import static org.hamcrest.CoreMatchers.is; /** * @author Mark Fisher diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java index 1c967bf6b2..932cd998c5 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java @@ -55,8 +55,8 @@ public class BufferingMessageHandlerIntegrationTest { @Test public void completesWithoutReleasingIncompleteCorrellations() throws Exception { Message message1 = correlatedMessage(1, 2, 1); - Message message2 = correlatedMessage(2, 2, 2); - Message message1a = correlatedMessage(1, 2, 1); + Message message2 = correlatedMessage(2, 2, 1); + Message message1a = correlatedMessage(1, 2, 2); Message message2a = correlatedMessage(2, 2, 2); defaultHandler.handleMessage(message1); defaultHandler.handleMessage(message2); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java index 68f8dbe70d..277a5bc2a0 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java @@ -60,22 +60,21 @@ public class BufferingMessageHandlerTest { @Test public void bufferCompletesNormally() throws Exception { String correlationKey = "key"; - Message message1 = testMessage(1); - Message message2 = testMessage(2); + Message message1 = testMessage(1, 1); + Message message2 = testMessage(2, 2); List> storedMessages = new ArrayList>(); + when(store.getAll(correlationKey)).thenReturn(storedMessages); when(correlationStrategy.getCorrelationKey(isA(Message.class))) .thenReturn(correlationKey); when(completionStrategy.isComplete(storedMessages)).thenReturn(false); - storedMessages.add(message1); - when(store.getAll(correlationKey)).thenReturn(storedMessages); buffer.handleMessageInternal(message1); + storedMessages.add(message1); - storedMessages.add(message2); - when(store.getAll(correlationKey)).thenReturn(storedMessages); when(completionStrategy.isComplete(storedMessages)).thenReturn(true); buffer.handleMessageInternal(message2); + storedMessages.add(message2); verify(store).put(message1); verify(store).put(message2); @@ -87,8 +86,8 @@ public class BufferingMessageHandlerTest { processAndSend(eq(correlationKey), eq(storedMessages), eq(outputChannel), isA(BufferedMessagesCallback.class)); } - private Message testMessage(int id) { + private Message testMessage(int id, int sequenceNumber) { return MessageBuilder.withPayload("test").setHeader(MessageHeaders.ID, - id).build(); + id).setSequenceNumber(sequenceNumber).build(); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java new file mode 100644 index 0000000000..e366f51114 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java @@ -0,0 +1,287 @@ +/* + * 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. + */ + +package org.springframework.integration.aggregator; + +import org.junit.After; +import static org.junit.Assert.*; +import static org.hamcrest.CoreMatchers.*; +import org.junit.Before; +import org.junit.Test; +import org.junit.Ignore; +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessageHeaders; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageHandler; +import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.aggregator.BufferingMessageHandler; +import org.springframework.integration.aggregator.MessagesProcessor; +import org.springframework.integration.aggregator.BufferedMessagesCallback; +import org.springframework.integration.store.SimpleMessageStore; +import org.springframework.integration.config.StubTaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.Trigger; +import org.springframework.test.annotation.Repeat; + +import java.util.Collection; +import java.util.Date; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.ScheduledFuture; + +/** + * @author Mark Fisher + * @author Marius Bogoevici + * @author Iwein Fuld + */ +public class NewAggregatorEndpointTests { + + private BufferingMessageHandler aggregator; + + @Before + public void configureAggregator() { + this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor()); + } + + @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 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 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); + Message reply1 = (Message) replyChannel1.receive(500); + assertNotNull(reply1); + assertThat(reply1.getPayload(), is(105)); + Message reply2 = (Message) replyChannel2.receive(500); + assertNotNull(reply2); + assertThat(reply2.getPayload(), is(2431)); + } + + @Test + public void testDiscardChannelForTrackedCorrelationId() { + QueueChannel replyChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); + this.aggregator.setDiscardChannel(discardChannel); + this.aggregator.handleMessage(createMessage(1, "tracked", 1, 1, replyChannel, null)); + Message received1 = replyChannel.receive(0); + assertEquals(1, received1.getPayload()); + assertNotNull("Expected aggregated message, but got null", received1); + this.aggregator.handleMessage(createMessage(2, "tracked", 1, 1, replyChannel, null)); + Message received2 = discardChannel.receive(0); + assertNotNull("Expected discarded message, but got null", received2); + assertEquals(2, received2.getPayload()); + } + + @Test + @Ignore + //dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE) + public void testTrackedCorrelationIdsCapacityAtLimit() { + this.aggregator.start(); + 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() { + this.aggregator.start(); + 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 + 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 testNullReturningAggregator() throws InterruptedException { + this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor()); + 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 class MultiplyingProcessor implements MessagesProcessor { + public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, + MessageChannel outputChannel, BufferedMessagesCallback processedCallback + ) { + Integer product = 1; + for (Message message : messagesUpForProcessing) { + product *= (Integer) message.getPayload(); + } + outputChannel.send(MessageBuilder.withPayload(product).build()); + + processedCallback.onProcessingOf( + messagesUpForProcessing.toArray(new Message[messagesUpForProcessing.size()]) + ); + processedCallback.onCompletionOf(correlationKey); + } + } + + private class NullReturningMessageProcessor implements MessagesProcessor { + public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { + //noop + } + } +} 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/NewConcurrentAggregatorEndpointTests.java similarity index 94% rename from org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java rename to org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java index 941a55123f..c0f871e753 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/NewConcurrentAggregatorEndpointTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.aggregator.integration; +package org.springframework.integration.aggregator; import org.junit.After; import static org.junit.Assert.*; @@ -47,7 +47,7 @@ import java.util.concurrent.TimeUnit; * @author Marius Bogoevici * @author Iwein Fuld */ -public class NewAggregatorEndpointTests { +public class NewConcurrentAggregatorEndpointTests { private TaskExecutor taskExecutor; @@ -172,22 +172,6 @@ public class NewAggregatorEndpointTests { assertThat(reply2.getPayload(), is(2431)); } - @Test - public void testDiscardChannelForTrackedCorrelationId() { - this.aggregator.start(); - QueueChannel replyChannel = new QueueChannel(); - QueueChannel discardChannel = new QueueChannel(); - this.aggregator.setDiscardChannel(discardChannel); - this.aggregator.handleMessage(createMessage(1, "tracked", 1, 1, replyChannel, null)); - Message received1 = replyChannel.receive(100); - assertEquals(1, received1.getPayload()); - assertNotNull("Expected aggregated message, but got null", received1); - this.aggregator.handleMessage(createMessage(2, "tracked", 1, 1, replyChannel, null)); - Message received2 = discardChannel.receive(1000); - assertNotNull("Expected discarded message, but got null", received2); - assertEquals(2, received2.getPayload()); - } - @Test @Ignore //dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE) @@ -237,7 +221,7 @@ public class NewAggregatorEndpointTests { this.aggregator.handleMessage(message); } - @Test + @Test public void testAdditionalMessageAfterCompletion() throws InterruptedException { this.aggregator.start(); QueueChannel replyChannel = new QueueChannel(); @@ -251,8 +235,6 @@ 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); - //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)); @@ -360,4 +342,4 @@ public class NewAggregatorEndpointTests { //noop } } -} +} \ No newline at end of file