diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java index 00c9106f7a..055fe6f1a2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java @@ -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> 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> 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> sorted = new ArrayList>(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> messages = group.getUnmarked(); if (messages.size() > 0) { List> sorted = new ArrayList>(messages); - Collections.sort(sorted, sequenceNumberComparator); + Collections.sort(sorted, comparator); for (Message message : sorted) { channelTemplate.send(message, outputChannel); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java index c9a1ac5fd8..167192e8a4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java @@ -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()); diff --git a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index b485099ce6..c47d7121e4 100644 --- a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -1792,6 +1792,15 @@ + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index 06d2a33e99..1998bfeca2 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -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>() { + @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); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index 65091f0f8f..f719e9e808 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -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 Message 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 Message 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> { + @SuppressWarnings("unchecked") + public int compare(Message o1, Message o2) { + return ((Comparable) o1.getPayload()).compareTo(o2.getPayload()); + } + } + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml index bf839f9574..a43fb9c399 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml @@ -25,6 +25,8 @@ + + + + + +