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 new file mode 100644 index 0000000000..d60cd936ec --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java @@ -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(); + } +} 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 9766c1e2be..77101e09b1 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 @@ -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(); - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java index 0a2d06dcf1..d0175b0328 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java @@ -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); - } - } - } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd index a917fdb26a..94a59e8a1c 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd @@ -2704,6 +2704,20 @@ endpoint itself is a Polling Consumer for a channel with a queue. + + + + + + + + + A reference to a bean that implements the release strategy. + The bean can be an implementation of the + ReleaseStrategy interface or a POJO + + + @@ -2729,20 +2743,6 @@ endpoint itself is a Polling Consumer for a channel with a queue. - - - - - - - - - A reference to a bean that implements the release strategy. - The bean can be an implementation of the - ReleaseStrategy interface or a POJO - - - Will store messages after their release. Mainly used for monitoring purposes. Default is 'true'