diff --git a/spring-integration-core/src/main/java/org/springframework/integration/MessageDeliveryException.java b/spring-integration-core/src/main/java/org/springframework/integration/MessageDeliveryException.java index 2c9d56193e..8a7e8c5866 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/MessageDeliveryException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/MessageDeliveryException.java @@ -16,6 +16,8 @@ package org.springframework.integration; +import org.springframework.integration.message.Message; + /** * Exception that indicates an error during message delivery. * @@ -24,16 +26,36 @@ package org.springframework.integration; @SuppressWarnings("serial") public class MessageDeliveryException extends MessagingException { + private Message undeliveredMessage; + + public MessageDeliveryException() { super(); } - public MessageDeliveryException(String message) { - super(message); + public MessageDeliveryException(Message undeliveredMessage) { + this.undeliveredMessage = undeliveredMessage; } - public MessageDeliveryException(String message, Throwable cause) { - super(message, cause); + public MessageDeliveryException(String description) { + super(description); + } + + public MessageDeliveryException(Message undeliveredMessage, String description) { + super(description); + this.undeliveredMessage = undeliveredMessage; + } + + public MessageDeliveryException(String description, Throwable cause) { + super(description, cause); + } + + + /** + * Return the undelivered {@link Message} if available, may be null. + */ + public Message getUndeliveredMessage() { + return this.undeliveredMessage; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java new file mode 100644 index 0000000000..9a02c457df --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java @@ -0,0 +1,128 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import java.util.concurrent.ConcurrentHashMap; + +import org.springframework.integration.MessageHandlingException; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.util.Assert; + +/** + * A {@link MessageHandler} implementation that waits for a complete + * group of {@link Message Messages} to arrive and then delegates to an + * {@link Aggregator} to combine them into a single {@link Message}. + *

+ * Each {@link Message} that is received by this handler will be associated with + * a group based upon the 'correlationId' property of its + * header. If no such property is available, a {@link MessageHandlingException} + * will be thrown. + *

+ * The default strategy for determining whether a group is complete is based on + * the 'sequenceSize' property of the header. Alternatively, a + * custom implementation of the {@link RoutingBarrierCompletionStrategy} may be + * provided. + *

+ * The 'timeout' value determines how long to wait for the + * complete group after the arrival of the first {@link Message} of the group. + * The default value is 1 minute. If the timeout elapses prior to completion, + * the handler will throw a {@link MessageHandlingException} by default. To + * prevent the exception and aggregate the group even when incomplete, set the + * 'shouldFailOnTimeout' property to 'false'. + * + * @author Mark Fisher + */ +public class AggregatingMessageHandler implements MessageHandler { + + private long timeout = 60000; + + private boolean shouldFailOnTimeout = true; + + private Aggregator aggregator; + + private RoutingBarrierCompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); + + private ConcurrentHashMap barriers = new ConcurrentHashMap(); + + + /** + * Create a handler that delegates to the provided aggregator to combine a + * group of messages into a single message. + */ + public AggregatingMessageHandler(Aggregator aggregator) { + Assert.notNull(aggregator, "'aggregator' must not be null"); + this.aggregator = aggregator; + } + + + /** + * Strategy to determine whether the group of messages is complete. + */ + public void setCompletionStrategy(RoutingBarrierCompletionStrategy completionStrategy) { + Assert.notNull(completionStrategy, "'completionStrategy' must not be null"); + this.completionStrategy = completionStrategy; + } + + /** + * Maximum time to wait (in milliseconds) for the completion strategy to + * become true. + */ + public void setTimeout(long timeout) { + Assert.isTrue(timeout >= 0, "'timeout' must not be negative"); + this.timeout = timeout; + } + + /** + * Specify whether this handler should throw a {@link MessageHandlingException} + * when a message group does not reach completion within the allotted time. The + * default is 'true'. Setting this to 'false' will cause + * the {@link Aggregator} to be invoked even when the group is incomplete. + */ + public void setShouldFailOnTimeout(boolean setShouldFailOnTimeout) { + this.shouldFailOnTimeout = setShouldFailOnTimeout; + } + + public Message handle(Message message) { + Object correlationId = message.getHeader().getCorrelationId(); + if (correlationId == null) { + throw new MessageHandlingException(this.getClass().getSimpleName() + + " requires the 'correlationId' property"); + } + RoutingBarrier barrier = barriers.putIfAbsent(correlationId, new RoutingBarrier(this.completionStrategy)); + if (barrier == null) { + try { + barrier = barriers.get(correlationId); + barrier.addMessage(message); + if (!barrier.waitForCompletion(this.timeout) && this.shouldFailOnTimeout) { + throw new MessageHandlingException("aggregation did not complete " + + "within the allotted time limit of " + this.timeout + " milliseconds"); + } + Message result = aggregator.aggregate(barrier.getMessages()); + return result; + } + finally { + this.barriers.remove(correlationId); + } + } + else { + barriers.get(correlationId).addMessage(message); + return null; + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java b/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java new file mode 100644 index 0000000000..2aca549b2e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java @@ -0,0 +1,33 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import java.util.List; + +import org.springframework.integration.message.Message; + +/** + * Strategy interface for aggregating a list of {@link Message Messages} into a + * single {@link Message}. + * + * @author Mark Fisher + */ +public interface Aggregator { + + Message aggregate(List> messages); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/MessageSequenceComparator.java b/spring-integration-core/src/main/java/org/springframework/integration/router/MessageSequenceComparator.java new file mode 100644 index 0000000000..bd6f39e6f8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/MessageSequenceComparator.java @@ -0,0 +1,37 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import java.util.Comparator; + +import org.springframework.integration.message.Message; + +/** + * A {@link Comparator} implementation based on the 'sequenceNumber' + * property of a {@link Message Message's} header. + * + * @author Mark Fisher + */ +public class MessageSequenceComparator implements Comparator> { + + public int compare(Message message1, Message message2) { + int s1 = message1.getHeader().getSequenceNumber(); + int s2 = message2.getHeader().getSequenceNumber(); + return (s1 < s2) ? -1 : (s1 == s2) ? 0 : 1; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java new file mode 100644 index 0000000000..2b8f7ef9ac --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java @@ -0,0 +1,104 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; + +import org.springframework.integration.message.Message; +import org.springframework.util.Assert; + +/** + * A rendezvous point for {@link Message Messages} that delegates to a + * {@link RoutingBarrierCompletionStrategy} to determine when a + * complete message group is available. + * + * @author Mark Fisher + */ +public class RoutingBarrier { + + private List> messages = new CopyOnWriteArrayList>(); + + private final RoutingBarrierCompletionStrategy completionStrategy; + + private volatile boolean complete = false; + + private ReentrantLock lock = new ReentrantLock(); + + private Condition condition = lock.newCondition(); + + + public RoutingBarrier(RoutingBarrierCompletionStrategy completionStrategy) { + Assert.notNull(completionStrategy, "'completionStrategy' must not be null"); + this.completionStrategy = completionStrategy; + } + + + public void addMessage(Message message) { + this.messages.add(message); + if (this.completionStrategy.isComplete(this.messages)) { + try { + this.lock.lock(); + if (!this.complete) { + this.complete = true; + this.condition.signalAll(); + } + } + finally { + this.lock.unlock(); + } + } + } + + public boolean waitForCompletion(long timeout) { + if (this.complete) { + return true; + } + lock.lock(); + try { + if (this.complete) { + return true; + } + if (timeout >= 0) { + return this.condition.await(timeout, TimeUnit.MILLISECONDS); + } + else { + this.condition.await(); + return true; + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + finally { + lock.unlock(); + } + } + + public boolean isComplete() { + return this.complete; + } + + public List> getMessages() { + return this.messages; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrierCompletionStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrierCompletionStrategy.java new file mode 100644 index 0000000000..0c9d970f77 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrierCompletionStrategy.java @@ -0,0 +1,34 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import java.util.List; + +import org.springframework.integration.message.Message; + +/** + * Strategy for determining when a group of messages reaches a state of + * completion (i.e. can trip a barrier). + * + * @author Mark Fisher + * @see RoutingBarrier + */ +public interface RoutingBarrierCompletionStrategy { + + boolean isComplete(List> messages); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java new file mode 100644 index 0000000000..229c430e4a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java @@ -0,0 +1,40 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import java.util.List; + +import org.springframework.integration.message.Message; +import org.springframework.util.CollectionUtils; + +/** + * An implementation of {@link RoutingBarrierCompletionStrategy} that simply + * compares the current size of the message list to the expected 'sequenceSize' + * according to the first {@link Message} in the list. + * + * @author Mark Fisher + */ +public class SequenceSizeCompletionStrategy implements RoutingBarrierCompletionStrategy { + + public boolean isComplete(List> messages) { + if (CollectionUtils.isEmpty(messages)) { + return false; + } + return (messages.size() >= messages.get(0).getHeader().getSequenceSize()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java index c1a51dcba9..3387eeaefe 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java @@ -173,7 +173,7 @@ public class DefaultMessageEndpointTests { endpoint.start(); endpoint.handle(new StringMessage(1, "test")); endpoint.stop(); - latch.await(200, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); Message reply = replyChannel.receive(100); assertNotNull(reply); @@ -199,7 +199,7 @@ public class DefaultMessageEndpointTests { endpoint.start(); endpoint.handle(new StringMessage(1, "test")); endpoint.stop(); - latch.await(200, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); Message reply = replyChannel.receive(0); assertNull(reply); @@ -224,7 +224,7 @@ public class DefaultMessageEndpointTests { endpoint.start(); endpoint.handle(new StringMessage(1, "test")); endpoint.stop(); - latch.await(200, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); Message reply = replyChannel.receive(0); assertNull(reply); @@ -250,7 +250,7 @@ public class DefaultMessageEndpointTests { message.getHeader().setReplyChannelName("replyChannel"); endpoint.handle(message); endpoint.stop(); - latch.await(200, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); Message reply = replyChannel.receive(100); assertNotNull(reply); @@ -277,7 +277,7 @@ public class DefaultMessageEndpointTests { endpoint.start(); endpoint.handle(new StringMessage(1, "test")); endpoint.stop(); - latch.await(200, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); Message reply = replyChannel.receive(100); assertNotNull(reply); @@ -305,7 +305,7 @@ public class DefaultMessageEndpointTests { message.getHeader().setReplyChannelName("replyChannel"); endpoint.handle(message); endpoint.stop(); - latch.await(200, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); Message reply = replyChannel.receive(100); assertNotNull(reply); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java new file mode 100644 index 0000000000..5920b1fab6 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java @@ -0,0 +1,197 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; + +import org.springframework.integration.MessageHandlingException; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +/** + * @author Mark Fisher + */ +public class AggregatingMessageHandlerTests { + + private final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + + + public AggregatingMessageHandlerTests() { + this.executor.setMaxPoolSize(10); + this.executor.setQueueCapacity(0); + this.executor.afterPropertiesSet(); + } + + + @Test + public void testCompleteGroupWithinTimeout() throws InterruptedException { + AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); + SimpleChannel replyChannel = new SimpleChannel(); + Message message1 = createMessage("123", "ABC", 3, 1, replyChannel); + Message message2 = createMessage("456", "ABC", 3, 2, replyChannel); + Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); + CountDownLatch latch = new CountDownLatch(3); + executor.execute(new AggregatorTestTask(aggregator, message1, latch)); + executor.execute(new AggregatorTestTask(aggregator, message2, latch)); + executor.execute(new AggregatorTestTask(aggregator, message3, latch)); + latch.await(1000, TimeUnit.MILLISECONDS); + Message reply = replyChannel.receive(100); + assertNotNull(reply); + assertEquals("123456789", reply.getPayload()); + } + + @Test + public void testShouldFailOnTimeoutByDefault() throws InterruptedException { + AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); + aggregator.setTimeout(10); + SimpleChannel replyChannel = new SimpleChannel(); + Message message1 = createMessage("123", "ABC", 2, 1, replyChannel); + CountDownLatch latch = new CountDownLatch(1); + AggregatorTestTask task = new AggregatorTestTask(aggregator, message1, latch); + executor.execute(task); + latch.await(500, TimeUnit.MILLISECONDS); + Message reply = replyChannel.receive(0); + assertNull(reply); + assertNotNull(task.getException()); + assertEquals(MessageHandlingException.class, task.getException().getClass()); + } + + @Test + public void testShouldFailOnTimeoutFalse() throws InterruptedException { + AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); + aggregator.setTimeout(10); + aggregator.setShouldFailOnTimeout(false); + SimpleChannel replyChannel = new SimpleChannel(); + Message message1 = createMessage("123", "ABC", 3, 1, replyChannel); + Message message2 = createMessage("456", "ABC", 3, 2, replyChannel); + CountDownLatch latch = new CountDownLatch(2); + AggregatorTestTask task1 = new AggregatorTestTask(aggregator, message1, latch); + AggregatorTestTask task2 = new AggregatorTestTask(aggregator, message2, latch); + executor.execute(task1); + executor.execute(task2); + latch.await(500, TimeUnit.MILLISECONDS); + Message reply = replyChannel.receive(100); + assertNotNull(reply); + assertEquals("123456", reply.getPayload()); + assertNull(task1.getException()); + assertNull(task2.getException()); + } + + @Test + public void testMultipleGroupsSimultaneously() throws InterruptedException { + AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); + SimpleChannel replyChannel1 = new SimpleChannel(); + SimpleChannel replyChannel2 = new SimpleChannel(); + Message message1 = createMessage("123", "ABC", 3, 1, replyChannel1); + Message message2 = createMessage("456", "ABC", 3, 2, replyChannel1); + Message message3 = createMessage("789", "ABC", 3, 3, replyChannel1); + Message message4 = createMessage("abc", "XYZ", 3, 1, replyChannel2); + Message message5 = createMessage("def", "XYZ", 3, 2, replyChannel2); + Message message6 = createMessage("ghi", "XYZ", 3, 3, replyChannel2); + CountDownLatch latch = new CountDownLatch(6); + executor.execute(new AggregatorTestTask(aggregator, message1, latch)); + executor.execute(new AggregatorTestTask(aggregator, message6, latch)); + executor.execute(new AggregatorTestTask(aggregator, message2, latch)); + executor.execute(new AggregatorTestTask(aggregator, message5, latch)); + executor.execute(new AggregatorTestTask(aggregator, message3, latch)); + executor.execute(new AggregatorTestTask(aggregator, message4, latch)); + latch.await(1000, TimeUnit.MILLISECONDS); + Message reply1 = replyChannel1.receive(500); + assertNotNull(reply1); + assertEquals("123456789", reply1.getPayload()); + Message reply2 = replyChannel2.receive(500); + assertNotNull(reply2); + assertEquals("abcdefghi", reply2.getPayload()); + } + + + private static Message createMessage(String payload, Object correlationId, + int sequenceSize, int sequenceNumber, MessageChannel replyChannel) { + StringMessage message = new StringMessage(payload); + message.getHeader().setCorrelationId(correlationId); + message.getHeader().setSequenceSize(sequenceSize); + message.getHeader().setSequenceNumber(sequenceNumber); + message.getHeader().setReplyChannel(replyChannel); + return message; + } + + + private static class TestAggregator implements Aggregator { + + public Message aggregate(List> messages) { + List> sortableList = new ArrayList>(messages); + Collections.sort(sortableList, new MessageSequenceComparator()); + StringBuffer buffer = new StringBuffer(); + for (Message message : sortableList) { + buffer.append(message.getPayload().toString()); + } + return new StringMessage(buffer.toString()); + } + } + + + private static class AggregatorTestTask implements Runnable { + + private AggregatingMessageHandler aggregator; + + private Message message; + + private Exception exception; + + private CountDownLatch latch; + + + AggregatorTestTask(AggregatingMessageHandler aggregator, Message message, CountDownLatch latch) { + this.aggregator = aggregator; + this.message = message; + this.latch = latch; + } + + public Exception getException() { + return this.exception; + } + + public void run() { + try { + Message result = this.aggregator.handle(message); + if (result != null) { + message.getHeader().getReplyChannel().send(result); + } + } + catch (Exception e) { + this.exception = e; + } + finally { + this.latch.countDown(); + } + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/MessageSequenceComparatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/MessageSequenceComparatorTests.java new file mode 100644 index 0000000000..1c321cf41b --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/MessageSequenceComparatorTests.java @@ -0,0 +1,68 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; + +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class MessageSequenceComparatorTests { + + @Test + public void testLessThan() { + MessageSequenceComparator comparator = new MessageSequenceComparator(); + StringMessage message1 = new StringMessage("test1"); + message1.getHeader().setSequenceNumber(1); + StringMessage message2 = new StringMessage("test2"); + message2.getHeader().setSequenceNumber(2); + assertEquals(-1, comparator.compare(message1, message2)); + } + + @Test + public void testEqual() { + MessageSequenceComparator comparator = new MessageSequenceComparator(); + StringMessage message1 = new StringMessage("test1"); + message1.getHeader().setSequenceNumber(3); + StringMessage message2 = new StringMessage("test2"); + message2.getHeader().setSequenceNumber(3); + assertEquals(0, comparator.compare(message1, message2)); + } + + @Test + public void testGreaterThan() { + MessageSequenceComparator comparator = new MessageSequenceComparator(); + StringMessage message1 = new StringMessage("test1"); + message1.getHeader().setSequenceNumber(5); + StringMessage message2 = new StringMessage("test2"); + message2.getHeader().setSequenceNumber(3); + assertEquals(1, comparator.compare(message1, message2)); + } + + @Test + public void testEqualWithDefaultValues() { + MessageSequenceComparator comparator = new MessageSequenceComparator(); + StringMessage message1 = new StringMessage("test1"); + StringMessage message2 = new StringMessage("test2"); + assertEquals(0, comparator.compare(message1, message2)); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/RoutingBarrierTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/RoutingBarrierTests.java new file mode 100644 index 0000000000..0cb367f863 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/RoutingBarrierTests.java @@ -0,0 +1,87 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.List; + +import org.junit.Test; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class RoutingBarrierTests { + + @Test + public void testBasicCompletionCheck() { + RoutingBarrier barrier = new RoutingBarrier(new TwoMessageCompletionStrategy()); + barrier.addMessage(new StringMessage("test1")); + assertFalse(barrier.isComplete()); + barrier.addMessage(new StringMessage("test2")); + assertTrue(barrier.isComplete()); + } + + @Test + public void testMessageRetrieval() { + RoutingBarrier barrier = new RoutingBarrier(new TwoMessageCompletionStrategy()); + barrier.addMessage(new StringMessage("test1")); + assertEquals(1, barrier.getMessages().size()); + barrier.addMessage(new StringMessage("test2")); + assertEquals(2, barrier.getMessages().size()); + } + + @Test + public void testWaitForCompletionTimesOut() { + RoutingBarrier barrier = new RoutingBarrier(new TwoMessageCompletionStrategy()); + barrier.addMessage(new StringMessage("test1")); + assertFalse(barrier.isComplete()); + assertFalse(barrier.waitForCompletion(10)); + } + + @Test + public void testWaitForCompletionReturnsTrueImmediately() { + RoutingBarrier barrier = new RoutingBarrier(new TwoMessageCompletionStrategy()); + barrier.addMessage(new StringMessage("test1")); + assertFalse(barrier.isComplete()); + barrier.addMessage(new StringMessage("test2")); + assertTrue(barrier.waitForCompletion(0)); + assertTrue(barrier.isComplete()); + } + + @Test + public void testEmptyMessageList() { + RoutingBarrier barrier = new RoutingBarrier(new TwoMessageCompletionStrategy()); + assertFalse(barrier.isComplete()); + assertFalse(barrier.waitForCompletion(0)); + assertEquals(0, barrier.getMessages().size()); + } + + + private static class TwoMessageCompletionStrategy implements RoutingBarrierCompletionStrategy { + + public boolean isComplete(List> messages) { + return (messages.size() == 2); + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/SequenceSizeCompletionStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/SequenceSizeCompletionStrategyTests.java new file mode 100644 index 0000000000..d41755248f --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/SequenceSizeCompletionStrategyTests.java @@ -0,0 +1,70 @@ +/* + * Copyright 2002-2007 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. + */ + +package org.springframework.integration.router; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Test; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class SequenceSizeCompletionStrategyTests { + + @Test + public void testIncompleteList() { + Message message = new StringMessage("test1"); + message.getHeader().setSequenceSize(2); + List> messages = new ArrayList>(); + messages.add(message); + SequenceSizeCompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); + assertFalse(completionStrategy.isComplete(messages)); + } + + @Test + public void testCompleteList() { + Message message1 = new StringMessage("test1"); + message1.getHeader().setSequenceSize(2); + Message message2 = new StringMessage("test2"); + message2.getHeader().setSequenceSize(2); + List> messages = new ArrayList>(); + messages.add(message1); + messages.add(message2); + SequenceSizeCompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); + assertTrue(completionStrategy.isComplete(messages)); + } + + @Test + public void testEmptyList() { + SequenceSizeCompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); + assertFalse(completionStrategy.isComplete(new ArrayList>())); + } + + @Test + public void testNullList() { + SequenceSizeCompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); + assertFalse(completionStrategy.isComplete(null)); + } + +}