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

INT-1227: add EL to resequencer and fix schema

Browse Source
This commit is contained in:
David Syer
2010-08-25 07:21:20 +00:00
parent d67cbd2984
commit 4ed390526c
6 changed files with 701 additions and 470 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -13,6 +13,7 @@
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;
@@ -28,9 +29,11 @@ import org.w3c.dom.Element;
*/
public class ResequencerParser 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_ATTRIBUTE = "correlation-strategy";
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";
@@ -40,12 +43,14 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
private static final String MESSAGE_STORE_ATTRIBUTE = "message-store";
private static final String COMPARATOR_ATTRIBUTE = "comparator";
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";
@Override
@@ -57,7 +62,8 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.ResequencingMessageGroupProcessor");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(processorBuilder, element, COMPARATOR_ATTRIBUTE);
// Comparator
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, COMPARATOR_REF_ATTRIBUTE);
String processorRef = BeanDefinitionReaderUtils.registerWithGeneratedName(processorBuilder.getBeanDefinition(),
parserContext.getRegistry());
@@ -69,14 +75,8 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".store.SimpleMessageStore").getBeanDefinition());
String correlationStrategyRef = getCorrelationStrategyRef(element, parserContext);
if (correlationStrategyRef != null) {
builder.addConstructorArgReference(correlationStrategyRef);
} else {
// Correlation strategy
builder.addConstructorArgValue(null);
}
// Correlation strategy
builder.addConstructorArgValue(getCorrelationStrategy(element, parserContext));
// Release strategy
builder.addConstructorArgValue(getReleaseStrategy(element, parserContext));
@@ -88,27 +88,16 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
return builder;
}
private Object getReleaseStrategy(Element element, ParserContext parserContext) {
String releaseStrategyRef = getReleasenStrategyRef(element, parserContext);
if (releaseStrategyRef == null) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.SequenceSizeReleaseStrategy");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE);
return builder.getBeanDefinition();
}
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 new RuntimeBeanReference(releaseStrategyRef);
}
private String getCorrelationStrategyRef(Element element, ParserContext parserContext) {
String ref = element.getAttribute(CORRELATION_STRATEGY_ATTRIBUTE);
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
@@ -118,18 +107,42 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
"java.lang.String");
String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder
.getBeanDefinition(), parserContext.getRegistry());
return adapterBeanName;
} else {
return ref;
return new RuntimeBeanReference(adapterBeanName);
}
else {
return new RuntimeBeanReference(ref);
}
}
return null;
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 String getReleasenStrategyRef(Element element, ParserContext parserContext) {
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
@@ -139,12 +152,30 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
"java.lang.String");
String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder
.getBeanDefinition(), parserContext.getRegistry());
return adapterBeanName;
} else {
return ref;
return new RuntimeBeanReference(adapterBeanName);
}
else {
return new RuntimeBeanReference(ref);
}
}
return null;
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);
}
}
}

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

File diff suppressed because it is too large Load Diff

19
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorExpressionIntegrationTests-context.xml Normal file
View File

@@ -0,0 +1,19 @@
<?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"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="input"/>
<aggregator expression="new java.util.ArrayList(#root).![payload].toString()" input-channel="input"
output-channel="output"/>
<channel id="output">
<queue capacity="5" />
</channel>
</beans:beans>

70
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorExpressionIntegrationTests.java Normal file
View File

@@ -0,0 +1,70 @@
/*
* 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.junit.Assert.assertEquals;
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.MessageHeaders;
import org.springframework.integration.core.GenericMessage;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Iwein Fuld
* @author Alex Peters
* @author Oleg Zhurakousky
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class AggregatorExpressionIntegrationTests {
@Autowired
@Qualifier("input")
private MessageChannel input;
@Autowired
@Qualifier("output")
private PollableChannel output;
@Test//(timeout=5000)
public void testVanillaAggregation() throws Exception {
for (int i = 0; i < 5; i++) {
Map<String, Object> headers = stubHeaders(i, 5, 1);
input.send(new GenericMessage<Integer>(i, headers));
}
assertEquals("[0, 1, 2, 3, 4]", output.receive().getPayload());
}
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);
headers.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize);
headers.put(MessageHeaders.CORRELATION_ID, correllationId);
return headers;
}
}

16
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerExpressionIntegrationTests-context.xml Normal file
View File

@@ -0,0 +1,16 @@
<?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" xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="input" />
<resequencer correlation-strategy-expression="headers['foo']" release-strategy-expression="size()>2" input-channel="input" output-channel="output" />
<channel id="output">
<queue capacity="5" />
</channel>
</beans:beans>

78
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerExpressionIntegrationTests.java Normal file
View File

@@ -0,0 +1,78 @@
/*
* 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.junit.Assert.assertEquals;
import java.util.ArrayList;
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.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.core.GenericMessage;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Iwein Fuld
* @author Alex Peters
* @author Oleg Zhurakousky
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class ResequencerExpressionIntegrationTests {
@Autowired
@Qualifier("input")
private MessageChannel input;
@Autowired
@Qualifier("output")
private PollableChannel output;
@Test//(timeout=5000)
public void testVanillaAggregation() throws Exception {
List<Message<?>> messages = new ArrayList<Message<?>>();
for (int i = 0; i < 5; i++) {
Map<String, Object> headers = stubHeaders(i, 5, 1);
messages.add(new GenericMessage<Integer>(i, headers));
}
input.send(messages.get(2));
input.send(messages.get(1));
input.send(messages.get(0));
assertEquals(0, output.receive().getPayload());
assertEquals(1, output.receive().getPayload());
assertEquals(2, output.receive().getPayload());
}
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);
headers.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize);
headers.put("foo", correllationId);
return headers;
}
}