diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java new file mode 100644 index 0000000000..ce2d317cd2 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java @@ -0,0 +1,110 @@ +/* + * Copyright 2002-2007 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; + +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.parsing.BeanComponentDefinition; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.beans.factory.xml.BeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.MessagingConfigurationException; +import org.springframework.integration.router.AggregatingMessageHandler; +import org.springframework.util.StringUtils; +import org.w3c.dom.Element; + +/** + * Parser for the aggregator element of the integration namespace. + * Registers the annotation-driven post-processors. + * + * @author Marius Bogoevici + */ +public class AggregatorParser implements BeanDefinitionParser { + + public static final String ID_ATTRIBUTE = "id"; + + public static final String REF_ATTRIBUTE = "ref"; + + public static final String METHOD_ATTRIBUTE = "method"; + + public static final String COMPLETION_STRATEGY_ATTRIBUTE = "completion-strategy"; + + public static final String DEFAULT_REPLY_CHANNEL_ATTRIBUTE = "default-reply-channel"; + + public static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel"; + + public static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout"; + + public static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE = "send-partial-result-on-timeout"; + + public static final String REAPER_INTERVAL_ATTRIBUTE = "reaper-interval"; + + public static final String TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE = "tracked-correlation-id-capacity"; + + private static final String COMPLETION_STRATEGY_PROPERTY = "completionStrategy"; + + private static final String DEFAULT_REPLY_CHANNEL_PROPERTY = "defaultReplyChannel"; + + private static final String DISCARD_CHANNEL_PROPERTY = "discardChannel"; + + private static final String SEND_TIMEOUT_PROPERTY = "sendTimeout"; + + private static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_PROPERTY = "sendPartialResultOnTimeout"; + + private static final String REAPER_INTERVAL_PROPERTY = "reaperInterval"; + + public static final String TRACKED_CORRELATION_ID_CAPACITY_PROPERTY = "trackedCorrelationIdCapacity"; + + public BeanDefinition parse(Element element, ParserContext parserContext) { + return parseAggregatorElement(element, parserContext, true); + } + + public static BeanDefinition parseAggregatorElement(Element element, ParserContext parserContext, boolean topLevel) { + final RootBeanDefinition aggregatorDef = new RootBeanDefinition(AggregatingMessageHandler.class); + aggregatorDef.setSource(parserContext.extractSource(element)); + final String id = element.getAttribute(ID_ATTRIBUTE); + final String ref = element.getAttribute(REF_ATTRIBUTE); + if (!StringUtils.hasText(ref)) { + throw new MessagingConfigurationException("The 'ref' attribute must be present"); + } + if (!topLevel && StringUtils.hasText(id)) { + parserContext.getReaderContext().error( + "The 'id' attribute is only supported for top-level elements.", + parserContext.extractSource(element)); + } + aggregatorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(ref)); + IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, COMPLETION_STRATEGY_PROPERTY, + element, COMPLETION_STRATEGY_ATTRIBUTE); + IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, DEFAULT_REPLY_CHANNEL_PROPERTY, + element, DEFAULT_REPLY_CHANNEL_ATTRIBUTE); + IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, DISCARD_CHANNEL_PROPERTY, element, + DISCARD_CHANNEL_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(aggregatorDef, SEND_TIMEOUT_PROPERTY, element, + SEND_TIMEOUT_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(aggregatorDef, SEND_PARTIAL_RESULT_ON_TIMEOUT_PROPERTY, + element, SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(aggregatorDef, REAPER_INTERVAL_PROPERTY, element, + REAPER_INTERVAL_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(aggregatorDef, TRACKED_CORRELATION_ID_CAPACITY_PROPERTY, + element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE); + String beanName = StringUtils.hasText(id) ? id : parserContext.getReaderContext().generateBeanName( + aggregatorDef); + parserContext.registerBeanComponent(new BeanComponentDefinition(aggregatorDef, beanName)); + return aggregatorDef; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java index aca537f8ec..49c89610f6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java @@ -34,6 +34,7 @@ import org.springframework.util.ClassUtils; * Namespace handler for the integration namespace. * * @author Mark Fisher + * @author Marius Bogoevici */ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport { @@ -52,6 +53,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport { registerBeanDefinitionParser("endpoint", new EndpointParser()); registerBeanDefinitionParser("handler", new HandlerParser()); registerBeanDefinitionParser("handler-chain", new HandlerParser()); + registerBeanDefinitionParser("aggregator", new AggregatorParser()); Map> parserMappings = this.loadAdapterParserMappings(); try { for (Map.Entry> entry : parserMappings.entrySet()) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java index 78886327fb..38ca60f781 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java @@ -18,6 +18,8 @@ package org.springframework.integration.config; import org.w3c.dom.Element; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.util.StringUtils; @@ -25,6 +27,7 @@ import org.springframework.util.StringUtils; * Shared utility methods for integration namespace parsers. * * @author Mark Fisher + * @author Marius Bogoevici */ public abstract class IntegrationNamespaceUtils { @@ -36,7 +39,6 @@ public abstract class IntegrationNamespaceUtils { private static final String KEEP_ALIVE_ATTRIBUTE = "keep-alive"; - public static ConcurrencyPolicy parseConcurrencyPolicy(Element element) { ConcurrencyPolicy policy = new ConcurrencyPolicy(); String coreSize = element.getAttribute(CORE_SIZE_ATTRIBUTE); @@ -58,4 +60,42 @@ public abstract class IntegrationNamespaceUtils { return policy; } + /** + * Populates the property identified by propertyName on the bean definition + * to the value of the attribute specified by attributeName, if that + * attribute is defined in the element + * + * @param beanDefinition - the bean definition to be configured + * @param propertyName - the name of the bean property to be set + * @param element - the XML element where the attribute should be defined + * @param attributeName - the name of the attribute whose value will be set + * on the property + */ + public static void setValueIfAttributeDefined(RootBeanDefinition beanDefinition, String propertyName, + Element element, String attributeName) { + final String attributeValue = element.getAttribute(attributeName); + if (StringUtils.hasText(attributeValue)) { + beanDefinition.getPropertyValues().addPropertyValue(propertyName, attributeValue); + } + } + + /** + * Populates the property given by propertyName on the given bean definition + * to a reference to a bean identified by the value of the attribute + * specified by attributeName, if that attribute is defined in the element + * + * @param beanDefinition - the bean definition to be configured + * @param propertyName - the name of the bean property to be set + * @param element - the XML element where the attribute should be defined + * @param attributeName - the id of the bean which will be used to populate + * the property + */ + public static void setBeanReferenceIfAttributeDefined(RootBeanDefinition beanDefinition, String propertyName, + Element element, String attributeName) { + final String attributeValue = element.getAttribute(attributeName); + if (StringUtils.hasText(attributeValue)) { + beanDefinition.getPropertyValues().addPropertyValue(propertyName, new RuntimeBeanReference(attributeValue)); + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index f142f7480a..f0c129cafc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -209,5 +209,27 @@ + + + + + Defines an aggregating message handler + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java new file mode 100644 index 0000000000..7733d0ffd0 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -0,0 +1,123 @@ +/* + * Copyright 2002-2007 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; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.springframework.context.ApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; +import org.springframework.integration.router.AggregatingMessageHandler; +import org.springframework.integration.router.Aggregator; +import org.springframework.integration.router.CompletionStrategy; +import org.springframework.util.ReflectionUtils; + +/** + * @author Marius Bogoevici + */ +public class AggregatorParserTests { + + private ApplicationContext context; + + @Before + public void setUp() { + context = new ClassPathXmlApplicationContext("aggregatorParserTests.xml", this.getClass()); + } + + @Test + public void testAggregation() { + AggregatingMessageHandler aggregatingHandler = (AggregatingMessageHandler) context + .getBean("aggregatorWithReference"); + TestAggregator aggregatorBean = (TestAggregator) context.getBean("aggregatorBean"); + List> outboundMessages = new ArrayList>(); + outboundMessages.add(createMessage("123", "id1", 3, 1, null)); + outboundMessages.add(createMessage("789", "id1", 3, 3, null)); + outboundMessages.add(createMessage("456", "id1", 3, 2, null)); + for (Message message : outboundMessages) { + aggregatingHandler.handle(message); + } + Assert.assertEquals("One and only one message must have been aggregated", 1, aggregatorBean + .getAggregatedMessages().size()); + Message aggregatedMessage = aggregatorBean.getAggregatedMessages().get("id1"); + Assert.assertEquals("The aggreggated message payload is not correct", "123456789", aggregatedMessage + .getPayload()); + } + + @Test + public void testPropertyAssignment() throws Exception { + AggregatingMessageHandler completeAggregatingMessageHandler = (AggregatingMessageHandler) context + .getBean("completelyDefinedAggregator"); + TestAggregator testAggregator = (TestAggregator) context.getBean("aggregatorBean"); + CompletionStrategy completionStrategy = (CompletionStrategy) context.getBean("completionStrategy"); + MessageChannel defaultReplyChannel = (MessageChannel) context.getBean("replyChannel"); + MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel"); + + Assert.assertEquals("The AggregatingMessageHandler is not injected with the appropriate Aggregator instance", + testAggregator, getPropertyValue(completeAggregatingMessageHandler, "aggregator", Aggregator.class)); + Assert.assertEquals( + "The AggregatingMessageHandler is not injected with the appropriate CompletionStrategy instance", + completionStrategy, getPropertyValue(completeAggregatingMessageHandler, "completionStrategy", + CompletionStrategy.class)); + Assert.assertEquals("The AggregatingMessageHandler is not injected with the appropriate default reply channel", + defaultReplyChannel, getPropertyValue(completeAggregatingMessageHandler, "defaultReplyChannel", + MessageChannel.class)); + Assert.assertEquals("The AggregatingMessageHandler is not injected with the appropriate discard channel", + discardChannel, getPropertyValue(completeAggregatingMessageHandler, "discardChannel", + MessageChannel.class)); + Assert.assertEquals("The AggregatingMessageHandler is not set with the appropriate timeout value", 86420000l, + getPropertyValue(completeAggregatingMessageHandler, "sendTimeout", long.class)); + Assert.assertEquals( + "The AggregatingMessageHandler is not configured with the appropriate 'send partial results on timeout' flag", + true, getPropertyValue(completeAggregatingMessageHandler, "sendPartialResultOnTimeout", + boolean.class)); + Assert.assertEquals("The AggregatingMessageHandler is not configured with the appropriate reaper interval", + 135l, getPropertyValue(completeAggregatingMessageHandler, "reaperInterval", long.class)); + Assert.assertEquals( + "The AggregatingMessageHandler is not configured with the appropriate tracked correlationId capacity", + 99, getPropertyValue(completeAggregatingMessageHandler, "trackedCorrelationIdCapacity", int.class)); + } + + private static Message createMessage(String payload, Object correlationId, int sequenceSize, int sequenceNumber, + MessageChannel replyChannel) { + StringMessage message = new StringMessage(payload); + message.getHeader().setCorrelationId(correlationId); + message.getHeader().setSequenceSize(sequenceSize); + message.getHeader().setSequenceNumber(sequenceNumber); + message.getHeader().setReturnAddress(replyChannel); + return message; + } + + /** + * Reading private fields through reflection, since they don't have setters + * @param beanUnderTest + * @param fieldName + * @return the value of the field + * @throws Exception + */ + private static Object getPropertyValue(Object beanUnderTest, String fieldName, Class type) throws Exception { + Field field = ReflectionUtils.findField(beanUnderTest.getClass(), fieldName, type); + ReflectionUtils.makeAccessible(field); + return field.get(beanUnderTest); + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregator.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregator.java new file mode 100644 index 0000000000..6caef00fdf --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregator.java @@ -0,0 +1,55 @@ +/* + * Copyright 2002-2007 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; + +import java.util.List; +import java.util.ArrayList; +import java.util.Collections; +import java.util.concurrent.ConcurrentHashMap; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; +import org.springframework.integration.router.Aggregator; +import org.springframework.integration.router.MessageSequenceComparator; + +/** + * @author Marius Bogoevici + */ +public class TestAggregator implements Aggregator { + + ConcurrentHashMap> aggregatedMessages = new ConcurrentHashMap>(); + + public Message aggregate(List> messages) { + List> sortableList = new ArrayList>(messages); + Collections.sort(sortableList, new MessageSequenceComparator()); + StringBuffer buffer = new StringBuffer(); + for (Message message : sortableList) { + buffer.append(message.getPayload().toString()); + } + Message returnedMessage = new StringMessage(buffer.toString()); + aggregatedMessages.put(messages.get(0).getHeader().getCorrelationId(), returnedMessage); + return returnedMessage; + } + + public ConcurrentHashMap> getAggregatedMessages() { + return aggregatedMessages; + } + +} + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestCompletionStrategy.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestCompletionStrategy.java new file mode 100644 index 0000000000..be822dbeb1 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestCompletionStrategy.java @@ -0,0 +1,33 @@ +/* + * Copyright 2002-2007 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; + +import java.util.List; + +import org.springframework.integration.message.Message; +import org.springframework.integration.router.CompletionStrategy; + +/** + * @author Marius Bogoevici + */ +public class TestCompletionStrategy implements CompletionStrategy { + + public boolean isComplete(List> messages) { + throw new UnsupportedOperationException("This is not intended to be implemented, but to verify injection into an "); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml new file mode 100644 index 0000000000..999bc131b2 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + +