From fc96b55cf3693b81a30a8b9b2dc61ca280fb965a Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Sun, 25 Apr 2010 17:56:06 +0000 Subject: [PATCH] INT-1093: - Added testcases from the issue - added convenience method for total completion to MessageGroup - resolved issue by invoking callback --- ...tractAggregatingMessageGroupProcessor.java | 10 +- .../integration/aggregator/MessageGroup.java | 17 ++-- .../scenarios/AggregationResendTest.java | 94 +++++++++++++++++++ .../scenarios/aggregation-resend-config.xml | 34 +++++++ 4 files changed, 142 insertions(+), 13 deletions(-) create mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/AggregationResendTest.java create mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/aggregation-resend-config.xml diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index 7a1dd17d4d..5c78678672 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -16,15 +16,8 @@ package org.springframework.integration.aggregator; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -32,6 +25,8 @@ import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.MessageBuilder; import org.springframework.util.Assert; +import java.util.*; + /** * Base class for MessageGroupProcessor implementations that aggregate the * group of Messages into a single Message. @@ -53,6 +48,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag Map headers = this.aggregateHeaders(group); Message message = MessageBuilder.withPayload(payload).copyHeadersIfAbsent(headers).build(); channelTemplate.send(message, outputChannel); + group.onCompleteProcessing(); } /** diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java index 68bd0e2493..d7ef4736db 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java @@ -16,14 +16,10 @@ package org.springframework.integration.aggregator; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.List; - import org.springframework.integration.core.Message; +import java.util.*; + /** * Represents a mutable group of correlated messages that is bound to a certain * {@link org.springframework.integration.store.MessageStore} and correlation @@ -128,4 +124,13 @@ public class MessageGroup { } } + /** + * This method is a shorthand for signaling that all messages in the group have been + * processed and that the group is completed. + */ + public void onCompleteProcessing() { + onProcessingOf(messages.toArray(new Message[messages.size()])); + onCompletion(); + } + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/AggregationResendTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/AggregationResendTest.java new file mode 100644 index 0000000000..8d8f8691eb --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/AggregationResendTest.java @@ -0,0 +1,94 @@ +/* + * 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. + */ + +package org.springframework.integration.aggregator.scenarios; + +import junit.framework.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import java.util.ArrayList; +import java.util.List; + +/** + * Tests courtesy of Sean Crotty (INT-1093) + * + * @author Iwein Fuld + */ +@ContextConfiguration(locations = {"aggregation-resend-config.xml"}) +@RunWith(SpringJUnit4ClassRunner.class) +public class AggregationResendTest { + + @Autowired + DirectChannel input_for_aggregator_with_explicit_timeout; + + @Autowired + DirectChannel input_for_aggregator_without_explicit_timeout; + + @Autowired + PollableChannel reply; + + @Test + /** + * We expect to get back only one Message from the aggregator. We set no + * explicit timeout value on the aggregator. But apparently is automatically + * times out after 60 seconds. So What we'll see is that we get one aggregate + * Message back immediately. Then we'll get another 3 after the 60 seconds. + */ + public void testAggregatorWithoutExplicitTimeoutReturnsOnlyOneMessage() throws Exception { + sendMessage(input_for_aggregator_with_explicit_timeout, 2000); + } + + @Test + /** + * We expect to get back only one Message from the aggregator. We set an + * explicit timeout value of 1 second on the aggregator. What we'll see is + * that we get one aggregate Message back immediately. Then we'll get another + * 3 after the 1 second. + */ + public void testAggregatorWithTimeoutReturnsOnlyOneMessage() throws Exception { + sendMessage(input_for_aggregator_without_explicit_timeout, 62000); + } + + private void sendMessage(DirectChannel channel, int waitSeconds) { + List list = new ArrayList(); + list.add("foo"); + list.add("bar"); + list.add("baz"); + + reply.purge(null); + channel.send(MessageBuilder.withPayload(list).setReplyChannel(reply).build()); + + Message replyMessage; + int messageCount = 0; + do { + replyMessage = reply.receive(waitSeconds); + if (null != replyMessage) { + System.out.println("Message Received: " + replyMessage); + messageCount++; + } + } while (null != replyMessage); + + Assert.assertEquals(1, messageCount); + } +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/aggregation-resend-config.xml b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/aggregation-resend-config.xml new file mode 100644 index 0000000000..79a7fc8fce --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/aggregation-resend-config.xml @@ -0,0 +1,34 @@ + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file