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

INT-603: add configurable comparator to Resequencer

Browse Source
This commit is contained in:
David Syer
2010-05-06 10:49:24 +00:00
parent 5bad12c86c
commit 7f72f94cf6
6 changed files with 129 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java
View File

@@ -1,17 +1,14 @@
/*
* 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.
*
* 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;
@@ -37,10 +34,19 @@ import org.springframework.integration.store.MessageGroup;
*/
public class Resequencer implements ReleaseStrategy, MessageGroupProcessor {
private volatile SequenceNumberComparator sequenceNumberComparator = new SequenceNumberComparator();
private volatile Comparator<Message<?>> comparator = new SequenceNumberComparator();
private volatile boolean releasePartialSequences;
/**
* A comparator to use to order messages before processing. The default is to order by sequence number.
*
* @param comparator the comparator to use to order messages
*/
public void setComparator(Comparator<Message<?>> comparator) {
this.comparator = comparator;
}
/**
* Flag that determines if partial sequences are allowed. If true then as soon as enough messages arrive that can be
* ordered they will be released, provided they all have sequence numbers greater than those already released.
@@ -54,7 +60,7 @@ public class Resequencer implements ReleaseStrategy, MessageGroupProcessor {
public boolean canRelease(MessageGroup messages) {
if (releasePartialSequences) {
List<Message<?>> sorted = new ArrayList<Message<?>>(messages.getUnmarked());
Collections.sort(sorted, sequenceNumberComparator);
Collections.sort(sorted, comparator);
int head = sorted.get(sorted.size() - 1).getHeaders().getSequenceNumber();
int tail = sorted.get(0).getHeaders().getSequenceNumber() - 1;
return tail == messages.getMarked().size() && head - tail == sorted.size();
@@ -66,7 +72,7 @@ public class Resequencer implements ReleaseStrategy, MessageGroupProcessor {
Collection<Message<?>> messages = group.getUnmarked();
if (messages.size() > 0) {
List<Message<?>> sorted = new ArrayList<Message<?>>(messages);
Collections.sort(sorted, sequenceNumberComparator);
Collections.sort(sorted, comparator);
for (Message<?> message : sorted) {
channelTemplate.send(message, outputChannel);
}

1
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java
View File

@@ -37,6 +37,7 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.Resequencer");
IntegrationNamespaceUtils.setValueIfAttributeDefined(processorBuilder, element, "release-partial-sequences");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(processorBuilder, element, "comparator");
String processorRef = BeanDefinitionReaderUtils.registerWithGeneratedName(processorBuilder
.getBeanDefinition(), parserContext.getRegistry());

9
org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -1792,6 +1792,15 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="comparator" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.Comparator" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="release-partial-sequences" type="xsd:string" />
<xsd:attribute name="send-partial-result-on-timeout" type="xsd:string" />
</xsd:extension>

28
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
View File

@@ -23,6 +23,7 @@ import static org.junit.Assert.assertNull;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import org.junit.Before;
import org.junit.Test;
@@ -72,6 +73,33 @@ public class ResequencerTests {
assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber());
}
@Test
public void testBasicResequencingWithCustomComparator() throws InterruptedException {
this.processor.setReleasePartialSequences(false);
this.processor.setComparator(new Comparator<Message<?>>() {
@SuppressWarnings("unchecked")
public int compare(Message<?> o1, Message<?> o2) {
return ((Comparable)o1.getPayload()).compareTo(o2.getPayload());
}
});
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("789", "ABC", 3, 1, replyChannel);
Message<?> message2 = createMessage("123", "ABC", 3, 2, replyChannel);
Message<?> message3 = createMessage("456", "ABC", 3, 3, 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(2), reply1.getHeaders().getSequenceNumber());
assertNotNull(reply2);
assertEquals(new Integer(3), reply2.getHeaders().getSequenceNumber());
assertNotNull(reply3);
assertEquals(new Integer(1), reply3.getHeaders().getSequenceNumber());
}
@Test
public void testResequencingWithDuplicateMessages() {
this.processor.setReleasePartialSequences(false);

105
org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

@@ -1,17 +1,14 @@
/*
* 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.
*
* 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.config;
@@ -23,6 +20,7 @@ import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.util.TestUtils.getPropertyValue;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import org.junit.Before;
@@ -32,6 +30,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.aggregator.CorrelationStrategyAdapter;
import org.springframework.integration.aggregator.Resequencer;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
@@ -43,18 +42,17 @@ import org.springframework.integration.test.util.TestUtils;
/**
* @author Marius Bogoevici
* @author Mark Fisher
* @author Dave Syer
*/
public class ResequencerParserTests {
private ApplicationContext context;
@Before
public void setUp() {
this.context = new ClassPathXmlApplicationContext("resequencerParserTests.xml", this.getClass());
}
@Test
public void testResequencing() {
MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel");
@@ -80,15 +78,17 @@ public class ResequencerParserTests {
@Test
public void testDefaultResequencerProperties() {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("defaultResequencer");
CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler", CorrelatingMessageHandler.class);
CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
CorrelatingMessageHandler.class);
assertNull(getPropertyValue(resequencer, "outputChannel"));
assertTrue(getPropertyValue(resequencer, "discardChannel") instanceof NullChannel);
assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value",
1000l, getPropertyValue(resequencer, "channelTemplate.sendTimeout"));
assertEquals("The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", 1000l, getPropertyValue(
resequencer, "channelTemplate.sendTimeout"));
assertEquals(
"The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
false, getPropertyValue(resequencer, "sendPartialResultOnTimeout"));
assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag",
false, getPropertyValue(getPropertyValue(resequencer, "outputProcessor"), "releasePartialSequences"));
false, getPropertyValue(getPropertyValue(resequencer, "outputProcessor"), "releasePartialSequences"));
}
@Test
@@ -96,31 +96,37 @@ public class ResequencerParserTests {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("completelyDefinedResequencer");
MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel");
CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler", CorrelatingMessageHandler.class);
assertEquals("The ResequencerEndpoint is not injected with the appropriate output channel",
outputChannel, getPropertyValue(resequencer, "outputChannel"));
assertEquals("The ResequencerEndpoint is not injected with the appropriate discard channel",
discardChannel, getPropertyValue(resequencer, "discardChannel"));
assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value",
86420000l, getPropertyValue(resequencer, "channelTemplate.sendTimeout"));
assertEquals("The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
CorrelatingMessageHandler.class);
assertEquals("The ResequencerEndpoint is not injected with the appropriate output channel", outputChannel,
getPropertyValue(resequencer, "outputChannel"));
assertEquals("The ResequencerEndpoint is not injected with the appropriate discard channel", discardChannel,
getPropertyValue(resequencer, "discardChannel"));
assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", 86420000l,
getPropertyValue(resequencer, "channelTemplate.sendTimeout"));
assertEquals(
"The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
true, getPropertyValue(resequencer, "sendPartialResultOnTimeout"));
assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag",
false, getPropertyValue(getPropertyValue(resequencer, "outputProcessor"), "releasePartialSequences"));
false, getPropertyValue(getPropertyValue(resequencer, "outputProcessor"), "releasePartialSequences"));
}
@Test
public void testCorrelationStrategyRefOnly() throws Exception {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithCorrelationStrategyRefOnly");
CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler", CorrelatingMessageHandler.class);
assertEquals("The ResequencerEndpoint is not configured with the appropriate CorrelationStrategy",
context.getBean("testCorrelationStrategy"), getPropertyValue(resequencer, "correlationStrategy"));
EventDrivenConsumer endpoint = (EventDrivenConsumer) context
.getBean("resequencerWithCorrelationStrategyRefOnly");
CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
CorrelatingMessageHandler.class);
assertEquals("The ResequencerEndpoint is not configured with the appropriate CorrelationStrategy", context
.getBean("testCorrelationStrategy"), getPropertyValue(resequencer, "correlationStrategy"));
}
@Test
public void testCorrelationStrategyRefAndMethod() throws Exception {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithCorrelationStrategyRefAndMethod");
CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler", CorrelatingMessageHandler.class);
EventDrivenConsumer endpoint = (EventDrivenConsumer) context
.getBean("resequencerWithCorrelationStrategyRefAndMethod");
CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
CorrelatingMessageHandler.class);
Object correlationStrategy = getPropertyValue(resequencer, "correlationStrategy");
assertEquals("The ResequencerEndpoint is not configured with a CorrelationStrategy adapter",
CorrelationStrategyAdapter.class, correlationStrategy.getClass());
@@ -128,17 +134,22 @@ public class ResequencerParserTests {
assertEquals("foo", adapter.getCorrelationKey(MessageBuilder.withPayload("not important").build()));
}
private static <T> Message<T> createMessage(T payload, Object correlationId,
int sequenceSize, int sequenceNumber, MessageChannel outputChannel) {
return MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId)
.setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(outputChannel)
.build();
@Test
public void testComparator() throws Exception {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithComparator");
CorrelatingMessageHandler handler = TestUtils.getPropertyValue(endpoint, "handler",
CorrelatingMessageHandler.class);
Resequencer resequencer = TestUtils.getPropertyValue(handler, "outputProcessor", Resequencer.class);
Object comparator = getPropertyValue(resequencer, "comparator");
assertEquals("The Resequencer is not configured with a TestComparator", TestComparator.class, comparator
.getClass());
}
private static <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel outputChannel) {
return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber).setReplyChannel(outputChannel).build();
}
static class TestCorrelationStrategy implements CorrelationStrategy {
@@ -147,7 +158,6 @@ public class ResequencerParserTests {
}
}
static class TestCorrelationStrategyPojo {
public Object foo(Object o) {
@@ -155,4 +165,11 @@ public class ResequencerParserTests {
}
}
static class TestComparator implements Comparator<Message<?>> {
@SuppressWarnings("unchecked")
public int compare(Message<?> o1, Message<?> o2) {
return ((Comparable) o1.getPayload()).compareTo(o2.getPayload());
}
}
}

9
org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml
View File

@@ -25,6 +25,8 @@
<channel id="inputChannel4"/>
<channel id="inputChannel5"/>
<resequencer id="completelyDefinedResequencer"
input-channel="inputChannel2"
output-channel="outputChannel"
@@ -42,6 +44,13 @@
correlation-strategy="testCorrelationStrategyPojo"
correlation-strategy-method="foo"/>
<resequencer id="resequencerWithComparator"
input-channel="inputChannel5"
comparator="testComparator"/>
<beans:bean id="testComparator"
class="org.springframework.integration.config.ResequencerParserTests$TestComparator"/>
<beans:bean id="testCorrelationStrategy"
class="org.springframework.integration.config.ResequencerParserTests$TestCorrelationStrategy"/>