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

IN PROGRESS - issue INT-642: Add DefaultAggregator

Browse Source 
http://jira.springframework.org/browse/INT-642

Polished Alex' patch
This commit is contained in:
Iwein Fuld
2009-05-20 14:56:36 +00:00
parent 2b93373323
commit 399eee83b1
7 changed files with 297 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultMessageAggregator.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2002-2008 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.aggregator;
import java.util.ArrayList;
import java.util.List;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
/**
* The Default Message Aggregator implementation that combines a group of
* messages into a single message containing a {@link List} of all payloads. The
* elements of the List are in order of their receiving. Any MessageHeader value
* is ignored except the <code>correlationId</code>.
*
* <p>
* n The default strategy for determining whether a group is complete is based
* on the '<code>sequenceSize</code>' property of the header. Alternatively, a
* custom implementation of the {@link CompletionStrategy} may be provided.
* </p>
* <p>
* All considerations regarding <code>timeout</code> and grouping by
* <code>correlationId</code> from {@link AbstractMessageBarrierHandler} apply
* here as well.
* </p>
*
* @author Alex Peters
*
*/
public class DefaultMessageAggregator extends AbstractMessageAggregator {
/**
* {@inheritDoc}
*/
@Override
protected Message<?> aggregateMessages(List<Message<?>> messages) {
List<Object> payloads = new ArrayList<Object>(messages.size());
for (Message<?> message : messages) {
payloads.add(message.getPayload());
}
return MessageBuilder.withPayload(payloads).build();
}
}

80
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
View File

@@ -20,7 +20,6 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
@@ -36,9 +35,9 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
private static final String COMPLETION_STRATEGY_METHOD_ATTRIBUTE = "completion-strategy-method";
private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy";
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_METHOD_ATTRIBUTE = "correlation-strategy-method";
private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
@@ -54,57 +53,74 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
private static final String COMPLETION_STRATEGY_PROPERTY = "completionStrategy";
private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy";
private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy";
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.MethodInvokingAggregator");
BeanDefinitionBuilder builder;
String ref = element.getAttribute(REF_ATTRIBUTE);
if (!StringUtils.hasText(ref)) {
parserContext.getReaderContext().error("The '" + REF_ATTRIBUTE + "' attribute is required.", element);
if (StringUtils.hasText(ref)) {
builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.MethodInvokingAggregator");
builder.addConstructorArgReference(ref);
if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
String method = element.getAttribute(METHOD_ATTRIBUTE);
builder.addConstructorArgValue(method);
}
}
builder.addConstructorArgReference(ref);
if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
String method = element.getAttribute(METHOD_ATTRIBUTE);
builder.addConstructorArgValue(method);
else {
builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.DefaultMessageAggregator");
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REAPER_INTERVAL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
DISCARD_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
SEND_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
REAPER_INTERVAL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE);
this.injectPropertyWithBean(COMPLETION_STRATEGY_REF_ATTRIBUTE, COMPLETION_STRATEGY_METHOD_ATTRIBUTE, COMPLETION_STRATEGY_PROPERTY,
"CompletionStrategyAdapter", element, builder, parserContext);
this.injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY,
"CorrelationStrategyAdapter", element, builder, parserContext);
this.injectPropertyWithBean(COMPLETION_STRATEGY_REF_ATTRIBUTE,
COMPLETION_STRATEGY_METHOD_ATTRIBUTE, COMPLETION_STRATEGY_PROPERTY,
"CompletionStrategyAdapter", element, builder, parserContext);
this.injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE,
CORRELATION_STRATEGY_METHOD_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY,
"CorrelationStrategyAdapter", element, builder, parserContext);
return builder;
}
private void injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute, String beanProperty, String adapterClass,
Element element, BeanDefinitionBuilder builder, ParserContext parserContext) {
final String beanRef = element.getAttribute(beanRefAttribute);
final String beanMethod = element.getAttribute(methodRefAttribute);
private void injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute,
String beanProperty, String adapterClass, Element element,
BeanDefinitionBuilder builder, ParserContext parserContext) {
final String beanRef = element.getAttribute(beanRefAttribute);
final String beanMethod = element.getAttribute(methodRefAttribute);
if (StringUtils.hasText(beanRef)) {
if (StringUtils.hasText(beanMethod)) {
String adapterBeanName = this.createAdapter(beanRef, beanMethod, adapterClass, parserContext);
String adapterBeanName = this.createAdapter(beanRef, beanMethod, adapterClass,
parserContext);
builder.addPropertyReference(beanProperty, adapterBeanName);
}
else {
builder.addPropertyReference(beanProperty, beanRef);
}
}
}
}
private String createAdapter(String ref, String method, String unqualifiedClassName, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator." + unqualifiedClassName);
private String createAdapter(String ref, String method, String unqualifiedClassName,
ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator."
+ unqualifiedClassName);
builder.addConstructorArgReference(ref);
builder.addConstructorArgValue(method);
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(),
parserContext.getRegistry());
}
}

59
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/DefaultMessageAggregatorTests.java Normal file
View File

@@ -0,0 +1,59 @@
/*
* Copyright 2002-2008 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.aggregator;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.awt.Button;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.core.Message;
/**
* @author Alex Peters
*
*/
public class DefaultMessageAggregatorTests {
DefaultMessageAggregator aggregator;
@Before
public void setUp() {
aggregator = new DefaultMessageAggregator();
}
@SuppressWarnings("unchecked")
@Test
public void aggregateMessages_withMultiplePayloads_allAsListInResultMsg() {
List<Serializable> anyPayloads = Arrays.asList("foo", "bar", 123L, new Button());
List<Message<?>> messageGroup = new ArrayList<Message<?>>(anyPayloads.size());
for (Serializable payload : anyPayloads) {
Message<Serializable> mock = mock(Message.class);
when(mock.getPayload()).thenReturn(payload);
messageGroup.add(mock);
}
Message<?> result = aggregator.aggregateMessages(messageGroup);
assertThat((List<Serializable>) result.getPayload(), is(anyPayloads));
}
}

17
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml
View File

@@ -6,22 +6,21 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<annotation-config/>
<annotation-config />
<channel id="input">
<queue capacity="5"/>
<queue capacity="5" />
</channel>
<aggregator ref="summer" method="sum" input-channel="input" output-channel="output">
<aggregator ref="summer" method="sum" input-channel="input"
output-channel="output">
<poller task-executor="executor" max-messages-per-poll="5">
<interval-trigger interval="20"/>
<interval-trigger interval="20" />
</poller>
</aggregator>
<beans:bean id="executor"
class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<beans:property name="corePoolSize" value="5" />
</beans:bean>
<thread-pool-task-executor id="executor"
core-size="5" />
<channel id="output">
<queue capacity="5" />

18
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java
View File

@@ -16,7 +16,9 @@
package org.springframework.integration.aggregator.integration;
import static org.junit.Assert.*;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import java.util.HashMap;
import java.util.List;
@@ -24,9 +26,10 @@ import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.aggregator.AbstractMessageAggregator;
import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
@@ -36,6 +39,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Iwein Fuld
* @author Alex Peters
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@@ -49,10 +53,12 @@ public class ConcurrentAggregatorIntegrationTests {
@Qualifier("output")
private PollableChannel output;
@Autowired
AbstractMessageAggregator aggregator;
@Test
public void configOk() throws Exception {
// nothing to assert
assertThat(aggregator, is(MethodInvokingAggregator.class));
}
@Test(timeout = 1000)
@@ -61,11 +67,10 @@ public class ConcurrentAggregatorIntegrationTests {
Map<String, Object> headers = stubHeaders(i, 5, 1);
input.send(new GenericMessage<Integer>(i, headers));
}
assertEquals(0+1+2+3+4, output.receive().getPayload());
assertEquals(0 + 1 + 2 + 3 + 4, output.receive().getPayload());
}
//configured in context associated with this test
// configured in context associated with this test
public static class SummingAggregator {
public Integer sum(List<Integer> numbers) {
int result = 0;
@@ -76,7 +81,6 @@ public class ConcurrentAggregatorIntegrationTests {
}
}
private Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correllationId) {
Map<String, Object> headers = new HashMap<String, Object>();
headers.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber);

29
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests-context.xml Normal file
View File

@@ -0,0 +1,29 @@
<?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-1.0.xsd">
<annotation-config />
<channel id="input">
<queue capacity="5" />
</channel>
<aggregator input-channel="input" output-channel="output">
<poller task-executor="executor" max-messages-per-poll="5">
<interval-trigger interval="20" />
</poller>
</aggregator>
<thread-pool-task-executor id="executor"
core-size="5" />
<channel id="output">
<queue capacity="5" />
</channel>
</beans:beans>

84
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java Normal file
View File

@@ -0,0 +1,84 @@
/*
* Copyright 2002-2008 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.aggregator.integration;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.aggregator.AbstractMessageAggregator;
import org.springframework.integration.aggregator.DefaultMessageAggregator;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Alex Peters
* @author Iwein Fuld
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class DefaultMessageAggregatorIntegrationTests {
@Autowired
@Qualifier("input")
MessageChannel input;
@Autowired
@Qualifier("output")
PollableChannel output;
@Autowired
AbstractMessageAggregator aggregator;
@Test
public void configOk() throws Exception {
assertThat(aggregator, is(DefaultMessageAggregator.class));
}
@SuppressWarnings("unchecked")
@Test(timeout = 1000)
public void aggregate() throws Exception {
for (int i = 0; i < 5; i++) {
Map<String, Object> headers = stubHeaders(i, 5, 1);
input.send(new GenericMessage<Integer>(i, headers));
}
Object payload = output.receive().getPayload();
assertThat(payload, is(List.class));
assertThat((List) payload, is(Arrays.asList(0, 1, 2, 3, 4)));
}
Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correllationId) {
Map<String, Object> headers = new HashMap<String, Object>();
headers.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber);
headers.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize);
headers.put(MessageHeaders.CORRELATION_ID, correllationId);
return headers;
}
}