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

Merge pull request #477 from olegz/INT-2518

Browse Source
This commit is contained in:
Gary Russell
2012-06-06 18:53:40 -04:00
parent 49960f3f36 c377b82d8a
commit 956e7bcd0d
4 changed files with 40 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java
View File

@@ -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<Message<?>> comparator = new SequenceNumberComparator();
private final Comparator<Message<?>> 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<Message<?>> comparator) {
this.comparator = comparator;
}
public Object processMessageGroup(MessageGroup group) {
Collection<Message<?>> messages = group.getMessages();
@@ -57,7 +52,7 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
}
partialSequence.add(message);
}
return partialSequence;
}
return null;

70
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
View File

@@ -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<Message<?>>() {
@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<Message<?>>() {
@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);

41
spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

@@ -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 <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel outputChannel) {
@@ -151,11 +137,4 @@ public class ResequencerParserTests {
}
}
static class TestComparator implements Comparator<Message<?>> {
@SuppressWarnings({ "unchecked", "rawtypes" })
public int compare(Message<?> o1, Message<?> o2) {
return ((Comparable) o1.getPayload()).compareTo(o2.getPayload());
}
}
}

17
spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml
View File

@@ -46,27 +46,10 @@
correlation-strategy="testCorrelationStrategyPojo"
correlation-strategy-method="foo"/>
<resequencer id="resequencerWithComparator"
input-channel="inputChannel5"
comparator="testComparator"/>
<!-- <resequencer id="resequencerWithReleaseStrategy" -->
<!-- input-channel="inputChannel6" -->
<!-- release-strategy="pojoReleaseStrategy" -->
<!-- release-strategy-method="checkCompletenessAsList"/> -->
<beans:bean id="testComparator"
class="org.springframework.integration.config.ResequencerParserTests$TestComparator"/>
<beans:bean id="testCorrelationStrategy"
class="org.springframework.integration.config.ResequencerParserTests$TestCorrelationStrategy"/>
<beans:bean id="testCorrelationStrategyPojo"
class="org.springframework.integration.config.ResequencerParserTests$TestCorrelationStrategyPojo"/>
<!-- <beans:bean id="pojoReleaseStrategy" -->
<!-- class="org.springframework.integration.config.MaxValueReleaseStrategy"> -->
<!-- <beans:constructor-arg value="10" /> -->
<!-- </beans:bean> -->
</beans:beans>