Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-330: added a check for superseded messages (duplicating sequence number)

Browse Source
This commit is contained in:
Iwein Fuld
2009-09-30 21:11:10 +00:00
parent e7e18fbbb0
commit 7a0b2587bf
6 changed files with 320 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java
View File

@@ -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<Message<?>> all = store.getAll(correlationKey);
complete(correlationKey, all, this.resolveReplyChannel(message, this.outputChannel, this.channelResolver));
List<Message<?>> 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<Message<?>> group) {
for (Message<?> member : group) {
if (member.getHeaders().getSequenceNumber() == message.getHeaders().getSequenceNumber()) {
return false;
}
}
return true;
}
private boolean complete(Object correlationKey, List<Message<?>> correlatedMessages, MessageChannel messageChannel) {
boolean processed = false;
if (completionStrategy.isComplete(correlatedMessages)) {

8
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java
View File

@@ -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

4
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java
View File

@@ -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);

15
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java
View File

@@ -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<Message<?>> storedMessages = new ArrayList<Message<?>>();
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();
}
}

287
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java Normal file
View File

@@ -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<Integer> reply1 = (Message<Integer>) replyChannel1.receive(500);
assertNotNull(reply1);
assertThat(reply1.getPayload(), is(105));
Message<Integer> reply2 = (Message<Integer>) 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<Object> 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<Message<?>> 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<Message<?>> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
//noop
}
}
}

26
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java → org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java
View File

@@ -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
}
}
}
}