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

adds support for an <aggregator> element in the integration namespace (INT-95). Currently supports only references to beans implementing the Aggregator interface. The 'method' attribute can be defined, but it is not currently in use.

Browse Source
This commit is contained in:
Marius Bogoevici
2008-02-27 19:15:06 +00:00
parent 559d858e3e
commit 44cb3e17a1
8 changed files with 412 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java Normal file
View File

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

2
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java
View File

@@ -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<String, Class<? extends BeanDefinitionParser>> parserMappings = this.loadAdapterParserMappings();
try {
for (Map.Entry<String, Class<? extends BeanDefinitionParser>> entry : parserMappings.entrySet()) {

42
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java
View File

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

22
spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -209,5 +209,27 @@
<xsd:attribute name="queue-capacity" type="xsd:int"/>
<xsd:attribute name="keep-alive" type="xsd:int"/>
</xsd:complexType>
<xsd:element name="aggregator">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation> Defines an aggregating message handler
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:attribute name="ref" type="xsd:string" use="required"/>
<xsd:attribute name="method" type="xsd:string" use="optional"/>
<xsd:attribute name="completion-strategy" type="xsd:string" use="optional"/>
<xsd:attribute name="default-reply-channel" type="xsd:string" use="optional"/>
<xsd:attribute name="discard-channel" type="xsd:string" use="optional"/>
<xsd:attribute name="send-timeout" type="xsd:long" use="optional"/>
<xsd:attribute name="send-partial-result-on-timeout" type="xsd:boolean" use="optional"/>
<xsd:attribute name="tracked-correlation-id-capacity" type="xsd:int" use="optional"/>
<xsd:attribute name="reaper-interval" type="xsd:long" use="optional"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
</xsd:schema>

123
spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java Normal file
View File

@@ -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<Message<?>> outboundMessages = new ArrayList<Message<?>>();
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);
}
}

55
spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregator.java Normal file
View File

@@ -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<Object, Message<?>> aggregatedMessages = new ConcurrentHashMap<Object, Message<?>>();
public Message<?> aggregate(List<Message<?>> messages) {
List<Message<?>> sortableList = new ArrayList<Message<?>>(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<Object, Message<?>> getAggregatedMessages() {
return aggregatedMessages;
}
}

33
spring-integration-core/src/test/java/org/springframework/integration/config/TestCompletionStrategy.java Normal file
View File

@@ -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<Message<?>> messages) {
throw new UnsupportedOperationException("This is not intended to be implemented, but to verify injection into an <aggregator>");
}
}

26
spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml Normal file
View File

@@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<aggregator id="aggregatorWithReference" ref="aggregatorBean"/>
<aggregator id="completelyDefinedAggregator"
ref="aggregatorBean"
completion-strategy="completionStrategy"
default-reply-channel="replyChannel"
discard-channel="discardChannel"
send-timeout="86420000"
send-partial-result-on-timeout="true"
reaper-interval="135"
tracked-correlation-id-capacity="99"/>
<channel id="replyChannel"/>
<channel id="discardChannel"/>
<beans:bean id="aggregatorBean" class="org.springframework.integration.config.TestAggregator"/>
<beans:bean id="completionStrategy" class="org.springframework.integration.config.TestCompletionStrategy"/>
</beans:beans>