diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java index 32cb2c41d8..d3c5814223 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java +++ b/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); + } + } + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd index 7998dd162d..d086950ab5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd @@ -938,6 +938,16 @@ + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index e619b5de8c..375c4a978c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/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 Message 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"; + } + } + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml index cced5aed7c..3a5a51e521 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/resequencerParserTests.xml @@ -21,6 +21,10 @@ + + + + + + + + + + + +