From c377b82d8a1a2259ff7d7d03cfe1661bff9fae10 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 6 Jun 2012 10:08:05 -0400 Subject: [PATCH] INT-2518 addressed Resequencer with custom comparator removed setComparator(..) method from ResequencingMessageGroupProcessor, fixed tests. More details as to why are available in JIRA https://jira.springsource.org/browse/INT-2518 --- .../ResequencingMessageGroupProcessor.java | 29 ++++---- .../aggregator/ResequencerTests.java | 70 +++++-------------- .../config/ResequencerParserTests.java | 41 +++-------- .../config/resequencerParserTests.xml | 17 ----- 4 files changed, 40 insertions(+), 117 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java index 8161a4daf1..e58ab8c690 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2011 the original author or authors. - * + * Copyright 2002-2012 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. @@ -13,11 +13,15 @@ package org.springframework.integration.aggregator; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; + import org.springframework.integration.Message; import org.springframework.integration.store.MessageGroup; -import java.util.*; - /** * This class implements all the strategy interfaces needed for a default resequencer. * @@ -28,17 +32,8 @@ import java.util.*; */ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor { - private volatile Comparator> comparator = new SequenceNumberComparator(); + private final Comparator> comparator = new SequenceNumberComparator(); - /** - * 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; - } - public Object processMessageGroup(MessageGroup group) { Collection> messages = group.getMessages(); @@ -57,7 +52,7 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor } partialSequence.add(message); } - + return partialSequence; } return null; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index c35de895f5..3f8bac57ca 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -16,13 +16,19 @@ package org.springframework.integration.aggregator; +import static org.hamcrest.Matchers.is; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import java.util.Comparator; import org.junit.Before; import org.junit.Test; + import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.channel.QueueChannel; @@ -30,13 +36,6 @@ import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; -import static org.hamcrest.Matchers.is; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertThat; - /** * @author Marius Bogoevici * @author Alex Peters @@ -48,9 +47,9 @@ public class ResequencerTests { private ResequencingMessageHandler resequencer; - private ResequencingMessageGroupProcessor processor = new ResequencingMessageGroupProcessor(); + private final ResequencingMessageGroupProcessor processor = new ResequencingMessageGroupProcessor(); - private MessageGroupStore store = new SimpleMessageStore(); + private final MessageGroupStore store = new SimpleMessageStore(); @Before public void configureResequencer() { @@ -76,31 +75,31 @@ public class ResequencerTests { assertNotNull(reply3); assertThat( reply3.getHeaders().getSequenceNumber(), is(3)); } - + @Test public void testBasicResequencingA() throws InterruptedException { SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); releaseStrategy.setReleasePartialSequences(true); this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy); - + QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 3, 1, replyChannel); Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); - + this.resequencer.handleMessage(message3); assertNull(replyChannel.receive(0)); this.resequencer.handleMessage(message1); assertNotNull(replyChannel.receive(0)); assertNull(replyChannel.receive(0)); } - + @Test public void testBasicUnboundedResequencing() throws InterruptedException { SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); releaseStrategy.setReleasePartialSequences(true); this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy); QueueChannel replyChannel = new QueueChannel(); - this.resequencer.setCorrelationStrategy(new CorrelationStrategy() { + this.resequencer.setCorrelationStrategy(new CorrelationStrategy() { public Object getCorrelationKey(Message message) { return "A"; } @@ -111,51 +110,24 @@ public class ResequencerTests { Message message3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setReplyChannel(replyChannel).build(); Message message4 = MessageBuilder.withPayload("4").setSequenceNumber(4).setReplyChannel(replyChannel).build(); Message message5 = MessageBuilder.withPayload("5").setSequenceNumber(5).setReplyChannel(replyChannel).build(); - + this.resequencer.handleMessage(message3); assertNull(replyChannel.receive(0)); this.resequencer.handleMessage(message1); assertNotNull(replyChannel.receive(0)); - + this.resequencer.handleMessage(message2); assertNotNull(replyChannel.receive(0)); assertNotNull(replyChannel.receive(0)); assertNull(replyChannel.receive(0)); - + this.resequencer.handleMessage(message5); assertNull(replyChannel.receive(0)); this.resequencer.handleMessage(message4); assertNotNull(replyChannel.receive(0)); } - - @Test - public void testBasicResequencingWithCustomComparator() throws InterruptedException { - this.processor.setComparator(new Comparator>() { - @SuppressWarnings({ "unchecked", "rawtypes" }) - 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() { QueueChannel replyChannel = new QueueChannel(); @@ -210,12 +182,6 @@ public class ResequencerTests { @Test public void testResequencingWithPartialSequenceAndComparator() throws InterruptedException { this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true)); - this.processor.setComparator(new Comparator>() { - @SuppressWarnings({ "unchecked", "rawtypes" }) - public int compare(Message o1, Message o2) { - return ((Comparable)o1.getPayload()).compareTo(o2.getPayload()); - } - }); QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("456", "ABC", 4, 2, replyChannel); Message message2 = createMessage("123", "ABC", 4, 1, replyChannel); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index cfa9a3dd09..f30d0069a7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2011 the original author or authors. - * + * Copyright 2002-2012 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. @@ -13,33 +13,31 @@ package org.springframework.integration.config; -import java.util.Comparator; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.springframework.integration.test.util.TestUtils.getPropertyValue; import org.junit.Before; import org.junit.Test; + import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.aggregator.CorrelationStrategy; import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy; -import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor; import org.springframework.integration.aggregator.ResequencingMessageHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; - -import static org.springframework.integration.test.util.TestUtils.getPropertyValue; - /** * @author Marius Bogoevici * @author Mark Fisher * @author Dave Syer + * @author Oleg Zhurakousky */ public class ResequencerParserTests { @@ -118,18 +116,6 @@ public class ResequencerParserTests { assertEquals("foo", adapter.getCorrelationKey(MessageBuilder.withPayload("not important").build())); } - @Test - public void testComparator() throws Exception { - EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithComparator"); - ResequencingMessageHandler handler = TestUtils.getPropertyValue(endpoint, "handler", - ResequencingMessageHandler.class); - ResequencingMessageGroupProcessor resequencer = TestUtils.getPropertyValue(handler, "outputProcessor", - ResequencingMessageGroupProcessor.class); - Object comparator = getPropertyValue(resequencer, "comparator"); - assertEquals("The Resequencer is not configured with a TestComparator", TestComparator.class, comparator - .getClass()); - } - @SuppressWarnings("unused") private static Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, MessageChannel outputChannel) { @@ -151,11 +137,4 @@ public class ResequencerParserTests { } } - static class TestComparator implements Comparator> { - @SuppressWarnings({ "unchecked", "rawtypes" }) - public int compare(Message o1, Message o2) { - return ((Comparable) o1.getPayload()).compareTo(o2.getPayload()); - } - } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml index ebb1127898..8881c65ed0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml @@ -46,27 +46,10 @@ correlation-strategy="testCorrelationStrategyPojo" correlation-strategy-method="foo"/> - - - - - - - - - - - - - -