diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java index 4782033bac..a78573d301 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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 @@ -12,19 +12,19 @@ */ package org.springframework.integration.config.xml; -import org.w3c.dom.Element; - import org.springframework.beans.BeanMetadataElement; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; /** * Base class for parsers that create an instance of {@link AbstractCorrelatingMessageHandler} * * @author Oleg Zhurakousky + * @author Stefan Ferstl * @since 2.1 * */ @@ -38,6 +38,14 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy"; + private static final String RELEASE_STRATEGY_REF_ATTRIBUTE = "release-strategy"; + + private static final String RELEASE_STRATEGY_METHOD_ATTRIBUTE = "release-strategy-method"; + + private static final String RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE = "release-strategy-expression"; + + private static final String RELEASE_STRATEGY_PROPERTY = "releaseStrategy"; + private static final String MESSAGE_STORE_ATTRIBUTE = "message-store"; private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel"; @@ -50,6 +58,10 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo this.injectPropertyWithAdapter(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE, CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY, "CorrelationStrategy", element, builder, processor, parserContext); + this.injectPropertyWithAdapter(RELEASE_STRATEGY_REF_ATTRIBUTE, RELEASE_STRATEGY_METHOD_ATTRIBUTE, + RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE, RELEASE_STRATEGY_PROPERTY, "ReleaseStrategy", + element, builder, processor, parserContext); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_STORE_ATTRIBUTE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java index 77101e09b1..a473829813 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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. @@ -31,22 +31,15 @@ import org.w3c.dom.Element; /** * Parser for the aggregator element of the integration namespace. Registers the annotation-driven * post-processors. - * + * * @author Marius Bogoevici * @author Mark Fisher * @author Oleg Zhurakousky * @author Dave Syer + * @author Stefan Ferstl */ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser { - - private static final String RELEASE_STRATEGY_REF_ATTRIBUTE = "release-strategy"; - private static final String RELEASE_STRATEGY_METHOD_ATTRIBUTE = "release-strategy-method"; - - private static final String RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE = "release-strategy-expression"; - - private static final String RELEASE_STRATEGY_PROPERTY = "releaseStrategy"; - private static final String EXPIRE_GROUPS_UPON_COMPLETION = "expire-groups-upon-completion"; @Override @@ -89,14 +82,10 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser { processorBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String"); } - + this.doParse(builder, element, processor, parserContext); - + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_COMPLETION); - - this.injectPropertyWithAdapter(RELEASE_STRATEGY_REF_ATTRIBUTE, RELEASE_STRATEGY_METHOD_ATTRIBUTE, - RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE, RELEASE_STRATEGY_PROPERTY, "ReleaseStrategy", element, builder, - processor, parserContext); return builder; } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd index 509bf38a44..2a1d2e02bc 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd @@ -3146,38 +3146,6 @@ is provided, the return value is expected to match a channel name exactly. - - - - - - - - - A reference to a bean that implements the release strategy. - The bean can be an implementation of the - ReleaseStrategy interface or a POJO - - - - - - - - - - - - A method defined on the bean referenced by release-strategy, that implements the completion - decision algorithm. - - - - - - A SpEL expression to evaluate against a root object that is the Collection of messages within the message group (e.g, size() > 6) - - @@ -3219,6 +3187,38 @@ is provided, the return value is expected to match a channel name exactly. based on the 'id' of the 'person' attribute of the message payload object) + + + + + + + + + A reference to a bean that implements the release strategy. + The bean can be an implementation of the + ReleaseStrategy interface or a POJO + + + + + + + + + + + + A method defined on the bean referenced by release-strategy, that implements the completion + decision algorithm. + + + + + + A SpEL expression to evaluate against a root object that is the Collection of messages within the message group (e.g, size() > 6) + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index f30d0069a7..f0f83168ab 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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 @@ -18,18 +18,23 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.springframework.integration.test.util.TestUtils.getPropertyValue; +import java.util.List; + import org.junit.Before; import org.junit.Test; - import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.aggregator.CorrelationStrategy; import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy; +import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy; +import org.springframework.integration.aggregator.ReleaseStrategy; import org.springframework.integration.aggregator.ResequencingMessageHandler; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; @@ -38,6 +43,7 @@ import org.springframework.integration.test.util.TestUtils; * @author Mark Fisher * @author Dave Syer * @author Oleg Zhurakousky + * @author Stefan Ferstl */ public class ResequencerParserTests { @@ -94,6 +100,34 @@ public class ResequencerParserTests { .getBean("testCorrelationStrategy"), getPropertyValue(resequencer, "correlationStrategy")); } + @Test + public void testReleaseStrategyRefOnly() throws Exception { + EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithReleaseStrategyRefOnly"); + ResequencingMessageHandler resequencer = getPropertyValue(endpoint, "handler", ResequencingMessageHandler.class); + assertEquals("The ResequencerEndpoint is not configured with the appropriate ReleaseStrategy", + context.getBean("testReleaseStrategy"), getPropertyValue(resequencer, "releaseStrategy")); + } + + @Test + public void testReleaseStrategyRefAndMethod() throws Exception { + EventDrivenConsumer endpoint = (EventDrivenConsumer) context + .getBean("resequencerWithReleaseStrategyRefAndMethod"); + ResequencingMessageHandler resequencer = getPropertyValue(endpoint, "handler", ResequencingMessageHandler.class); + + Object releaseStrategyBean = context.getBean("testReleaseStrategyPojo"); + assertTrue("Release strategy is not of the expected type", + releaseStrategyBean instanceof TestReleaseStrategyPojo); + TestReleaseStrategyPojo expectedReleaseStrategy = (TestReleaseStrategyPojo) releaseStrategyBean; + + int currentInvocationCount = expectedReleaseStrategy.invocationCount; + ReleaseStrategy effectiveReleaseStrategy = (ReleaseStrategy) getPropertyValue(resequencer, "releaseStrategy"); + assertTrue("The release strategy is expected to be a MethodInvokingReleaseStrategy", + effectiveReleaseStrategy instanceof MethodInvokingReleaseStrategy); + effectiveReleaseStrategy.canRelease(new SimpleMessageGroup("test")); + assertEquals("The ResequencerEndpoint was not invoked the expected number of times;", + currentInvocationCount + 1, expectedReleaseStrategy.invocationCount); + } + @Test public void shouldSetReleasePartialSequencesFlag(){ EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("completelyDefinedResequencer"); @@ -137,4 +171,19 @@ public class ResequencerParserTests { } } + static class TestReleaseStrategy implements ReleaseStrategy { + public boolean canRelease(MessageGroup group) { + return true; + } + } + + static class TestReleaseStrategyPojo { + private int invocationCount = 0; + + public boolean bar(List> messages) { + invocationCount++; + return true; + } + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml index 8881c65ed0..e40dfddfe1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml @@ -46,10 +46,24 @@ correlation-strategy="testCorrelationStrategyPojo" correlation-strategy-method="foo"/> + + + + + + + diff --git a/src/reference/docbook/aggregator.xml b/src/reference/docbook/aggregator.xml index 529db8b0d3..5e5f8dbfce 100644 --- a/src/reference/docbook/aggregator.xml +++ b/src/reference/docbook/aggregator.xml @@ -323,7 +323,7 @@ then you should simply provide an implementation of the ReleaseStrate
- Configuring Aggregator + Configuring an Aggregator
Configuring an Aggregator with XML @@ -343,17 +343,19 @@ then you should simply provide an implementation of the ReleaseStrate order="1" ]]> ]]> ]]> @@ -435,7 +437,7 @@ then you should simply provide an implementation of the ReleaseStrate algorithm. The bean can be an implementation of the CorrelationStrategy interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined as well. Optional (by default, the aggregator will use - the CORRELATION_ID header) . + the MessageHeaders.CORRELATION_ID header) . @@ -445,27 +447,34 @@ then you should simply provide an implementation of the ReleaseStrate restrictions (requires correlation-strategy to be present). - + + A SpEL expression representing the correlation strategy. + Example: "headers['foo']". Only one of + correlation-strategy + or correlation-strategy-expression is allowed. + + + A reference to a bean defined in the application context. The bean must implement the aggregation logic as described above. Optional (by default the list of aggregated Messages will become a payload of the output message). - + A method defined on the bean referenced by ref, that implements the message aggregation algorithm. Optional, depends on ref attribute being defined. - + A reference to a bean that implements the release strategy. The bean can be an implementation of the ReleaseStrategy interface or a POJO. In the latter case the release-strategy-method attribute must be defined as well. Optional (by default, the - aggregator will use the SEQUENCE_SIZE header attribute). + aggregator will use the MessageHeaders.SEQUENCE_SIZE header attribute). - + A method defined on the bean referenced by release-strategy, that implements the completion decision algorithm. Optional, with @@ -473,7 +482,15 @@ then you should simply provide an implementation of the ReleaseStrate present). - + + A SpEL expression representing the release strategy; the root object for the + expression is a Collection of Messages. + Example: "size() == 5". Only one of + release-strategy + or release-strategy-expression is allowed. + + + When set to true (default false), completed groups are removed from the message store, allowing subsequent messages with the same correlation to form a new group. The default behavior diff --git a/src/reference/docbook/resequencer.xml b/src/reference/docbook/resequencer.xml index 63f7556e2c..177e18f141 100644 --- a/src/reference/docbook/resequencer.xml +++ b/src/reference/docbook/resequencer.xml @@ -45,7 +45,14 @@ message-store="messageStore" ]]> ]]> + send-timeout="86420000" ]]>]]> @@ -117,6 +124,52 @@ The timeout for sending out messages. Optional. + + + A reference to a bean that implements the message correlation (grouping) + algorithm. The bean can be an implementation of the CorrelationStrategy + interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined + as well. Optional (by default, the aggregator will use + the MessageHeaders.CORRELATION_ID header) . + + + + A method defined on the bean referenced by + correlation-strategy, that implements the + correlation decision algorithm. Optional, with + restrictions (requires correlation-strategy to be + present). + + + + A SpEL expression representing the correlation strategy. + Example: "headers['foo']". Only one of correlation-strategy + or correlation-strategy-expression is allowed. + + + + A reference to a bean that implements the release strategy. + The bean can be an implementation of the ReleaseStrategy interface + or a POJO. In the latter case the release-strategy-method + attribute must be defined as well. Optional (by default, the + aggregator will use the MessageHeaders.SEQUENCE_SIZE header attribute). + + + + A method defined on the bean referenced by + release-strategy, that implements the + completion decision algorithm. Optional, with + restrictions (requires release-strategy to be + present). + + + + A SpEL expression representing the release strategy; the root object for the + expression is a Collection of Messages. + Example: "size() == 5". Only one of + release-strategy + or release-strategy-expression is allowed. +