INT-654

org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 the original author or authors.
+ * Copyright 2002-2009 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.
@@ -19,7 +19,9 @@ package org.springframework.integration.config.xml;
 import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.util.StringUtils;
 
 /**
  * Parser for the <resequencer> element.
@@ -33,6 +35,7 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
 				IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.Resequencer");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "discard-channel");
+		this.configureCorrelationStrategy(builder, element, parserContext);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "release-partial-sequences");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-partial-result-on-timeout");
@@ -43,4 +46,24 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
 		return builder;
 	}
 
+	private void configureCorrelationStrategy(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
+		String ref = element.getAttribute("correlation-strategy");
+		String method = element.getAttribute("correlation-strategy-method");
+		String correlationStrategyProperty = "correlationStrategy";
+		if (StringUtils.hasText(ref)) {
+			if (StringUtils.hasText(method)) {
+				BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
+						IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelationStrategyAdapter");
+				adapterBuilder.addConstructorArgReference(ref);
+				adapterBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String");
+				String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
+						adapterBuilder.getBeanDefinition(), parserContext.getRegistry());
+				builder.addPropertyReference(correlationStrategyProperty, adapterBeanName);
+			}
+			else {
+				builder.addPropertyReference(correlationStrategyProperty, ref);
+			}
+		}
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd

@@ -938,6 +938,16 @@
 			</xsd:annotation>
 			<xsd:complexContent>
 				<xsd:extension base="handlerEndpointType">
+					<xsd:attribute name="correlation-strategy" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="java.lang.Object" />
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="correlation-strategy-method" type="xsd:string" />
 					<xsd:attribute name="discard-channel" type="xsd:string">
 						<xsd:annotation>
 							<xsd:appinfo>

org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 the original author or authors.
+ * Copyright 2002-2009 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.
@@ -30,6 +30,8 @@ import org.junit.Test;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.aggregator.CorrelationStrategy;
+import org.springframework.integration.aggregator.CorrelationStrategyAdapter;
 import org.springframework.integration.aggregator.Resequencer;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.core.Message;
@@ -39,6 +41,7 @@ import org.springframework.integration.message.MessageBuilder;
 
 /**
  * @author Marius Bogoevici
+ * @author Mark Fisher
  */
 public class ResequencerParserTests {
 
@@ -117,6 +120,25 @@ public class ResequencerParserTests {
 				false, getPropertyValue(resequencer, "releasePartialSequences"));
 	}
 
+	@Test
+	public void testCorrelationStrategyRefOnly() throws Exception {
+		EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithCorrelationStrategyRefOnly");
+		Resequencer resequencer = (Resequencer) new DirectFieldAccessor(endpoint).getPropertyValue("handler");
+		assertEquals("The ResequencerEndpoint is not configured with the appropriate CorrelationStrategy",
+				context.getBean("testCorrelationStrategy"), getPropertyValue(resequencer, "correlationStrategy"));
+	}
+
+	@Test
+	public void testCorrelationStrategyRefAndMethod() throws Exception {
+		EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithCorrelationStrategyRefAndMethod");
+		Resequencer resequencer = (Resequencer) new DirectFieldAccessor(endpoint).getPropertyValue("handler");
+		Object correlationStrategy = getPropertyValue(resequencer, "correlationStrategy");
+		assertEquals("The ResequencerEndpoint is not configured with a CorrelationStrategy adapter",
+				CorrelationStrategyAdapter.class, correlationStrategy.getClass());
+		CorrelationStrategyAdapter adapter = (CorrelationStrategyAdapter) correlationStrategy;
+		assertEquals("foo", adapter.getCorrelationKey(MessageBuilder.withPayload("not important").build()));
+	}
+
 
 	private static <T> Message<T> createMessage(T payload, Object correlationId,
 			int sequenceSize, int sequenceNumber, MessageChannel outputChannel) {
@@ -128,4 +150,20 @@ public class ResequencerParserTests {
 				.build();
 	}
 
+
+	static class TestCorrelationStrategy implements CorrelationStrategy {
+
+		public Object getCorrelationKey(Message<?> message) {
+			return "test";
+		}
+	}
+
+
+	static class TestCorrelationStrategyPojo {
+
+		public Object foo(Object o) {
+			return "foo";
+		}
+	}
+
 }

org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml

@@ -21,6 +21,10 @@
 
 	<channel id="inputChannel2"/>
 
+	<channel id="inputChannel3"/>
+
+	<channel id="inputChannel4"/>
+
 	<resequencer id="completelyDefinedResequencer" 
 		input-channel="inputChannel2"
 		output-channel="outputChannel" 
@@ -32,4 +36,19 @@
 		timeout="42" 
 		release-partial-sequences="false"/>
 
+	<resequencer id="resequencerWithCorrelationStrategyRefOnly"
+		input-channel="inputChannel3"
+		correlation-strategy="testCorrelationStrategy"/>
+
+	<resequencer id="resequencerWithCorrelationStrategyRefAndMethod"
+		input-channel="inputChannel4"
+		correlation-strategy="testCorrelationStrategyPojo"
+		correlation-strategy-method="foo"/>
+
+	<beans:bean id="testCorrelationStrategy"
+		class="org.springframework.integration.config.ResequencerParserTests$TestCorrelationStrategy"/>
+
+	<beans:bean id="testCorrelationStrategyPojo"
+		class="org.springframework.integration.config.ResequencerParserTests$TestCorrelationStrategyPojo"/>
+
 </beans:beans>