Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #696 from ferstl/INT-2863

Browse Source
This commit is contained in:
Gary Russell
2013-01-15 13:15:32 -05:00
parent 3c59d377f4 f9aea7d5f5
commit f3b1c9d9d2
7 changed files with 204 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java
View File

@@ -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);

21
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
View File

@@ -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 <em>aggregator</em> 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;
}

64
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd
View File

@@ -3146,38 +3146,6 @@ is provided, the return value is expected to match a channel name exactly.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="release-strategy" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.lang.Object" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A reference to a bean that implements the release strategy.
The bean can be an implementation of the
ReleaseStrategy interface or a POJO
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="release-strategy-method" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation>
<tool:expected-method type-ref="@release-strategy" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A method defined on the bean referenced by release-strategy, that implements the completion
decision algorithm.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="release-strategy-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>A SpEL expression to evaluate against a root object that is the Collection of messages within the message group (e.g, size() > 6)</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -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)</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="release-strategy" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.lang.Object" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A reference to a bean that implements the release strategy.
The bean can be an implementation of the
ReleaseStrategy interface or a POJO
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="release-strategy-method" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation>
<tool:expected-method type-ref="@release-strategy" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A method defined on the bean referenced by release-strategy, that implements the completion
decision algorithm.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="release-strategy-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>A SpEL expression to evaluate against a root object that is the Collection of messages within the message group (e.g, size() > 6)</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="discard-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

53
spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

@@ -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<Message<?>> messages) {
invocationCount++;
return true;
}
}
}

14
spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml
View File

@@ -46,10 +46,24 @@
correlation-strategy="testCorrelationStrategyPojo"
correlation-strategy-method="foo"/>
<resequencer id="resequencerWithReleaseStrategyRefOnly"
input-channel="inputChannel5"
release-strategy="testReleaseStrategy"/>
<resequencer id="resequencerWithReleaseStrategyRefAndMethod"
input-channel="inputChannel6"
release-strategy="testReleaseStrategyPojo"
release-strategy-method="bar"/>
<beans:bean id="testCorrelationStrategy"
class="org.springframework.integration.config.ResequencerParserTests$TestCorrelationStrategy"/>
<beans:bean id="testCorrelationStrategyPojo"
class="org.springframework.integration.config.ResequencerParserTests$TestCorrelationStrategyPojo"/>
<beans:bean id="testReleaseStrategy"
class="org.springframework.integration.config.ResequencerParserTests$TestReleaseStrategy" />
<beans:bean id="testReleaseStrategyPojo"
class="org.springframework.integration.config.ResequencerParserTests$TestReleaseStrategyPojo" />
</beans:beans>

49
src/reference/docbook/aggregator.xml
View File

@@ -323,7 +323,7 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
</section>
<section id="aggregator-config">
<title>Configuring Aggregator</title>
<title>Configuring an Aggregator</title>
<section id="aggregator-xml">
<title>Configuring an Aggregator with XML</title>
@@ -343,17 +343,19 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
order="1" ]]><co id="aggxml07" /><![CDATA[
send-partial-result-on-expiry="false" ]]><co id="aggxml08" /><![CDATA[
send-timeout="1000" ]]><co id="aggxml09" /><![CDATA[
correlation-strategy="correlationStrategyBean" ]]><co id="aggxml10" /><![CDATA[
correlation-strategy-method="correlate" ]]><co id="aggxml11" /><![CDATA[
ref="aggregatorBean" ]]><co id="aggxml12" /><![CDATA[
method="aggregate" ]]><co id="aggxml13" /><![CDATA[
release-strategy="releaseStrategyBean" ]]><co id="aggxml14" /><![CDATA[
release-strategy-method="release" ]]><co id="aggxml15" /><![CDATA[
correlation-strategy-expression="headers['foo']" ]]><co id="aggxml12" /><![CDATA[
expire-groups-upon-completion="false"/> ]]><co id="aggxml16" /><![CDATA[
ref="aggregatorBean" ]]><co id="aggxml13" /><![CDATA[
method="aggregate" ]]><co id="aggxml14" /><![CDATA[
release-strategy="releaseStrategyBean" ]]><co id="aggxml15" /><![CDATA[
release-strategy-method="release" ]]><co id="aggxml16" /><![CDATA[
release-strategy-expression="size() == 5" ]]><co id="aggxml17" /><![CDATA[
expire-groups-upon-completion="false"/> ]]><co id="aggxml18" /><![CDATA[
<int:channel id="outputChannel"/>
@@ -435,7 +437,7 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
algorithm. The bean can be an implementation of the <interfacename>CorrelationStrategy</interfacename>
interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined
as well. <emphasis>Optional (by default, the aggregator will use
the CORRELATION_ID header) </emphasis>.</para>
the <code>MessageHeaders.CORRELATION_ID</code> header) </emphasis>.</para>
</callout>
<callout arearefs="aggxml11">
@@ -445,27 +447,34 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
restrictions (requires <code>correlation-strategy</code> to be
present).</emphasis></para>
</callout>
<callout arearefs="aggxml12">
<para>A SpEL expression representing the correlation strategy.
Example: <code>"headers['foo']"</code>. Only one of
<code>correlation-strategy</code>
or <code>correlation-strategy-expression</code> is allowed.</para>
</callout>
<callout arearefs="aggxml13">
<para>A reference to a bean defined in the application context. The bean must implement the aggregation logic
as described above. <emphasis>Optional (by default the list of aggregated Messages will become a
payload of the output message).</emphasis></para>
</callout>
<callout arearefs="aggxml13">
<callout arearefs="aggxml14">
<para>A method defined on the bean referenced by <code>ref</code>,
that implements the message aggregation
algorithm. <emphasis>Optional, depends on <code>ref</code> attribute being defined.</emphasis></para>
</callout>
<callout arearefs="aggxml14">
<callout arearefs="aggxml15">
<para>A reference to a bean that implements the release strategy.
The bean can be an implementation of the <interfacename>ReleaseStrategy</interfacename> interface
or a POJO. In the latter case the release-strategy-method
attribute must be defined as well. <emphasis>Optional (by default, the
aggregator will use the SEQUENCE_SIZE header attribute)</emphasis>.</para>
aggregator will use the <code>MessageHeaders.SEQUENCE_SIZE</code> header attribute)</emphasis>.</para>
</callout>
<callout arearefs="aggxml15">
<callout arearefs="aggxml16">
<para>A method defined on the bean referenced by
<code>release-strategy</code>, that implements the
completion decision algorithm. <emphasis>Optional, with
@@ -473,7 +482,15 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
present).</emphasis></para>
</callout>
<callout arearefs="aggxml16">
<callout arearefs="aggxml17">
<para>A SpEL expression representing the release strategy; the root object for the
expression is a <code>Collection</code> of <code>Message</code>s.
Example: <code>"size() == 5"</code>. Only one of
<code>release-strategy</code>
or <code>release-strategy-expression</code> is allowed.</para>
</callout>
<callout arearefs="aggxml18">
<para>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

55
src/reference/docbook/resequencer.xml
View File

@@ -45,7 +45,14 @@
message-store="messageStore" ]]><co id="resxml6-co" linkends="resxml6" /><![CDATA[
send-partial-result-on-expiry="true" ]]><co id="resxml7-co"
linkends="resxml7" /><![CDATA[
send-timeout="86420000" ]]><co id="resxml10-co" linkends="resxml10" /><![CDATA[ /> ]]></programlisting>
send-timeout="86420000" ]]><co id="resxml10-co" linkends="resxml10" /><![CDATA[
correlation-strategy="correlationStrategyBean" ]]><co id="resxml11-co" linkends="resxml11" /><![CDATA[
correlation-strategy-method="correlate" ]]><co id="resxml12-co" linkends="resxml12" /><![CDATA[
correlation-strategy-expression="headers['foo']" ]]><co id="resxml13-co" linkends="resxml13" /><![CDATA[
release-strategy="releaseStrategyBean" ]]><co id="resxml14-co" linkends="resxml14" /><![CDATA[
release-strategy-method="release" ]]><co id="resxml15-co" linkends="resxml15" /><![CDATA[
release-strategy-expression="size() == 10" />]]><co id="resxml16-co" linkends="resxml16" /></programlisting>
<para><calloutlist>
<callout arearefs="resxml1-co" id="resxml1">
@@ -117,6 +124,52 @@
<para>The timeout for sending out messages.
<emphasis>Optional</emphasis>.</para>
</callout>
<callout arearefs="resxml11-co" id="resxml11">
<para>A reference to a bean that implements the message correlation (grouping)
algorithm. The bean can be an implementation of the <interfacename>CorrelationStrategy</interfacename>
interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined
as well. <emphasis>Optional (by default, the aggregator will use
the <code>MessageHeaders.CORRELATION_ID</code> header) </emphasis>.</para>
</callout>
<callout arearefs="resxml12-co" id="resxml12">
<para>A method defined on the bean referenced by
<code>correlation-strategy</code>, that implements the
correlation decision algorithm. <emphasis>Optional, with
restrictions (requires <code>correlation-strategy</code> to be
present).</emphasis></para>
</callout>
<callout arearefs="resxml13-co" id="resxml13">
<para>A SpEL expression representing the correlation strategy.
Example: <code>"headers['foo']"</code>. Only one of <code>correlation-strategy</code>
or <code>correlation-strategy-expression</code> is allowed.</para>
</callout>
<callout arearefs="resxml14-co" id="resxml14">
<para>A reference to a bean that implements the release strategy.
The bean can be an implementation of the <interfacename>ReleaseStrategy</interfacename> interface
or a POJO. In the latter case the release-strategy-method
attribute must be defined as well. <emphasis>Optional (by default, the
aggregator will use the <code>MessageHeaders.SEQUENCE_SIZE</code> header attribute)</emphasis>.</para>
</callout>
<callout arearefs="resxml15-co" id="resxml15">
<para>A method defined on the bean referenced by
<code>release-strategy</code>, that implements the
completion decision algorithm. <emphasis>Optional, with
restrictions (requires <code>release-strategy</code> to be
present).</emphasis></para>
</callout>
<callout arearefs="resxml16-co" id="resxml16">
<para>A SpEL expression representing the release strategy; the root object for the
expression is a <code>Collection</code> of <code>Message</code>s.
Example: <code>"size() == 5"</code>. Only one of
<code>release-strategy</code>
or <code>release-strategy-expression</code> is allowed.</para>
</callout>
</calloutlist></para>
<note>