The <aggregator/> element now creates a HandlerEndpoint instance. It now requires "input-channel" and "output-channel" (and it no longer accepts "default-reply-channel"). It also no longer accepts a <completion-strategy/> sub-element. Instead, for a pojo-based adapter, add the "completion-strategy-method" attribute along with the "completion-strategy" reference (INT-285).

2008-07-06 20:35:36 +00:00
parent 2128c1e21e
commit 5dc6488de0
15 changed files with 312 additions and 138 deletions
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java
@@ -80,6 +80,10 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
 		return true;
 	}

+	protected boolean shouldCreateAdapter(Element element) {
+		return StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE));
+	}
+
 	@Override
 	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
 		String ref = element.getAttribute(REF_ATTRIBUTE);
@@ -87,7 +91,7 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
 			throw new ConfigurationException("The '" + REF_ATTRIBUTE + "' attribute is required.");
 		}
 		String method = element.getAttribute(METHOD_ATTRIBUTE);
-		if (StringUtils.hasText(method)) {
+		if (this.shouldCreateAdapter(element)) {
 			String adapterBeanName = this.parseAdapter(ref, method, element, parserContext);
 			builder.addConstructorArgReference(adapterBeanName);
 		}
@@ -141,7 +145,7 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
 	}


-	private String parseAdapter(String ref, String method, Element element, ParserContext parserContext) {
+	protected String parseAdapter(String ref, String method, Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(this.getHandlerAdapterClass());
 		builder.addPropertyValue("object", new RuntimeBeanReference(ref));
 		builder.addPropertyValue("methodName", method);
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AggregatorParser.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AggregatorParser.java
@@ -17,15 +17,11 @@
 package org.springframework.integration.config;

 import org.w3c.dom.Element;
-import org.w3c.dom.NodeList;

-import org.springframework.beans.factory.config.BeanDefinition;
-import org.springframework.beans.factory.config.RuntimeBeanReference;
-import org.springframework.beans.factory.parsing.BeanComponentDefinition;
-import org.springframework.beans.factory.support.RootBeanDefinition;
-import org.springframework.beans.factory.xml.BeanDefinitionParser;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.ConfigurationException;
+import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.router.AggregatingMessageHandler;
 import org.springframework.integration.router.AggregatorAdapter;
 import org.springframework.integration.router.CompletionStrategyAdapter;
@@ -36,18 +32,13 @@ import org.springframework.util.StringUtils;
 * Registers the annotation-driven post-processors.
 * 
 * @author Marius Bogoevici
+ * @author Mark Fisher
 */
-public class AggregatorParser implements BeanDefinitionParser {
+public class AggregatorParser extends AbstractHandlerEndpointParser {

-	public static final String ID_ATTRIBUTE = "id";
+	public static final String COMPLETION_STRATEGY_REF_ATTRIBUTE = "completion-strategy";

-	public static final String REF_ATTRIBUTE = "ref";
-
-	public static final String METHOD_ATTRIBUTE = "method";
-
-	public static final String COMPLETION_STRATEGY_ATTRIBUTE = "completion-strategy";
-
-	public static final String DEFAULT_REPLY_CHANNEL_ATTRIBUTE = "default-reply-channel";
+	public static final String COMPLETION_STRATEGY_METHOD_ATTRIBUTE = "completion-strategy-method";

 	public static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";

@@ -79,86 +70,61 @@ public class AggregatorParser implements BeanDefinitionParser {

 	public static final String AGGREGATOR_ELEMENT = "aggregator";

-	public static final String COMPLETION_STRATEGY_ELEMENT = "completion-strategy";

-
-	public BeanDefinition parse(Element element, ParserContext parserContext) {
-		return parseAggregatorElement(element, parserContext, true);
+	@Override
+	protected Class<? extends MessageHandler> getHandlerAdapterClass() {
+		return AggregatingMessageHandler.class;
 	}

-	private BeanDefinition parseAggregatorElement(Element element, ParserContext parserContext, boolean topLevel) {
-		final RootBeanDefinition aggregatorDef = new RootBeanDefinition(AggregatingMessageHandler.class);
-		aggregatorDef.setSource(parserContext.extractSource(element));
-		final String id = element.getAttribute(ID_ATTRIBUTE);
-		final String ref = element.getAttribute(REF_ATTRIBUTE);
-		final String method = element.getAttribute(METHOD_ATTRIBUTE);
-		final String completionStrategyRef = element.getAttribute(COMPLETION_STRATEGY_ATTRIBUTE);
-		final NodeList completionStrategyChildElements = element.getElementsByTagName(COMPLETION_STRATEGY_ELEMENT);
-		if (!StringUtils.hasText(ref)) {
-			throw new ConfigurationException("The 'ref' attribute must be present");
-		}
-		if (!topLevel && StringUtils.hasText(id)) {
-			parserContext.getReaderContext().error(
-					"The 'id' attribute is only supported for top-level <aggregator> elements.",
-					parserContext.extractSource(element));
-		}
-		if (completionStrategyChildElements.getLength() > 0 && StringUtils.hasText(completionStrategyRef)) {
-			parserContext
-					.getReaderContext()
-					.error(
-							"The 'completion-strategy' element is only supported when no 'completion-strategy' attribute is specified.",
-							parserContext.extractSource(element));
-		}
-		if (!StringUtils.hasText(method)) {
-			aggregatorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(ref));
+	@Override
+	protected boolean shouldCreateAdapter(Element element) {
+		return true;
+	}
+
+	@Override
+	protected String parseAdapter(String ref, String method, Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(this.getHandlerAdapterClass());
+		if (StringUtils.hasText(method)) {
+			String aggregatorAdapterBeanName = this.createAdapter(ref, method, parserContext, AggregatorAdapter.class);
+			builder.addConstructorArgReference(aggregatorAdapterBeanName);
 		}
 		else {
-			String adapterBeanName = createAdapterAndReturnBeanName(parserContext, ref, method, AggregatorAdapter.class);
-			aggregatorDef.getConstructorArgumentValues().addGenericArgumentValue(
-					new RuntimeBeanReference(adapterBeanName));
+			builder.addConstructorArgReference(ref);
 		}
-
+		final String completionStrategyRef = element.getAttribute(COMPLETION_STRATEGY_REF_ATTRIBUTE);
+		final String completionStrategyMethod = element.getAttribute(COMPLETION_STRATEGY_METHOD_ATTRIBUTE);
 		if (StringUtils.hasText(completionStrategyRef)) {
-			aggregatorDef.getPropertyValues().addPropertyValue(COMPLETION_STRATEGY_PROPERTY,
-					new RuntimeBeanReference(completionStrategyRef));
+			if (StringUtils.hasText(completionStrategyMethod)) {
+				String adapterBeanName = this.createAdapter(completionStrategyRef,
+						completionStrategyMethod, parserContext, CompletionStrategyAdapter.class);
+				builder.addPropertyReference(COMPLETION_STRATEGY_PROPERTY, adapterBeanName);	
+			}
+			else {
+				builder.addPropertyReference(COMPLETION_STRATEGY_PROPERTY, completionStrategyRef);
+			}
 		}
-		else if (completionStrategyChildElements.getLength() > 0) {
-			Element completionStrategyElement = (Element) completionStrategyChildElements.item(0);
-			String childCompletionStrategyReference = completionStrategyElement.getAttribute(REF_ATTRIBUTE);
-			String childCompletionStrategyMethod = completionStrategyElement.getAttribute(METHOD_ATTRIBUTE);
-			String adapterBeanName = createAdapterAndReturnBeanName(parserContext, childCompletionStrategyReference,
-					childCompletionStrategyMethod, CompletionStrategyAdapter.class);
-			aggregatorDef.getPropertyValues().addPropertyValue(COMPLETION_STRATEGY_PROPERTY,
-					new RuntimeBeanReference(adapterBeanName));
-		}
-
-		IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, DEFAULT_REPLY_CHANNEL_PROPERTY,
-				element, DEFAULT_REPLY_CHANNEL_ATTRIBUTE);
-		IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, DISCARD_CHANNEL_PROPERTY, element,
+		IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(builder, DEFAULT_REPLY_CHANNEL_PROPERTY,
+				element, OUTPUT_CHANNEL_ATTRIBUTE);
+		IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(builder, DISCARD_CHANNEL_PROPERTY, element,
 				DISCARD_CHANNEL_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(aggregatorDef, SEND_TIMEOUT_PROPERTY, element,
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, SEND_TIMEOUT_PROPERTY, element,
 				SEND_TIMEOUT_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(aggregatorDef, SEND_PARTIAL_RESULT_ON_TIMEOUT_PROPERTY,
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, SEND_PARTIAL_RESULT_ON_TIMEOUT_PROPERTY,
 				element, SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(aggregatorDef, REAPER_INTERVAL_PROPERTY, element,
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, REAPER_INTERVAL_PROPERTY, element,
 				REAPER_INTERVAL_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(aggregatorDef, TRACKED_CORRELATION_ID_CAPACITY_PROPERTY,
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, TRACKED_CORRELATION_ID_CAPACITY_PROPERTY,
 				element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(aggregatorDef, TIMEOUT, element, TIMEOUT_ATTRIBUTE);
-		String beanName = StringUtils.hasText(id) ? id : parserContext.getReaderContext().generateBeanName(
-				aggregatorDef);
-		parserContext.registerBeanComponent(new BeanComponentDefinition(aggregatorDef, beanName));
-		return aggregatorDef;
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, TIMEOUT, element, TIMEOUT_ATTRIBUTE);
+		return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
 	}

-	private String createAdapterAndReturnBeanName(ParserContext parserContext, final String ref, final String method,
-			Class<?> adapterClass) {
-		BeanDefinition adapterDefinition = new RootBeanDefinition(adapterClass);
-		adapterDefinition.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(ref));
-		adapterDefinition.getConstructorArgumentValues().addGenericArgumentValue(method);
-		String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDefinition);
-		parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDefinition, adapterBeanName));
-		return adapterBeanName;
+	private String createAdapter(String ref, String method, ParserContext parserContext, Class<?> adapterClass) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(adapterClass);
+		builder.addConstructorArgReference(ref);
+		builder.addConstructorArgValue(method);
+		return BeanDefinitionReaderUtils.registerWithGeneratedName(
+				builder.getBeanDefinition(), parserContext.getRegistry());
 	}

 }
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java
@@ -19,9 +19,8 @@ package org.springframework.integration.config;
 import org.w3c.dom.Element;

 import org.springframework.beans.factory.config.BeanDefinitionHolder;
-import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.parsing.BeanComponentDefinition;
-import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.util.StringUtils;
@@ -45,11 +44,11 @@ public abstract class IntegrationNamespaceUtils {
 	 * @param attributeName - the name of the attribute whose value will be set
 	 * on the property
 	 */
-	public static void setValueIfAttributeDefined(RootBeanDefinition beanDefinition, String propertyName,
+	public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, String propertyName,
 			Element element, String attributeName) {
 		final String attributeValue = element.getAttribute(attributeName);
 		if (StringUtils.hasText(attributeValue)) {
-			beanDefinition.getPropertyValues().addPropertyValue(propertyName, attributeValue);
+			builder.addPropertyValue(propertyName, attributeValue);
 		}
 	}

@@ -64,11 +63,11 @@ public abstract class IntegrationNamespaceUtils {
 	 * @param attributeName - the id of the bean which will be used to populate
 	 * the property
 	 */
-	public static void setBeanReferenceIfAttributeDefined(RootBeanDefinition beanDefinition, String propertyName,
+	public static void setBeanReferenceIfAttributeDefined(BeanDefinitionBuilder builder, String propertyName,
 			Element element, String attributeName) {
 		final String attributeValue = element.getAttribute(attributeName);
 		if (StringUtils.hasText(attributeValue)) {
-			beanDefinition.getPropertyValues().addPropertyValue(propertyName, new RuntimeBeanReference(attributeValue));
+			builder.addPropertyReference(propertyName, attributeValue);
 		}
 	}

--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
@@ -355,15 +355,9 @@
 				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:complexContent>
-				<xsd:extension base="beans:identifiedType">
-					<xsd:sequence>
-						<xsd:element name="completion-strategy" minOccurs="0" maxOccurs="1"/>
-					</xsd:sequence>
-					<xsd:attribute name="ref" type="xsd:string" use="required"/>
-					<xsd:attribute name="method" type="xsd:string" use="optional"/>
+				<xsd:extension base="inputOutputHandlerEndpointType">
 					<xsd:attribute name="completion-strategy" type="xsd:string" use="optional"/>
 					<xsd:attribute name="completion-strategy-method" type="xsd:string" use="optional"/>
-					<xsd:attribute name="default-reply-channel" type="xsd:string" use="optional"/>
 					<xsd:attribute name="discard-channel" type="xsd:string" use="optional"/>
 					<xsd:attribute name="send-timeout" type="xsd:long" use="optional"/>
 					<xsd:attribute name="send-partial-result-on-timeout"  type="xsd:boolean" use="optional"/>
--- a/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java
@@ -62,7 +62,7 @@ public class AggregatingMessageHandler extends AbstractMessageBarrierHandler {
 		this(aggregator, null);
 	}

-	
+
 	/**
 	 * Strategy to determine whether the group of messages is complete.
 	 */
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
@@ -26,10 +26,10 @@ import org.junit.Test;

 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanCreationException;
-import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.endpoint.HandlerEndpoint;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.router.AggregatingMessageHandler;
@@ -39,6 +39,7 @@ import org.springframework.integration.util.MethodInvoker;

 /**
 * @author Marius Bogoevici
+ * @author Mark Fisher
 */
 public class AggregatorParserTests {

@@ -52,8 +53,8 @@ public class AggregatorParserTests {

 	@Test
 	public void testAggregation() {
-		AggregatingMessageHandler aggregatingHandler = (AggregatingMessageHandler) context
-				.getBean("aggregatorWithReference");
+		HandlerEndpoint endpoint = (HandlerEndpoint) context.getBean("aggregatorWithReference");
+		AggregatingMessageHandler aggregatingHandler = (AggregatingMessageHandler) endpoint.getHandler();
 		TestAggregator aggregatorBean = (TestAggregator) context.getBean("aggregatorBean");
 		List<Message<?>> outboundMessages = new ArrayList<Message<?>>();
 		outboundMessages.add(createMessage("123", "id1", 3, 1, null));
@@ -71,8 +72,9 @@ public class AggregatorParserTests {

 	@Test
 	public void testPropertyAssignment() throws Exception {
-		AggregatingMessageHandler completeAggregatingMessageHandler = (AggregatingMessageHandler) context
-				.getBean("completelyDefinedAggregator");
+		HandlerEndpoint endpoint = (HandlerEndpoint) context.getBean("completelyDefinedAggregator");
+		AggregatingMessageHandler completeAggregatingMessageHandler =
+				(AggregatingMessageHandler) endpoint.getHandler(); 
 		TestAggregator testAggregator = (TestAggregator) context.getBean("aggregatorBean");
 		CompletionStrategy completionStrategy = (CompletionStrategy) context.getBean("completionStrategy");
 		MessageChannel defaultReplyChannel = (MessageChannel) context.getBean("replyChannel");
@@ -103,8 +105,9 @@ public class AggregatorParserTests {

 	@Test
 	public void testSimpleJavaBeanAggregator() {
-		AggregatingMessageHandler addingAggregator = (AggregatingMessageHandler) context.getBean("aggregatorWithReferenceAndMethod");
 		List<Message<?>> outboundMessages = new ArrayList<Message<?>>();
+		HandlerEndpoint endpoint = (HandlerEndpoint) context.getBean("aggregatorWithReferenceAndMethod");
+		AggregatingMessageHandler addingAggregator = (AggregatingMessageHandler) endpoint.getHandler();
 		outboundMessages.add(createMessage(1l, "id1", 3, 1, null));
 		outboundMessages.add(createMessage(2l, "id1", 3, 3, null));
 		outboundMessages.add(createMessage(3l, "id1", 3, 2, null));
@@ -121,21 +124,23 @@ public class AggregatorParserTests {
 		context = new ClassPathXmlApplicationContext("invalidMethodNameAggregator.xml", this.getClass());		
 	}

-	@Test(expected=BeanDefinitionParsingException.class)
+	@Test(expected=BeanCreationException.class)
 	public void testDuplicateCompletionStrategyDefinition() {
-		context = new ClassPathXmlApplicationContext("completionStrategyMethodWithMissingReference.xml", this.getClass());		
+		context = new ClassPathXmlApplicationContext(
+				"completionStrategyMethodWithMissingReference.xml", this.getClass());		
 	}

 	@Test
 	public void testAggregatorWithPojoCompletionStrategy(){
-		AggregatingMessageHandler aggregatorWithPojoCompletionStrategy = (AggregatingMessageHandler) context.getBean("aggregatorWithPojoCompletionStrategy");
-		CompletionStrategy completionStrategy = (CompletionStrategy)new DirectFieldAccessor(aggregatorWithPojoCompletionStrategy).getPropertyValue("completionStrategy");
+		HandlerEndpoint endpoint = (HandlerEndpoint) context.getBean("aggregatorWithPojoCompletionStrategy");
+		AggregatingMessageHandler aggregatorWithPojoCompletionStrategy = (AggregatingMessageHandler) endpoint.getHandler();
+		CompletionStrategy completionStrategy = (CompletionStrategy)
+				new DirectFieldAccessor(aggregatorWithPojoCompletionStrategy).getPropertyValue("completionStrategy");
 		Assert.assertTrue(completionStrategy instanceof CompletionStrategyAdapter);
 		DirectFieldAccessor completionStrategyAccessor = new DirectFieldAccessor(completionStrategy);
 		MethodInvoker invoker = (MethodInvoker) completionStrategyAccessor.getPropertyValue("invoker");
 		Assert.assertTrue(new DirectFieldAccessor(invoker).getPropertyValue("object") instanceof MaxValueCompletionStrategy);
 		Assert.assertTrue(((Method)completionStrategyAccessor.getPropertyValue("method")).getName().equals("checkCompleteness"));
-		
 		aggregatorWithPojoCompletionStrategy.handle(createMessage(1l, "id1", 0 , 0, null));
 		aggregatorWithPojoCompletionStrategy.handle(createMessage(2l, "id1", 0 , 0, null));
 		aggregatorWithPojoCompletionStrategy.handle(createMessage(3l, "id1", 0 , 0, null));
@@ -148,9 +153,9 @@ public class AggregatorParserTests {
 		Assert.assertEquals(11l, reply.getPayload());
 	}

-	@Test(expected=BeanDefinitionParsingException.class)
-	public void testAggregatorWithDuplicateCompletionStrategy() {
-		context = new ClassPathXmlApplicationContext("duplicateCompletionStrategy.xml", this.getClass());		
+	@Test(expected=BeanCreationException.class)
+	public void testAggregatorWithInvalidCompletionStrategyMethod() {
+		context = new ClassPathXmlApplicationContext("invalidCompletionStrategyMethod.xml", this.getClass());		
 	}


--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml
@@ -9,27 +9,37 @@

 	<message-bus/>

-	<aggregator id="aggregatorWithReference" ref="aggregatorBean" />
+	<channel id="inputChannel"/>
+	<channel id="replyChannel"/>
+	<channel id="discardChannel"/>

-	<aggregator id="completelyDefinedAggregator" ref="aggregatorBean"
+	<aggregator id="aggregatorWithReference" ref="aggregatorBean" input-channel="inputChannel"/>
+
+	<aggregator id="completelyDefinedAggregator"
+		input-channel="inputChannel"
+		output-channel="replyChannel"
+		discard-channel="discardChannel"
+		ref="aggregatorBean" 
 		completion-strategy="completionStrategy"
-		default-reply-channel="replyChannel" discard-channel="discardChannel"
-		send-timeout="86420000" send-partial-result-on-timeout="true"
-		reaper-interval="135" tracked-correlation-id-capacity="99"
-		timeout="42" />
+		send-timeout="86420000"
+		send-partial-result-on-timeout="true"
+		reaper-interval="135"
+		tracked-correlation-id-capacity="99"
+		timeout="42"/>

-	<aggregator id="aggregatorWithReferenceAndMethod" ref="adderBean"
-		method="add" default-reply-channel="replyChannel" />
+	<aggregator id="aggregatorWithReferenceAndMethod"
+		ref="adderBean"
+		method="add"
+		input-channel="inputChannel"
+		output-channel="replyChannel"/>

 	<aggregator id="aggregatorWithPojoCompletionStrategy"
-		ref="adderBean" method="add" default-reply-channel="replyChannel">
-		<completion-strategy ref="pojoCompletionStrategy"
-			method="checkCompleteness" />
-	</aggregator>
-
-	<channel id="replyChannel" />
-
-	<channel id="discardChannel" />
+		input-channel="inputChannel"
+		output-channel="replyChannel"
+		ref="adderBean"
+		method="add"
+		completion-strategy="pojoCompletionStrategy"
+		completion-strategy-method="checkCompleteness"/>

 	<beans:bean id="aggregatorBean"
 		class="org.springframework.integration.config.TestAggregator" />
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/completionStrategyMethodWithMissingReference.xml
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/completionStrategyMethodWithMissingReference.xml
@@ -9,14 +9,15 @@
 				
 	<message-bus/>
 	
-	<aggregator id="aggregator" ref="adderBean" method="add" completion-strategy="testCompletionStrategy" default-reply-channel="replyChannel">
-		<completion-strategy ref="testCompletionStrategy"/>
+	<aggregator id="aggregator" ref="adderBean" method="add" completion-strategy="testCompletionStrategy"
+		input-channel="input-channel" output-channel="replyChannel">
 	</aggregator>

+	<channel id="inputChannel"/>
 	<channel id="replyChannel"/>

 	<beans:bean id="adderBean" class="org.springframework.integration.config.Adder"/>
 	
-	<beans:bean id="completionStrategyBean" class="org.springframework.integration.config.TestCompletionStrategy"></beans:bean>
+	<beans:bean id="completionStrategyBean" class="org.springframework.integration.config.TestCompletionStrategy"/>

 </beans:beans>
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/invalidCompletionStrategyMethod.xml
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/invalidCompletionStrategyMethod.xml
@@ -7,13 +7,15 @@
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">

-	<aggregator id="aggregatorWithPojoCompletionStrategy" completion-strategy="completionStrategy"
-		ref="adderBean" method="add" default-reply-channel="replyChannel">
-		<completion-strategy ref="pojoCompletionStrategy"
-			method="checkCompleteness" />
-	</aggregator>
+	<aggregator id="aggregatorWithPojoCompletionStrategy"
+		completion-strategy="completionStrategy"
+		ref="adderBean" method="add"
+		input-channel="inputChannel"
+		output-channel="replyChannel"
+		completion-strategy-method="invalidMethodName"/>

-	<channel id="replyChannel" />
+	<channel id="inputChannel"/>
+	<channel id="replyChannel"/>

 	<beans:bean id="adderBean"
 		class="org.springframework.integration.config.Adder" />
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/invalidMethodNameAggregator.xml
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/invalidMethodNameAggregator.xml
@@ -8,11 +8,13 @@
 			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
 			
 	<message-bus/>		
-			
-	<aggregator id="aggregatorWithReferenceAndMethod" ref="adderBean" method="substract" default-reply-channel="replyChannel"/>

+	<channel id="inputChannel"/>
 	<channel id="replyChannel"/>

+	<aggregator id="aggregatorWithReferenceAndMethod" ref="adderBean" method="substract"
+		input-channel="inputChannel" output-channel="replyChannel"/>
+
 	<beans:bean id="adderBean" class="org.springframework.integration.config.Adder"/>

 </beans:beans>
--- a/org.springframework.integration/src/test/java/org/springframework/integration/router/config/NumberAdder.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/router/config/NumberAdder.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2002-2008 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.config;
+
+import java.util.List;
+
+/**
+ * @author Mark Fisher
+ */
+public class NumberAdder {
+
+	public Integer sum(List<Integer> values) {
+		int result = 0;
+		for (Integer value : values) {
+			result += value;
+		}
+		return result;
+	}
+
+}
--- a/org.springframework.integration/src/test/java/org/springframework/integration/router/config/NumberSplitter.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/router/config/NumberSplitter.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2002-2008 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.config;
+
+import java.util.List;
+
+/**
+ * @author Mark Fisher
+ */
+public class NumberSplitter {
+
+	public List<Integer> split(Numbers numbers) {
+		return numbers.getValues();
+	}
+
+}
--- a/org.springframework.integration/src/test/java/org/springframework/integration/router/config/Numbers.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/router/config/Numbers.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-2008 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.config;
+
+import java.util.List;
+
+/**
+ * @author Mark Fisher
+ */
+public class Numbers {
+
+	private final List<Integer> values;
+
+
+	public Numbers(List<Integer> values) {
+		this.values = values;
+	}
+
+	public List<Integer> getValues() {
+		return this.values;
+	}
+
+}
--- a/org.springframework.integration/src/test/java/org/springframework/integration/router/config/SplitterAggregatorTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/router/config/SplitterAggregatorTests.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2002-2008 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.config;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.message.Message;
+
+/**
+ * @author Mark Fisher
+ */
+public class SplitterAggregatorTests {
+
+	private final AtomicInteger count = new AtomicInteger();
+
+
+	@Test
+	public void testSplitterAndAggregator() {
+		ApplicationContext context = new ClassPathXmlApplicationContext(
+				"splitterAggregatorTests.xml", this.getClass());
+		MessageChannel inputChannel = (MessageChannel) context.getBean("numbers");
+		MessageChannel outputChannel = (MessageChannel) context.getBean("results");
+		inputChannel.send(new GenericMessage<Numbers>(this.nextTen()));
+		Message<?> result1 = outputChannel.receive(1000);
+		assertNotNull(result1);
+		assertEquals(Integer.class, result1.getPayload().getClass());
+		assertEquals(55, result1.getPayload());
+		inputChannel.send(new GenericMessage<Numbers>(this.nextTen()));		
+		Message<?> result2 = outputChannel.receive(1000);
+		assertNotNull(result2);
+		assertEquals(Integer.class, result2.getPayload().getClass());
+		assertEquals(155, result2.getPayload());
+	}
+
+	private Numbers nextTen() {
+		List<Integer> values = new ArrayList<Integer>();
+		for (int i = 0; i < 10; i++) {
+			values.add(this.count.incrementAndGet());
+		}
+		return new Numbers(values);
+	}
+
+}
--- a/org.springframework.integration/src/test/java/org/springframework/integration/router/config/splitterAggregatorTests.xml
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/router/config/splitterAggregatorTests.xml
@@ -0,0 +1,22 @@
+<?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-core-1.0.xsd">
+
+	<message-bus/>
+
+	<channel id="numbers"/>
+	<channel id="splits"/>
+	<channel id="results"/>
+
+	<splitter ref="splitter" method="split" input-channel="numbers" output-channel="splits"/>
+	<aggregator ref="aggregator" method="sum" input-channel="splits" output-channel="results"/>
+
+	<beans:bean id="splitter" class="org.springframework.integration.router.config.NumberSplitter"/>
+	<beans:bean id="aggregator" class="org.springframework.integration.router.config.NumberAdder"/>
+
+</beans:beans>