INT-1093:

- Added testcases from the issue
- added convenience method for total completion to MessageGroup
- resolved issue by invoking callback

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<String, Object> headers = this.aggregateHeaders(group);
 		Message<?> message = MessageBuilder.withPayload(payload).copyHeadersIfAbsent(headers).build();
 		channelTemplate.send(message, outputChannel);
+		group.onCompleteProcessing();
 	}
 
 	/**

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();
+    }
+
 }

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<String> list = new ArrayList<String>();
+		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);
+	}
+}

org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/aggregation-resend-config.xml

@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
+			 xmlns="http://www.springframework.org/schema/integration"
+			 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+			 xsi:schemaLocation="http://www.springframework.org/schema/beans
+            http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+            http://www.springframework.org/schema/integration
+            http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
+
+	<channel id="input_for_aggregator_with_explicit_timeout"/>
+	<channel id="input_for_aggregator_without_explicit_timeout"/>
+	<channel id="aggregator_with_explicit_timeout_channel"/>
+	<channel id="aggregator_without_explicit_timeout_channel"/>
+
+	<channel id="reply">
+		<queue capacity="10"/>
+	</channel>
+
+	<splitter id="splitter_to_aggregator_with_explicit_timeout"
+			  input-channel="input_for_aggregator_with_explicit_timeout"
+			  output-channel="aggregator_with_explicit_timeout_channel"/>
+
+	<splitter id="splitter_to_aggregator_without_explicit_timeout"
+			  input-channel="input_for_aggregator_without_explicit_timeout"
+			  output-channel="aggregator_without_explicit_timeout_channel"/>
+
+	<aggregator id="aggregator_with_explicit_timeout"
+				timeout="1000"
+				input-channel="aggregator_with_explicit_timeout_channel"/>
+
+	<aggregator id="aggregator_without_explicit_timeout"
+				input-channel="aggregator_without_explicit_timeout_channel"/>
+
+</beans:beans>