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

INT-2863 Add Release Strategy to Resequencer

Browse Source 
- allow the configuration of release strategies on resequencers
-- adjust the xsd to allow the definition of a release strategy
-- move release strategy parsing to AbstractCorrelatingMessageHandlerParser
-- add unit tests

For reference see: https://jira.springsource.org/browse/INT-2863

INT-2863 Polishing

Fix white space; add @author; update copyright.

INT-2863 Resequencer/Aggregator Docs

Add docs for correlation and release strategis to Resequencer.

Add *.expression docs to Aggregator.
This commit is contained in:
Stefan Ferstl
2012-12-18 23:51:43 +01:00
committed by Gary Russell
parent 3c59d377f4
commit f9aea7d5f5
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>