OPEN - issue INT-412: Not possible to run parallell aggregators

http://jira.springframework.org/browse/INT-412
Added testcases (food for INT-413)

org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml

@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
+	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">
+	<message-bus enable-annotations="true"/>
+	
+	<channel id="input" />
+	<aggregator ref="summer" method="sum" input-channel="input"
+		output-channel="output">
+		<!-- This doesn't work (see INT-413) -->
+	<!-- 
+	<poller task-executor="executor" interval="20" max-messages-per-poll="5"/>
+	 -->
+	</aggregator>
+	<beans:bean id="executor"
+		class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
+		<beans:property name="maxPoolSize" value="10" />
+	</beans:bean>
+	<channel id="output">
+		<queue capacity="5" />
+	</channel>
+	<beans:bean id="summer"
+		class="org.springframework.integration.aggregator.integration.ConcurrentAggregatorIntegrationTests$SummingAggregator" />
+</beans:beans>

org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java

@@ -0,0 +1,64 @@
+package org.springframework.integration.aggregator.integration;
+
+import static org.junit.Assert.*;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.PollableChannel;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.message.MessageHeaders;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+public class ConcurrentAggregatorIntegrationTests {
+	@Autowired
+	@Qualifier("input")
+	private MessageChannel input;
+
+	@Autowired
+	@Qualifier("output")
+	private PollableChannel output;
+
+	@Test
+	public void configOk() throws Exception {
+		// nothing to assert
+	}
+
+	@Test(timeout = 1000)
+	public void aggregate() throws Exception {
+		for (int i = 0; i < 5; i++) {
+			Map<String, Object> headers = stubHeaders(i, 5, 1);
+			input.send(new GenericMessage<Integer>(i, headers));
+		}
+		assertEquals(0+1+2+3+4, output.receive().getPayload());
+	}
+
+	//configured in context associated with this test
+	public static class SummingAggregator {
+		public Integer sum(List<Integer> numbers) {
+			int result = 0;
+			for (Integer number : numbers) {
+				result += number;
+			}
+			return result;
+		}
+	}
+
+	private Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correllationId) {
+		Map<String, Object> headers = new HashMap<String, Object>();
+		headers.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber);
+		headers.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize);
+		headers.put(MessageHeaders.CORRELATION_ID, correllationId);
+		return headers;
+	}
+
+}