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

Merge pull request #136 from olegz/INT-1296

Browse Source 
refactored Aggregator/Resequencer schemas and parsers
This commit is contained in:
Mark Fisher
2011-10-19 10:23:35 -04:00
parent 2884526543 dd3959a56f
commit 1e5f24fb09
4 changed files with 130 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,99 @@
/*
* Copyright 2002-2011 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*/
package org.springframework.integration.config.xml;
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
* @since 2.1
*
*/
public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractConsumerEndpointParser {
private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy";
private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method";
private static final String CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE = "correlation-strategy-expression";
private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy";
private static final String MESSAGE_STORE_ATTRIBUTE = "message-store";
private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
private static final String SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE = "send-partial-result-on-expiry";
private static final String KEEP_RELEASED_MESSAGES = "keep-released-messages";
protected void doParse(BeanDefinitionBuilder builder, Element element, BeanMetadataElement processor, ParserContext parserContext){
this.injectPropertyWithAdapter(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE,
CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY, "CorrelationStrategy",
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);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, KEEP_RELEASED_MESSAGES);
}
protected void injectPropertyWithAdapter(String beanRefAttribute, String methodRefAttribute,
String expressionAttribute, String beanProperty, String adapterClass, Element element,
BeanDefinitionBuilder builder, BeanMetadataElement processor, ParserContext parserContext) {
final String beanRef = element.getAttribute(beanRefAttribute);
final String beanMethod = element.getAttribute(methodRefAttribute);
final String expression = element.getAttribute(expressionAttribute);
BeanMetadataElement adapter = null;
if (StringUtils.hasText(beanRef)) {
adapter = this.createAdapter(new RuntimeBeanReference(beanRef), beanMethod, adapterClass, parserContext);
}
else if (processor != null) {
adapter = this.createAdapter(processor, beanMethod, adapterClass, parserContext);
}
else if (StringUtils.hasText(expression)) {
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.ExpressionEvaluating"
+ adapterClass);
adapterBuilder.addConstructorArgValue(expression);
adapter = adapterBuilder.getBeanDefinition();
}
else {
adapter = this.createAdapter(null, beanMethod, adapterClass, parserContext);
}
builder.addPropertyValue(beanProperty, adapter);
}
private BeanMetadataElement createAdapter(BeanMetadataElement ref, String method, String unqualifiedClassName,
ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".config." + unqualifiedClassName
+ "FactoryBean");
builder.addConstructorArgValue(ref);
if (StringUtils.hasText(method)) {
builder.addConstructorArgValue(method);
}
return builder.getBeanDefinition();
}
}

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

@@ -22,6 +22,8 @@ import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor;
import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
@@ -35,38 +37,18 @@ import org.w3c.dom.Element;
* @author Oleg Zhurakousky
* @author Dave Syer
*/
public class AggregatorParser extends AbstractConsumerEndpointParser {
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 CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy";
private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method";
private static final String CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE = "correlation-strategy-expression";
private static final String MESSAGE_STORE_ATTRIBUTE = "message-store";
private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
private static final String SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE = "send-partial-result-on-expiry";
private static final String RELEASE_STRATEGY_PROPERTY = "releaseStrategy";
private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy";
private static final String EXPIRE_GROUPS_UPON_COMPLETION = "expire-groups-upon-completion";
private static final String KEEP_RELEASED_MESSAGES = "keep-released-messages";
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanComponentDefinition innerHandlerDefinition = IntegrationNamespaceUtils.parseInnerHandlerDefinition(element,
@@ -92,14 +74,12 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
else {
if (StringUtils.hasText(element.getAttribute(EXPRESSION_ATTRIBUTE))) {
String expression = element.getAttribute(EXPRESSION_ATTRIBUTE);
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.ExpressionEvaluatingMessageGroupProcessor");
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionEvaluatingMessageGroupProcessor.class);
adapterBuilder.addConstructorArgValue(expression);
builder.addConstructorArgValue(adapterBuilder.getBeanDefinition());
}
else {
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor")
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(DefaultAggregatingMessageGroupProcessor.class)
.getBeanDefinition());
}
}
@@ -109,60 +89,15 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
processorBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method,
"java.lang.String");
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_STORE_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
this.doParse(builder, element, processor, parserContext);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_COMPLETION);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, KEEP_RELEASED_MESSAGES);
this.injectPropertyWithAdapter(RELEASE_STRATEGY_REF_ATTRIBUTE, RELEASE_STRATEGY_METHOD_ATTRIBUTE,
RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE, RELEASE_STRATEGY_PROPERTY, "ReleaseStrategy", element, builder,
processor, parserContext);
this.injectPropertyWithAdapter(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE,
CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY, "CorrelationStrategy",
element, builder, processor, parserContext);
return builder;
}
private void injectPropertyWithAdapter(String beanRefAttribute, String methodRefAttribute,
String expressionAttribute, String beanProperty, String adapterClass, Element element,
BeanDefinitionBuilder builder, BeanMetadataElement processor, ParserContext parserContext) {
final String beanRef = element.getAttribute(beanRefAttribute);
final String beanMethod = element.getAttribute(methodRefAttribute);
final String expression = element.getAttribute(expressionAttribute);
BeanMetadataElement adapter = null;
if (StringUtils.hasText(beanRef)) {
adapter = this.createAdapter(new RuntimeBeanReference(beanRef), beanMethod, adapterClass, parserContext);
}
else if (processor != null) {
adapter = this.createAdapter(processor, beanMethod, adapterClass, parserContext);
}
else if (StringUtils.hasText(expression)) {
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.ExpressionEvaluating"
+ adapterClass);
adapterBuilder.addConstructorArgValue(expression);
adapter = adapterBuilder.getBeanDefinition();
}
else {
adapter = this.createAdapter(null, beanMethod, adapterClass, parserContext);
}
builder.addPropertyValue(beanProperty, adapter);
}
private BeanMetadataElement createAdapter(BeanMetadataElement ref, String method, String unqualifiedClassName,
ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".config." + unqualifiedClassName
+ "FactoryBean");
builder.addConstructorArgValue(ref);
if (StringUtils.hasText(method)) {
builder.addConstructorArgValue(method);
}
return builder.getBeanDefinition();
}
}

133
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java
View File

@@ -13,14 +13,11 @@
package org.springframework.integration.config.xml;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
import org.springframework.integration.aggregator.ResequencingMessageHandler;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
@@ -31,34 +28,13 @@ import org.w3c.dom.Element;
* @author Iwein Fuld
* @author Oleg Zhurakousky
*/
public class ResequencerParser extends AbstractConsumerEndpointParser {
public class ResequencerParser extends AbstractCorrelatingMessageHandlerParser {
private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy";
private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method";
private static final String CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE = "correlation-strategy-expression";
private static final String SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE = "send-partial-result-on-expiry";
private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
private static final String MESSAGE_STORE_ATTRIBUTE = "message-store";
private static final String COMPARATOR_REF_ATTRIBUTE = "comparator";
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_PARTIAL_SEQUENCES_ATTRIBUTE = "release-partial-sequences";
private static final String KEEP_RELEASED_MESSAGES = "keep-released-messages";
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
@@ -77,110 +53,11 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
// Message store
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".store.SimpleMessageStore").getBeanDefinition());
// Correlation strategy
builder.addConstructorArgValue(getCorrelationStrategy(element, parserContext));
// Release strategy
builder.addConstructorArgValue(getReleaseStrategy(element, parserContext));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_STORE_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE);
this.doParse(builder, element, processorBuilder.getBeanDefinition(), parserContext);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, KEEP_RELEASED_MESSAGES);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
return builder;
}
private BeanMetadataElement getCorrelationStrategy(Element element, ParserContext parserContext) {
String ref = element.getAttribute(CORRELATION_STRATEGY_REF_ATTRIBUTE);
String expression = element.getAttribute(CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE);
String method = element.getAttribute(CORRELATION_STRATEGY_METHOD_ATTRIBUTE);
if (StringUtils.hasText(ref)) {
if (StringUtils.hasText(expression)) {
parserContext.getReaderContext().error(
"Only one of correlation strategy expression and bean reference must be specified", element);
return null;
}
if (StringUtils.hasText(method)) {
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.MethodInvokingCorrelationStrategy");
adapterBuilder.addConstructorArgReference(ref);
adapterBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method,
"java.lang.String");
String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder
.getBeanDefinition(), parserContext.getRegistry());
return new RuntimeBeanReference(adapterBeanName);
}
else {
return new RuntimeBeanReference(ref);
}
}
else {
if (!StringUtils.hasText(expression)) {
return null;
}
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.ExpressionEvaluatingCorrelationStrategy");
adapterBuilder.addConstructorArgValue(expression);
String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder
.getBeanDefinition(), parserContext.getRegistry());
return new RuntimeBeanReference(adapterBeanName);
}
}
private BeanMetadataElement getReleaseStrategy(Element element, ParserContext parserContext) {
String ref = element.getAttribute(RELEASE_STRATEGY_REF_ATTRIBUTE);
String method = element.getAttribute(RELEASE_STRATEGY_METHOD_ATTRIBUTE);
String expression = element.getAttribute(RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE);
if (StringUtils.hasText(ref)) {
if (StringUtils.hasText(expression)) {
parserContext.getReaderContext().error(
"Only one of release strategy expression and bean reference must be specified", element);
return null;
}
if (StringUtils.hasText(element.getAttribute(RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE))) {
parserContext.getReaderContext().error(
"Only one of " + RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE + " and " + RELEASE_STRATEGY_REF_ATTRIBUTE
+ " can be specified at once", element);
return null;
}
if (StringUtils.hasText(method)) {
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.MethodInvokingReleaseStrategy");
adapterBuilder.addConstructorArgReference(ref);
adapterBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method,
"java.lang.String");
String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder
.getBeanDefinition(), parserContext.getRegistry());
return new RuntimeBeanReference(adapterBeanName);
}
else {
return new RuntimeBeanReference(ref);
}
}
else {
if (!StringUtils.hasText(expression)) {
return null;
}
if (StringUtils.hasText(element.getAttribute(RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE))) {
parserContext.getReaderContext().error(
"Only one of " + RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE + " and "
+ RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE + " can be specified at once", element);
return null;
}
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.ExpressionEvaluatingReleaseStrategy");
adapterBuilder.addConstructorArgValue(expression);
String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder
.getBeanDefinition(), parserContext.getRegistry());
return new RuntimeBeanReference(adapterBeanName);
}
}
}

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

@@ -2704,6 +2704,20 @@ endpoint itself is a Polling Consumer for a channel with a queue.
</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>
@@ -2729,20 +2743,6 @@ endpoint itself is a Polling Consumer for a channel with a queue.
<xsd:complexType name="correlating-message-handler-type">
<xsd:complexContent>
<xsd:extension base="innerEndpointDefinitionAware">
<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="keep-released-messages" type="xsd:string">
<xsd:annotation>
<xsd:documentation>Will store messages after their release. Mainly used for monitoring purposes. Default is 'true'</xsd:documentation>