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

INT-1105: special case for no sequence size (the throttling use case)

Browse Source
This commit is contained in:
David Syer
2010-05-02 06:11:53 +00:00
parent 437fa54ba8
commit ee0b597e3b
4 changed files with 40 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -173,7 +173,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
}
outputProcessor.processAndSend(group, channelTemplate, this.resolveReplyChannel(message,
this.outputChannel));
if (group.isComplete()) {
if (group.isComplete() || group.getSequenceSize()==0) {
complete(group);
}
else {

4
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java
View File

@@ -92,8 +92,8 @@ public class MessageGroup {
return true;
}
int sequenceSize = getSequenceSize();
// If there is no sequence then it must be complete....
return sequenceSize == 0 || sequenceSize == size();
// If there is no sequence then it must be incomplete....
return sequenceSize > 0 && sequenceSize == size();
}
public int getSequenceSize() {

20
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java
View File

@@ -16,6 +16,10 @@
package org.springframework.integration.aggregator;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.core.Message;
@@ -24,8 +28,6 @@ import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
import static org.mockito.Mockito.*;
public class CorrelatingMessageHandlerIntegrationTest {
private MessageStore store = new SimpleMessageStore(100);
@@ -64,6 +66,20 @@ public class CorrelatingMessageHandlerIntegrationTest {
verify(discardChannel).send(message2);
}
@Test
public void completesIfNoSequence() throws Exception {
defaultHandler.setReleaseStrategy(new MessageCountReleaseStrategy(2));
Message<?> message1 = MessageBuilder.withPayload(1).setCorrelationId("foo").build();
Message<?> message2 = MessageBuilder.withPayload(2).setCorrelationId("foo").build();
Message<?> message3 = MessageBuilder.withPayload(3).setCorrelationId("foo").build();
defaultHandler.handleMessage(message1);
verify(outputChannel, never()).send(message3);
defaultHandler.handleMessage(message2);
verify(outputChannel).send(message2);
defaultHandler.handleMessage(message3);
verify(outputChannel, never()).send(message3);
}
@Test
public void completesWithoutReleasingIncompleteCorrelations() throws Exception {
Message<?> message1 = correlatedMessage(1, 2, 1);

57
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java
View File

@@ -63,24 +63,15 @@ public class AggregatorWithCorrelationStrategyTests {
@Test
public void testCorrelationAndCompletion() {
inputChannel.send(MessageBuilder.withPayload("A1").setSequenceNumber(0).setSequenceSize(3)
.build());
inputChannel.send(MessageBuilder.withPayload("B2").setSequenceNumber(0).setSequenceSize(3)
.build());
inputChannel.send(MessageBuilder.withPayload("C3").setSequenceNumber(0).setSequenceSize(3)
.build());
inputChannel.send(MessageBuilder.withPayload("A4").setSequenceNumber(1).setSequenceSize(3)
.build());
inputChannel.send(MessageBuilder.withPayload("B5").setSequenceNumber(1).setSequenceSize(3)
.build());
inputChannel.send(MessageBuilder.withPayload("C6").setSequenceNumber(1).setSequenceSize(3)
.build());
inputChannel.send(MessageBuilder.withPayload("A7").setSequenceNumber(2).setSequenceSize(3)
.build());
inputChannel.send(MessageBuilder.withPayload("B8").setSequenceNumber(2).setSequenceSize(3)
.build());
inputChannel.send(MessageBuilder.withPayload("C9").setSequenceNumber(2).setSequenceSize(3)
.build());
inputChannel.send(MessageBuilder.withPayload("A1").build());
inputChannel.send(MessageBuilder.withPayload("B2").build());
inputChannel.send(MessageBuilder.withPayload("C3").build());
inputChannel.send(MessageBuilder.withPayload("A4").build());
inputChannel.send(MessageBuilder.withPayload("B5").build());
inputChannel.send(MessageBuilder.withPayload("C6").build());
inputChannel.send(MessageBuilder.withPayload("A7").build());
inputChannel.send(MessageBuilder.withPayload("B8").build());
inputChannel.send(MessageBuilder.withPayload("C9").build());
receiveAndCompare(outputChannel, "A1", "A4", "A7");
receiveAndCompare(outputChannel, "B2", "B5", "B8");
receiveAndCompare(outputChannel, "C3", "C6", "C9");
@@ -91,34 +82,25 @@ public class AggregatorWithCorrelationStrategyTests {
// the test verifies how a pojo strategy is applied
// Strings are correlated by their first letter, integers are correlated
// by the last digit
pojoInputChannel.send(MessageBuilder.withPayload("X1")
.setSequenceNumber(0).setSequenceSize(3).build());
pojoInputChannel.send(MessageBuilder.withPayload(93).setSequenceNumber(
0).setSequenceSize(3).build());
pojoInputChannel.send(MessageBuilder.withPayload("X4")
.setSequenceNumber(1).setSequenceSize(3).build());
pojoInputChannel.send(MessageBuilder.withPayload(113)
.setSequenceNumber(1).setSequenceSize(3).build());
pojoInputChannel.send(MessageBuilder.withPayload("X7")
.setSequenceNumber(2).setSequenceSize(3).build());
pojoInputChannel.send(MessageBuilder.withPayload(213)
.setSequenceNumber(2).setSequenceSize(3).build());
pojoInputChannel.send(MessageBuilder.withPayload("X1").build());
pojoInputChannel.send(MessageBuilder.withPayload(93).build());
pojoInputChannel.send(MessageBuilder.withPayload("X4").build());
pojoInputChannel.send(MessageBuilder.withPayload(113).build());
pojoInputChannel.send(MessageBuilder.withPayload("X7").build());
pojoInputChannel.send(MessageBuilder.withPayload(213).build());
receiveAndCompare(pojoOutputChannel, "X1", "X4", "X7");
receiveAndCompare(pojoOutputChannel, "93", "113", "213");
}
private void receiveAndCompare(PollableChannel outputChannel,
String... expectedValues) {
private void receiveAndCompare(PollableChannel outputChannel, String... expectedValues) {
Message<?> message = outputChannel.receive(500);
Assert.assertNotNull(message);
for (String expectedValue : expectedValues) {
assertThat((String) message.getPayload(),
containsString(expectedValue));
assertThat((String) message.getPayload(), containsString(expectedValue));
}
}
public static class MessageCountReleaseStrategy implements
ReleaseStrategy {
public static class MessageCountReleaseStrategy implements ReleaseStrategy {
private final int expectedSize;
@@ -132,8 +114,7 @@ public class AggregatorWithCorrelationStrategyTests {
}
public static class FirstLetterCorrelationStrategy implements
CorrelationStrategy {
public static class FirstLetterCorrelationStrategy implements CorrelationStrategy {
public Object getCorrelationKey(Message<?> message) {
return message.getPayload().toString().subSequence(0, 1);