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

INT-818: EL-based aggregator

Browse Source 
* adapted some code from patch to new idioms
* added namespace support
* tidied up *Adapter -> MethodInvoking* for consistency
This commit is contained in:
David Syer
2010-07-29 08:56:20 +00:00
parent a5c14782fb
commit d92a1245d3
28 changed files with 728 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractExpressionEvaluatingMessageListProcessor.java Normal file
View File

@@ -0,0 +1,123 @@
/*
* Copyright 2002-2010 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.Collection;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.context.expression.MapAccessor;
import org.springframework.core.convert.ConversionService;
import org.springframework.expression.AccessException;
import org.springframework.expression.BeanResolver;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.ParseException;
import org.springframework.expression.spel.SpelParserConfiguration;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.expression.spel.support.StandardTypeConverter;
import org.springframework.integration.Message;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.transformer.MessageTransformationException;
/**
* A base class for aggregators that evaluates a SpEL expression with the message list as the root object within the
* evaluation context.
*
* @author Dave Syer
* @since 2.0
*/
public class AbstractExpressionEvaluatingMessageListProcessor implements BeanFactoryAware {
private final ExpressionParser parser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
private final Expression expression;
private volatile Class<?> expectedType = null;
private final StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
/**
* Create an {@link ExpressionEvaluatingMessageProcessor} for the given expression String.
*/
public AbstractExpressionEvaluatingMessageListProcessor(String expression) {
try {
this.expression = parser.parseExpression(expression);
this.getEvaluationContext().addPropertyAccessor(new MapAccessor());
}
catch (ParseException e) {
throw new IllegalArgumentException("Failed to parse expression.", e);
}
}
/**
* Set the result type expected from evaluation of the expression.
*/
public void setExpectedType(Class<?> expectedType) {
this.expectedType = expectedType;
}
/**
* Specify a BeanFactory in order to enable resolution via <code>@beanName</code> in the expression.
*/
public void setBeanFactory(final BeanFactory beanFactory) {
if (beanFactory != null) {
this.getEvaluationContext().setBeanResolver(new BeanResolver() {
public Object resolve(EvaluationContext context, String beanName) throws AccessException {
return beanFactory.getBean(beanName);
}
});
}
}
public void setConversionService(ConversionService conversionService) {
if (conversionService != null) {
this.evaluationContext.setTypeConverter(new StandardTypeConverter(conversionService));
}
}
protected StandardEvaluationContext getEvaluationContext() {
return this.evaluationContext;
}
protected Object evaluateExpression(Expression expression, Collection<? extends Message<?>> messages,
Class<?> expectedType) {
try {
return (expectedType != null) ? expression.getValue(this.evaluationContext, messages, expectedType)
: expression.getValue(this.evaluationContext, messages);
}
catch (EvaluationException e) {
Throwable cause = e.getCause();
throw new MessageTransformationException("Expression evaluation failed.", cause == null ? e : cause);
}
catch (Exception e) {
throw new MessageTransformationException("Expression evaluation failed.", e);
}
}
/**
* Processes the Message by evaluating the expression with that Message as the root object. The expression
* evaluation result Object will be returned.
*/
protected Object process(Collection<? extends Message<?>> messages) {
return this.evaluateExpression(this.expression, messages, this.expectedType);
}
}

39
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.java Normal file
View File

@@ -0,0 +1,39 @@
/*
* Copyright 2002-2009 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 org.springframework.integration.Message;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
/**
* {@link CorrelationStrategy} implementation that evaluates an expression.
*
* @author Dave Syer
*/
public class ExpressionEvaluatingCorrelationStrategy implements CorrelationStrategy {
private final ExpressionEvaluatingMessageProcessor processor;
public ExpressionEvaluatingCorrelationStrategy(String expression) {
this.processor = new ExpressionEvaluatingMessageProcessor(expression);
}
public Object getCorrelationKey(Message<?> message) {
return processor.processMessage(message);
}
}

33
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessor.java Normal file
View File

@@ -0,0 +1,33 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.core.MessageBuilder;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.store.MessageGroup;
/**
* A {@link MessageGroupProcessor} implementation that evaluates a SpEL expression. The SpEL context root is the list of
* all Messages in the group. The evaluation result can be any Object and is send as new Message payload to the output
* channel.
*
* @author Alex Peters
* @author Dave Syer
*
*/
public class ExpressionEvaluatingMessageGroupProcessor extends AbstractExpressionEvaluatingMessageListProcessor
implements MessageGroupProcessor {
public ExpressionEvaluatingMessageGroupProcessor(String expression) {
super(expression);
}
/**
* Evaluate the expression provided on the unmarked messages (a collection) in the group, and delegate to the
* {@link MessagingTemplate} to send dowstream.
*/
public void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate, MessageChannel outputChannel) {
Object newPayload = process(group.getUnmarked());
messagingTemplate.send(outputChannel, MessageBuilder.withPayload(newPayload).build());
}
}

41
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* 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 org.springframework.integration.store.MessageGroup;
/**
* A {@link ReleaseStrategy} that evaluates an expression.
*
* @author Dave Syer
*/
public class ExpressionEvaluatingReleaseStrategy extends AbstractExpressionEvaluatingMessageListProcessor implements
ReleaseStrategy {
public ExpressionEvaluatingReleaseStrategy(String expression) {
super(expression);
}
/**
* Evaluate the expression provided on the unmarked messages (a collection) in the group and return the result (must
* be boolean).
*/
public boolean canRelease(MessageGroup messages) {
return ((Boolean) process(messages.getUnmarked())).booleanValue();
}
}

7
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java
View File

@@ -38,7 +38,7 @@ import org.springframework.util.ReflectionUtils;
* @author Iwein Fuld
* @author Dave Syer
*/
public class MessageListMethodAdapter {
public class MessageListMethodAdapter implements MessageListProcessor {
private final DefaultMethodInvoker invoker;
@@ -68,7 +68,10 @@ public class MessageListMethodAdapter {
return method;
}
public final Object executeMethod(Collection<? extends Message<?>> messages) {
/* (non-Javadoc)
* @see org.springframework.integration.aggregator.MessageListProcessor#executeMethod(java.util.Collection)
*/
public final Object process(Collection<? extends Message<?>> messages) {
try {
if (isMethodParameterParameterized(this.method) && isHavingActualTypeArguments(this.method)
&& (isActualTypeRawMessage(this.method) || isActualTypeParameterizedMessage(this.method))) {

2
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java
View File

@@ -38,7 +38,7 @@ import org.springframework.util.ReflectionUtils;
*/
public class MessageListMethodAdapterHelper {
public MessageListMethodAdapter getAdapter(Object candidate, Class<? extends Annotation> annotationType) {
public MessageListProcessor getAdapter(Object candidate, Class<? extends Annotation> annotationType) {
Method method = findAggregatorMethod(candidate, annotationType);
if (method == null) {
return null;

30
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListProcessor.java Normal file
View File

@@ -0,0 +1,30 @@
/*
* Copyright 2002-2010 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.Collection;
import org.springframework.integration.Message;
/**
* @author dsyer
*
*/
public interface MessageListProcessor {
Object process(Collection<? extends Message<?>> messages);
}

6
spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java → spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java
View File

@@ -28,15 +28,15 @@ import org.springframework.util.Assert;
* @author Marius Bogoevici
* @author Dave Syer
*/
public class CorrelationStrategyAdapter implements CorrelationStrategy {
public class MethodInvokingCorrelationStrategy implements CorrelationStrategy {
private final MethodInvokingMessageProcessor processor;
public CorrelationStrategyAdapter(Object object, String methodName) {
public MethodInvokingCorrelationStrategy(Object object, String methodName) {
this.processor = new MethodInvokingMessageProcessor(object, methodName, true);
}
public CorrelationStrategyAdapter(Object object, Method method) {
public MethodInvokingCorrelationStrategy(Object object, Method method) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(method, "'method' must not be null");
Assert.isTrue(!Void.TYPE.equals(method.getReturnType()), "Method return type must not be void");

4
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java
View File

@@ -34,7 +34,7 @@ import org.springframework.util.Assert;
*/
public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor {
private final MessageListMethodAdapter adapter;
private final MessageListProcessor adapter;
/**
* Creates a wrapper around the object passed in. This constructor will look for a method that can process
@@ -71,7 +71,7 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
@Override
protected final Object aggregatePayloads(MessageGroup group) {
final Collection<Message<?>> messagesUpForProcessing = group.getUnmarked();
Object result = this.adapter.executeMethod(messagesUpForProcessing);
Object result = this.adapter.process(messagesUpForProcessing);
return result;
}

8
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java → spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java
View File

@@ -30,23 +30,23 @@ import org.springframework.util.Assert;
* @author Marius Bogoevici
* @author Dave Syer
*/
public class ReleaseStrategyAdapter implements ReleaseStrategy {
public class MethodInvokingReleaseStrategy implements ReleaseStrategy {
private final MessageListMethodAdapter adapter;
public ReleaseStrategyAdapter(Object object, Method method) {
public MethodInvokingReleaseStrategy(Object object, Method method) {
adapter = new MessageListMethodAdapter(object, method);
this.assertMethodReturnsBoolean();
}
public ReleaseStrategyAdapter(Object object, String methodName) {
public MethodInvokingReleaseStrategy(Object object, String methodName) {
adapter = new MessageListMethodAdapter(object, methodName);
this.assertMethodReturnsBoolean();
}
public boolean canRelease(MessageGroup messages) {
return ((Boolean) adapter.executeMethod(messages.getUnmarked())).booleanValue() && messages.getMarked().isEmpty();
return ((Boolean) adapter.process(messages.getUnmarked())).booleanValue();
}
private void assertMethodReturnsBoolean() {

6
spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java
View File

@@ -20,7 +20,7 @@ import java.lang.reflect.Method;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.aggregator.CorrelationStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
import org.springframework.util.StringUtils;
@@ -57,12 +57,12 @@ public class CorrelationStrategyFactoryBean implements FactoryBean<CorrelationSt
}
if (target != null) {
if (StringUtils.hasText(methodName)) {
delegate = new CorrelationStrategyAdapter(target, methodName);
delegate = new MethodInvokingCorrelationStrategy(target, methodName);
}
else {
Method method = AnnotationFinder.findAnnotatedMethod(target, org.springframework.integration.annotation.CorrelationStrategy.class);
if (method != null) {
delegate = new CorrelationStrategyAdapter(target, method);
delegate = new MethodInvokingCorrelationStrategy(target, method);
}
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java
View File

@@ -19,7 +19,7 @@ import java.lang.reflect.Method;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.aggregator.ReleaseStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy;
import org.springframework.util.StringUtils;
@@ -56,12 +56,12 @@ public class ReleaseStrategyFactoryBean implements FactoryBean<ReleaseStrategy>
}
if (target != null) {
if (StringUtils.hasText(methodName)) {
delegate = new ReleaseStrategyAdapter(target, methodName);
delegate = new MethodInvokingReleaseStrategy(target, methodName);
}
else {
Method method = AnnotationFinder.findAnnotatedMethod(target, org.springframework.integration.annotation.ReleaseStrategy.class);
if (method != null) {
delegate = new ReleaseStrategyAdapter(target, method);
delegate = new MethodInvokingReleaseStrategy(target, method);
}
}
}

20
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -23,9 +23,9 @@ import java.util.concurrent.atomic.AtomicReference;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
import org.springframework.integration.aggregator.CorrelationStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
import org.springframework.integration.aggregator.ReleaseStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CorrelationStrategy;
import org.springframework.integration.annotation.ReleaseStrategy;
@@ -51,8 +51,8 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
@Override
protected MessageHandler createHandler(Object bean, Method method, Aggregator annotation) {
MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(bean, method);
ReleaseStrategyAdapter releaseStrategy = getReleaseStrategy(bean);
CorrelationStrategyAdapter correlationStrategy = getCorrelationStrategy(bean);
MethodInvokingReleaseStrategy releaseStrategy = getReleaseStrategy(bean);
MethodInvokingCorrelationStrategy correlationStrategy = getCorrelationStrategy(bean);
CorrelatingMessageHandler handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, releaseStrategy);
String discardChannelName = annotation.discardChannel();
if (StringUtils.hasText(discardChannelName)) {
@@ -71,26 +71,26 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
return handler;
}
private ReleaseStrategyAdapter getReleaseStrategy(final Object bean) {
final AtomicReference<ReleaseStrategyAdapter> reference = new AtomicReference<ReleaseStrategyAdapter>();
private MethodInvokingReleaseStrategy getReleaseStrategy(final Object bean) {
final AtomicReference<MethodInvokingReleaseStrategy> reference = new AtomicReference<MethodInvokingReleaseStrategy>();
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, ReleaseStrategy.class);
if (annotation != null) {
reference.set(new ReleaseStrategyAdapter(bean, method));
reference.set(new MethodInvokingReleaseStrategy(bean, method));
}
}
});
return reference.get();
}
private CorrelationStrategyAdapter getCorrelationStrategy(final Object bean) {
final AtomicReference<CorrelationStrategyAdapter> reference = new AtomicReference<CorrelationStrategyAdapter>();
private MethodInvokingCorrelationStrategy getCorrelationStrategy(final Object bean) {
final AtomicReference<MethodInvokingCorrelationStrategy> reference = new AtomicReference<MethodInvokingCorrelationStrategy>();
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CorrelationStrategy.class);
if (annotation != null) {
reference.set(new CorrelationStrategyAdapter(bean, method));
reference.set(new MethodInvokingCorrelationStrategy(bean, method));
}
}
});

45
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
View File

@@ -39,10 +39,14 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
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";
@@ -82,9 +86,18 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
processorBuilder.addConstructorArgValue(processor);
}
else {
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor")
.getBeanDefinition());
if (StringUtils.hasText(element.getAttribute(EXPRESSION_ATTRIBUTE))) {
String expression = element.getAttribute(EXPRESSION_ATTRIBUTE);
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.ExpressionEvaluatingMessageGroupProcessor");
adapterBuilder.addConstructorArgValue(expression);
builder.addConstructorArgValue(adapterBuilder.getBeanDefinition());
}
else {
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor")
.getBeanDefinition());
}
}
if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
@@ -99,20 +112,21 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
this.injectPropertyWithBean(RELEASE_STRATEGY_REF_ATTRIBUTE, RELEASE_STRATEGY_METHOD_ATTRIBUTE,
RELEASE_STRATEGY_PROPERTY, "ReleaseStrategy", element, builder, processor, parserContext);
this
.injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE,
CORRELATION_STRATEGY_PROPERTY, "CorrelationStrategy", element, builder, processor,
parserContext);
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 injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute, String beanProperty,
String adapterClass, Element element, BeanDefinitionBuilder builder,
BeanMetadataElement processor, ParserContext parserContext) {
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);
@@ -120,6 +134,13 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
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);
}

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

@@ -112,7 +112,7 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
if (StringUtils.hasText(method)) {
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.CorrelationStrategyAdapter");
+ ".aggregator.MethodInvokingCorrelationStrategy");
adapterBuilder.addConstructorArgReference(ref);
adapterBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method,
"java.lang.String");
@@ -133,7 +133,7 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
if (StringUtils.hasText(method)) {
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.ReleaseStrategyAdapter");
+ ".aggregator.MethodInvokingReleaseStrategy");
adapterBuilder.addConstructorArgReference(ref);
adapterBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method,
"java.lang.String");

4
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProcessor.java
View File

@@ -51,10 +51,10 @@ public abstract class AbstractMessageProcessor implements MessageProcessor {
}
catch (EvaluationException e) {
Throwable cause = e.getCause();
throw new MessageHandlingException(message, "Expression evaluation failed.", cause==null ? e : cause);
throw new MessageHandlingException(message, "Expression evaluation failed: "+expression.getExpressionString(), cause==null ? e : cause);
}
catch (Exception e) {
throw new MessageHandlingException(message, "Expression evaluation failed.", e);
throw new MessageHandlingException(message, "Expression evaluation failed: "+expression.getExpressionString(), e);
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java
View File

@@ -28,6 +28,7 @@ import org.springframework.expression.ParseException;
import org.springframework.expression.spel.SpelParserConfiguration;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.Message;
import org.springframework.util.Assert;
/**
* A {@link MessageProcessor} implementation that evaluates a SpEL expression
@@ -49,6 +50,7 @@ public class ExpressionEvaluatingMessageProcessor extends AbstractMessageProcess
* Create an {@link ExpressionEvaluatingMessageProcessor} for the given expression String.
*/
public ExpressionEvaluatingMessageProcessor(String expression) {
Assert.hasLength(expression, "The expression must be non empty");
try {
this.expression = parser.parseExpression(expression);
this.getEvaluationContext().addPropertyAccessor(new MapAccessor());

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

@@ -1950,6 +1950,13 @@ Name of the header whose value to use.
<xsd:complexType name="aggregator-type">
<xsd:complexContent>
<xsd:extension base="innerEndpointDefinitionAware">
<xsd:attribute name="expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A SpEL expression to be evaluated against the input message list as its root object.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="release-strategy" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
@@ -1968,6 +1975,11 @@ Name of the header whose value to use.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="release-strategy-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>An expression to apply to the message group</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="correlation-strategy" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
@@ -1986,6 +1998,11 @@ Name of the header whose value to use.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="correlation-strategy-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>An expression to apply to the message group</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="discard-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

10
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java
View File

@@ -39,32 +39,32 @@ public class CorrelationStrategyAdapterTests {
@Test
public void testMethodName() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new SimpleMessageCorrelator(), "getKey");
MethodInvokingCorrelationStrategy adapter = new MethodInvokingCorrelationStrategy(new SimpleMessageCorrelator(), "getKey");
assertEquals("b", adapter.getCorrelationKey(message));
}
@Test
public void testCorrelationStrategyAdapterObjectMethod() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new SimpleMessageCorrelator(),
MethodInvokingCorrelationStrategy adapter = new MethodInvokingCorrelationStrategy(new SimpleMessageCorrelator(),
ReflectionUtils.findMethod(SimpleMessageCorrelator.class, "getKey", Message.class));
assertEquals("b", adapter.getCorrelationKey(message));
}
@Test
public void testCorrelationStrategyAdapterPojoMethod() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new SimplePojoCorrelator(), "getKey");
MethodInvokingCorrelationStrategy adapter = new MethodInvokingCorrelationStrategy(new SimplePojoCorrelator(), "getKey");
assertEquals("foo", adapter.getCorrelationKey(message));
}
@Test
public void testHeaderPojoMethod() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new SimpleHeaderCorrelator(), "getKey");
MethodInvokingCorrelationStrategy adapter = new MethodInvokingCorrelationStrategy(new SimpleHeaderCorrelator(), "getKey");
assertEquals("b", adapter.getCorrelationKey(message));
}
@Test
public void testHeadersPojoMethod() {
CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new MultiHeaderCorrelator(),
MethodInvokingCorrelationStrategy adapter = new MethodInvokingCorrelationStrategy(new MultiHeaderCorrelator(),
ReflectionUtils.findMethod(MultiHeaderCorrelator.class, "getKey", String.class, String.class));
assertEquals("bd", adapter.getCorrelationKey(message));
}

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

@@ -0,0 +1,34 @@
package org.springframework.integration.aggregator;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import org.junit.Test;
import org.springframework.integration.core.GenericMessage;
/**
* @author Alex Peters
*
*/
public class ExpressionEvaluatingCorrelationStrategyTests {
private ExpressionEvaluatingCorrelationStrategy strategy;
@Test(expected = IllegalArgumentException.class)
public void testCreateInstanceWithEmptyExpressionFails() throws Exception {
strategy = new ExpressionEvaluatingCorrelationStrategy("");
}
@Test(expected = IllegalArgumentException.class)
public void testCreateInstanceWithNullExpressionFails() throws Exception {
strategy = new ExpressionEvaluatingCorrelationStrategy(null);
}
@Test
public void testCorrelationKeyWithMethodInvokingExpression() throws Exception {
strategy = new ExpressionEvaluatingCorrelationStrategy("payload.substring(0,1)");
Object correlationKey = strategy.getCorrelationKey(new GenericMessage<String>("bla"));
assertThat(correlationKey, is(String.class));
assertThat((String) correlationKey, is("b"));
}
}

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

@@ -0,0 +1,143 @@
package org.springframework.integration.aggregator;
import static org.junit.matchers.JUnitMatchers.hasItems;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.hamcrest.Description;
import org.hamcrest.Factory;
import org.hamcrest.Matcher;
import org.hamcrest.TypeSafeMatcher;
import org.hamcrest.core.IsEqual;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Matchers;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import org.springframework.integration.Message;
import org.springframework.integration.core.GenericMessage;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.store.MessageGroup;
/**
* @author Alex Peters
*
*/
@RunWith(MockitoJUnitRunner.class)
public class ExpressionEvaluatingMessageGroupProcessorTests {
private ExpressionEvaluatingMessageGroupProcessor processor;
private MessagingTemplate template = new MessagingTemplate();
@Mock
private MessageChannel outputChannel;
@Mock
private MessageGroup group;
List<Message<?>> messages = new ArrayList<Message<?>>();
@Before
@SuppressWarnings("unchecked")
public void setup() {
messages.clear();
for (int i = 0; i < 5; i++) {
messages.add(new GenericMessage(i + 1));
}
}
@Test
public void testProcessAndSendWithSizeExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("#root.size()");
processor.processAndSend(group, template, outputChannel);
verify(outputChannel).send(messageWithPayload(5));
}
@Test
public void testProcessAndSendWithProjectionExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("![payload]");
processor.processAndSend(group, template, outputChannel);
verify(outputChannel).send(messageWithPayload(hasItems(1, 2, 3, 4, 5)));
}
@Test
public void testProcessAndSendWithFilterAndProjectionExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("?[payload>2].![payload]");
processor.processAndSend(group, template, outputChannel);
verify(outputChannel).send(messageWithPayload(hasItems(3, 4, 5)));
}
@Test
public void testProcessAndSendWithFilterAndProjectionAndMethodInvokingExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor(String.format("T(%s).sum(?[payload>2].![payload])",
getClass().getName()));
processor.processAndSend(group, template, outputChannel);
verify(outputChannel).send(messageWithPayload(3 + 4 + 5));
}
private Message<?> messageWithPayload(Matcher<?> matcher) {
return Matchers.argThat(PayloadMatcher.hasPayload(matcher));
}
private Message<?> messageWithPayload(int i) {
return Matchers.argThat(PayloadMatcher.hasPayload(IsEqual.equalTo(i)));
}
/*
* sample static method invoked by SpEL
*/
public static Integer sum(Collection<Integer> values) {
int result = 0;
for (Integer value : values) {
result += value;
}
return result;
}
private static class PayloadMatcher extends TypeSafeMatcher<Message<?>> {
private final Matcher<?> matcher;
/**
* @param matcher
*/
PayloadMatcher(Matcher<?> matcher) {
super();
this.matcher = matcher;
}
/**
* {@inheritDoc}
*/
@Override
public boolean matchesSafely(Message<?> message) {
return matcher.matches(message.getPayload());
}
/**
* {@inheritDoc}
*/
//@Override
public void describeTo(Description description) {
description.appendText("a Message with payload: ").appendDescriptionOf(matcher);
}
@Factory
public static <T> Matcher<Message<?>> hasPayload(Matcher<T> payloadMatcher) {
return new PayloadMatcher(payloadMatcher);
}
}
}

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

@@ -0,0 +1,48 @@
package org.springframework.integration.aggregator;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.core.GenericMessage;
import org.springframework.integration.store.SimpleMessageGroup;
/**
* @author Alex Peters
* @author Dave Syer
*
*/
public class ExpressionEvaluatingReleaseStrategyTests {
private ExpressionEvaluatingReleaseStrategy strategy;
private SimpleMessageGroup messages = new SimpleMessageGroup("foo");
@Before
@SuppressWarnings("unchecked")
public void setup() {
for (int i = 0; i < 5; i++) {
messages.add(new GenericMessage(i + 1));
}
}
@Test
public void testCompletedWithSizeSpelEvaluated() throws Exception {
strategy = new ExpressionEvaluatingReleaseStrategy("#root.size()==5");
assertThat(strategy.canRelease(messages), is(true));
}
@Test
public void testCompletedWithFilterSpelEvaluated() throws Exception {
strategy = new ExpressionEvaluatingReleaseStrategy("!?[payload==5].empty");
assertThat(strategy.canRelease(messages), is(true));
}
@Test
public void testCompletedWithFilterSpelReturnsNotCompleted() throws Exception {
strategy = new ExpressionEvaluatingReleaseStrategy("!?[payload==6].empty");
assertThat(strategy.canRelease(messages), is(false));
}
}

34
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java
View File

@@ -42,21 +42,21 @@ public class ReleaseStrategyAdapterTests {
@Test
public void testTrueConvertedProperly() {
ReleaseStrategyAdapter adapter = new ReleaseStrategyAdapter(new AlwaysTrueReleaseStrategy(),
MethodInvokingReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new AlwaysTrueReleaseStrategy(),
"checkCompleteness");
Assert.assertTrue(adapter.canRelease(createListOfMessages(0)));
}
@Test
public void testFalseConvertedProperly() {
ReleaseStrategyAdapter adapter = new ReleaseStrategyAdapter(new AlwaysFalseReleaseStrategy(),
MethodInvokingReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new AlwaysFalseReleaseStrategy(),
"checkCompleteness");
Assert.assertTrue(!adapter.canRelease(createListOfMessages(0)));
}
@Test
public void testAdapterWithNonParameterizedMessageListBasedMethod() {
ReleaseStrategy adapter = new ReleaseStrategyAdapter(simpleReleaseStrategy,
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy,
"checkCompletenessOnNonParameterizedListOfMessages");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
@@ -64,7 +64,7 @@ public class ReleaseStrategyAdapterTests {
@Test
public void testAdapterWithWildcardParametrizedMessageBasedMethod() {
ReleaseStrategy adapter = new ReleaseStrategyAdapter(simpleReleaseStrategy,
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy,
"checkCompletenessOnListOfMessagesParametrizedWithWildcard");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
@@ -72,7 +72,7 @@ public class ReleaseStrategyAdapterTests {
@Test
public void testAdapterWithTypeParametrizedMessageBasedMethod() {
ReleaseStrategy adapter = new ReleaseStrategyAdapter(simpleReleaseStrategy,
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy,
"checkCompletenessOnListOfMessagesParametrizedWithString");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
@@ -80,69 +80,69 @@ public class ReleaseStrategyAdapterTests {
@Test
public void testAdapterWithPojoBasedMethod() {
ReleaseStrategy adapter = new ReleaseStrategyAdapter(simpleReleaseStrategy, "checkCompletenessOnListOfStrings");
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "checkCompletenessOnListOfStrings");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test
public void testAdapterWithPojoBasedMethodReturningObject() {
ReleaseStrategy adapter = new ReleaseStrategyAdapter(simpleReleaseStrategy, "checkCompletenessOnListOfStrings");
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "checkCompletenessOnListOfStrings");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test(expected = IllegalArgumentException.class)
public void testAdapterWithWrongMethodName() {
new ReleaseStrategyAdapter(simpleReleaseStrategy, "methodThatDoesNotExist");
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "methodThatDoesNotExist");
}
@Test(expected = IllegalArgumentException.class)
public void testInvalidParameterTypeUsingMethodName() {
new ReleaseStrategyAdapter(simpleReleaseStrategy, "invalidParameterType");
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "invalidParameterType");
}
@Test(expected = IllegalArgumentException.class)
public void testTooManyParametersUsingMethodName() {
new ReleaseStrategyAdapter(simpleReleaseStrategy, "tooManyParameters");
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "tooManyParameters");
}
@Test(expected = IllegalArgumentException.class)
public void testNotEnoughParametersUsingMethodName() {
new ReleaseStrategyAdapter(simpleReleaseStrategy, "notEnoughParameters");
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "notEnoughParameters");
}
@Test(expected = IllegalArgumentException.class)
public void testListSubclassParameterUsingMethodName() {
new ReleaseStrategyAdapter(simpleReleaseStrategy, "ListSubclassParameter");
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "ListSubclassParameter");
}
@Test(expected = IllegalArgumentException.class)
public void testWrongReturnType() throws SecurityException, NoSuchMethodError {
new ReleaseStrategyAdapter(simpleReleaseStrategy, "wrongReturnType");
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "wrongReturnType");
}
@Test(expected = IllegalArgumentException.class)
public void testTooManyParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
new ReleaseStrategyAdapter(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod(
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod(
"tooManyParameters", List.class, List.class));
}
@Test(expected = IllegalArgumentException.class)
public void testNotEnoughParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
new ReleaseStrategyAdapter(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod(
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod(
"notEnoughParameters", new Class[] {}));
}
@Test(expected = IllegalArgumentException.class)
public void testListSubclassParameterUsingMethodObject() throws SecurityException, NoSuchMethodException {
new ReleaseStrategyAdapter(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod(
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod(
"ListSubclassParameter", new Class[] { LinkedList.class }));
}
@Test(expected = IllegalArgumentException.class)
public void testWrongReturnTypeUsingMethodObject() throws SecurityException, NoSuchMethodException {
new ReleaseStrategyAdapter(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod("wrongReturnType",
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod("wrongReturnType",
new Class[] { List.class }));
}

2
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java
View File

@@ -54,7 +54,7 @@ public class DefaultMessageAggregatorIntegrationTests {
@SuppressWarnings("unchecked")
@Test(timeout = 1000)
public void aggregate() throws Exception {
public void testAggregation() throws Exception {
for (int i = 0; i < 5; i++) {
Map<String, Object> headers = stubHeaders(i, 5, 1);
input.send(new GenericMessage<Integer>(i, headers));

194
spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
View File

@@ -23,6 +23,7 @@ import static org.junit.Assert.assertThat;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Assert;
import org.junit.Before;
@@ -32,14 +33,19 @@ import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.aggregator.MessageListMethodAdapter;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.aggregator.ReleaseStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.core.MessageBuilder;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.MethodInvoker;
@@ -51,57 +57,75 @@ import org.springframework.integration.util.MethodInvoker;
*/
public class AggregatorParserTests {
private ApplicationContext context;
private ApplicationContext context;
@Before
public void setUp() {
this.context = new ClassPathXmlApplicationContext("aggregatorParserTests.xml", this.getClass());
}
@Before
public void setUp() {
this.context = new ClassPathXmlApplicationContext("aggregatorParserTests.xml", this.getClass());
}
@Test
public void testAggregation() {
MessageChannel input = (MessageChannel) context.getBean("aggregatorWithReferenceInput");
TestAggregatorBean aggregatorBean = (TestAggregatorBean) 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) {
input.send(message);
}
assertEquals("One and only one message must have been aggregated", 1, aggregatorBean.getAggregatedMessages()
.size());
Message<?> aggregatedMessage = aggregatorBean.getAggregatedMessages().get("id1");
assertEquals("The aggregated message payload is not correct", "123456789", aggregatedMessage.getPayload());
}
@Test
public void testAggregation() {
MessageChannel input = (MessageChannel) context.getBean("aggregatorWithReferenceInput");
TestAggregatorBean aggregatorBean = (TestAggregatorBean) 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) {
input.send(message);
}
assertEquals("One and only one message must have been aggregated", 1, aggregatorBean
.getAggregatedMessages().size());
Message<?> aggregatedMessage = aggregatorBean.getAggregatedMessages().get("id1");
assertEquals("The aggregated message payload is not correct", "123456789", aggregatedMessage
.getPayload());
}
@Test
public void testAggregationByExpression() {
MessageChannel input = (MessageChannel) context.getBean("aggregatorWithExpressionsInput");
SubscribableChannel outputChannel = (SubscribableChannel) context.getBean("aggregatorWithExpressionsOutput");
final AtomicReference<Message<?>> aggregatedMessage = new AtomicReference<Message<?>>();
outputChannel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
aggregatedMessage.set(message);
}
});
List<Message<?>> outboundMessages = new ArrayList<Message<?>>();
outboundMessages.add(MessageBuilder.withPayload("123").setHeader("foo", "1").build());
outboundMessages.add(MessageBuilder.withPayload("456").setHeader("foo", "1").build());
outboundMessages.add(MessageBuilder.withPayload("789").setHeader("foo", "1").build());
for (Message<?> message : outboundMessages) {
input.send(message);
}
assertEquals("The aggregated message payload is not correct", "[123]", aggregatedMessage.get().getPayload().toString());
}
@Test
public void testPropertyAssignment() throws Exception {
EventDrivenConsumer endpoint =
(EventDrivenConsumer) context.getBean("completelyDefinedAggregator");
ReleaseStrategy releaseStrategy = (ReleaseStrategy) context.getBean("releaseStrategy");
CorrelationStrategy correlationStrategy = (CorrelationStrategy) context.getBean("correlationStrategy");
MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel");
Object consumer = new DirectFieldAccessor(endpoint).getPropertyValue("handler");
assertThat(consumer, is(CorrelatingMessageHandler.class));
DirectFieldAccessor accessor = new DirectFieldAccessor(consumer);
Method expectedMethod = TestAggregatorBean.class.getMethod("createSingleMessageFromGroup", List.class);
assertEquals("The MethodInvokingAggregator is not injected with the appropriate aggregation method",
expectedMethod, ((MessageListMethodAdapter) new DirectFieldAccessor(accessor.getPropertyValue("outputProcessor")).getPropertyValue("adapter")).getMethod());
assertEquals(
"The AggregatorEndpoint is not injected with the appropriate ReleaseStrategy instance",
releaseStrategy, accessor.getPropertyValue("releaseStrategy"));
assertEquals("The AggregatorEndpoint is not injected with the appropriate CorrelationStrategy instance",
correlationStrategy, accessor.getPropertyValue("correlationStrategy"));
@Test
public void testPropertyAssignment() throws Exception {
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("completelyDefinedAggregator");
ReleaseStrategy releaseStrategy = (ReleaseStrategy) context.getBean("releaseStrategy");
CorrelationStrategy correlationStrategy = (CorrelationStrategy) context.getBean("correlationStrategy");
MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel");
Object consumer = new DirectFieldAccessor(endpoint).getPropertyValue("handler");
assertThat(consumer, is(CorrelatingMessageHandler.class));
DirectFieldAccessor accessor = new DirectFieldAccessor(consumer);
Method expectedMethod = TestAggregatorBean.class.getMethod("createSingleMessageFromGroup", List.class);
assertEquals("The MethodInvokingAggregator is not injected with the appropriate aggregation method",
expectedMethod, ((MessageListMethodAdapter) new DirectFieldAccessor(accessor
.getPropertyValue("outputProcessor")).getPropertyValue("adapter")).getMethod());
assertEquals("The AggregatorEndpoint is not injected with the appropriate ReleaseStrategy instance",
releaseStrategy, accessor.getPropertyValue("releaseStrategy"));
assertEquals("The AggregatorEndpoint is not injected with the appropriate CorrelationStrategy instance",
correlationStrategy, accessor.getPropertyValue("correlationStrategy"));
Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate output channel",
outputChannel, accessor.getPropertyValue("outputChannel"));
Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate discard channel",
discardChannel, accessor.getPropertyValue("discardChannel"));
Assert.assertEquals("The AggregatorEndpoint is not set with the appropriate timeout value",
86420000l, TestUtils.getPropertyValue(consumer, "messagingTemplate.sendTimeout"));
Assert.assertEquals("The AggregatorEndpoint is not set with the appropriate timeout value", 86420000l,
TestUtils.getPropertyValue(consumer, "messagingTemplate.sendTimeout"));
Assert.assertEquals(
"The AggregatorEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
true, accessor.getPropertyValue("sendPartialResultOnExpiry"));
@@ -110,8 +134,7 @@ public class AggregatorParserTests {
@Test
public void testSimpleJavaBeanAggregator() {
List<Message<?>> outboundMessages = new ArrayList<Message<?>>();
MessageChannel input =
(MessageChannel) context.getBean("aggregatorWithReferenceAndMethodInput");
MessageChannel input = (MessageChannel) context.getBean("aggregatorWithReferenceAndMethodInput");
outboundMessages.add(createMessage(1l, "id1", 3, 1, null));
outboundMessages.add(createMessage(2l, "id1", 3, 3, null));
outboundMessages.add(createMessage(3l, "id1", 3, 2, null));
@@ -123,54 +146,51 @@ public class AggregatorParserTests {
Assert.assertEquals(6l, response.getPayload());
}
@Test(expected=BeanCreationException.class)
@Test(expected = BeanCreationException.class)
public void testMissingMethodOnAggregator() {
context = new ClassPathXmlApplicationContext("invalidMethodNameAggregator.xml", this.getClass());
context = new ClassPathXmlApplicationContext("invalidMethodNameAggregator.xml", this.getClass());
}
@Test(expected=BeanCreationException.class)
@Test(expected = BeanCreationException.class)
public void testDuplicateReleaseStrategyDefinition() {
context = new ClassPathXmlApplicationContext(
"ReleaseStrategyMethodWithMissingReference.xml", this.getClass());
context = new ClassPathXmlApplicationContext("ReleaseStrategyMethodWithMissingReference.xml", this.getClass());
}
@Test
public void testAggregatorWithPojoReleaseStrategy() {
MessageChannel input = (MessageChannel) context.getBean("aggregatorWithPojoReleaseStrategyInput");
EventDrivenConsumer endpoint =
(EventDrivenConsumer) context.getBean("aggregatorWithPojoReleaseStrategy");
ReleaseStrategy releaseStrategy = (ReleaseStrategy) new DirectFieldAccessor(
new DirectFieldAccessor(endpoint).getPropertyValue("handler")).getPropertyValue("releaseStrategy");
Assert.assertTrue(releaseStrategy instanceof ReleaseStrategyAdapter);
DirectFieldAccessor releaseStrategyAccessor = new DirectFieldAccessor(new DirectFieldAccessor(releaseStrategy).getPropertyValue("adapter"));
MethodInvoker invoker = (MethodInvoker) releaseStrategyAccessor.getPropertyValue("invoker");
Assert.assertTrue(new DirectFieldAccessor(invoker).getPropertyValue("object") instanceof MaxValueReleaseStrategy);
Assert.assertTrue(((Method) releaseStrategyAccessor.getPropertyValue("method")).getName().equals("checkCompleteness"));
input.send(createMessage(1l, "correllationId", 4, 0, null));
input.send(createMessage(2l, "correllationId", 4, 1, null));
input.send(createMessage(3l, "correllationId", 4, 2, null));
PollableChannel outputChannel = (PollableChannel) context.getBean("outputChannel");
Message<?> reply = outputChannel.receive(0);
Assert.assertNull(reply);
input.send(createMessage(5l, "correllationId", 4, 3, null));
reply = outputChannel.receive(0);
Assert.assertNotNull(reply);
assertEquals(11l, reply.getPayload());
}
@Test
public void testAggregatorWithPojoReleaseStrategy() {
MessageChannel input = (MessageChannel) context.getBean("aggregatorWithPojoReleaseStrategyInput");
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("aggregatorWithPojoReleaseStrategy");
ReleaseStrategy releaseStrategy = (ReleaseStrategy) new DirectFieldAccessor(new DirectFieldAccessor(endpoint)
.getPropertyValue("handler")).getPropertyValue("releaseStrategy");
Assert.assertTrue(releaseStrategy instanceof MethodInvokingReleaseStrategy);
DirectFieldAccessor releaseStrategyAccessor = new DirectFieldAccessor(new DirectFieldAccessor(releaseStrategy)
.getPropertyValue("adapter"));
MethodInvoker invoker = (MethodInvoker) releaseStrategyAccessor.getPropertyValue("invoker");
Assert
.assertTrue(new DirectFieldAccessor(invoker).getPropertyValue("object") instanceof MaxValueReleaseStrategy);
Assert.assertTrue(((Method) releaseStrategyAccessor.getPropertyValue("method")).getName().equals(
"checkCompleteness"));
input.send(createMessage(1l, "correllationId", 4, 0, null));
input.send(createMessage(2l, "correllationId", 4, 1, null));
input.send(createMessage(3l, "correllationId", 4, 2, null));
PollableChannel outputChannel = (PollableChannel) context.getBean("outputChannel");
Message<?> reply = outputChannel.receive(0);
Assert.assertNull(reply);
input.send(createMessage(5l, "correllationId", 4, 3, null));
reply = outputChannel.receive(0);
Assert.assertNotNull(reply);
assertEquals(11l, reply.getPayload());
}
@Test(expected = BeanCreationException.class)
public void testAggregatorWithInvalidReleaseStrategyMethod() {
context = new ClassPathXmlApplicationContext("invalidReleaseStrategyMethod.xml", this.getClass());
}
@Test(expected = BeanCreationException.class)
public void testAggregatorWithInvalidReleaseStrategyMethod() {
context = new ClassPathXmlApplicationContext("invalidReleaseStrategyMethod.xml", this.getClass());
}
private static <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel outputChannel) {
return MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId)
.setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(outputChannel).build();
}
private static <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel outputChannel) {
return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber).setReplyChannel(outputChannel).build();
}
}

10
spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

@@ -30,8 +30,8 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.aggregator.CorrelationStrategyAdapter;
import org.springframework.integration.aggregator.ReleaseStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.core.MessageBuilder;
@@ -130,8 +130,8 @@ public class ResequencerParserTests {
CorrelatingMessageHandler.class);
Object correlationStrategy = getPropertyValue(resequencer, "correlationStrategy");
assertEquals("The ResequencerEndpoint is not configured with a CorrelationStrategy adapter",
CorrelationStrategyAdapter.class, correlationStrategy.getClass());
CorrelationStrategyAdapter adapter = (CorrelationStrategyAdapter) correlationStrategy;
MethodInvokingCorrelationStrategy.class, correlationStrategy.getClass());
MethodInvokingCorrelationStrategy adapter = (MethodInvokingCorrelationStrategy) correlationStrategy;
assertEquals("foo", adapter.getCorrelationKey(MessageBuilder.withPayload("not important").build()));
}
@@ -153,7 +153,7 @@ public class ResequencerParserTests {
CorrelatingMessageHandler handler = TestUtils.getPropertyValue(endpoint, "handler",
CorrelatingMessageHandler.class);
Object releaseStrategy = getPropertyValue(handler, "releaseStrategy");
assertEquals("The Resequencer is not configured with an adapter", ReleaseStrategyAdapter.class, releaseStrategy
assertEquals("The Resequencer is not configured with an adapter", MethodInvokingReleaseStrategy.class, releaseStrategy
.getClass());
}

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

@@ -29,6 +29,15 @@
send-timeout="86420000"
send-partial-result-on-expiry="true"/>
<channel id="aggregatorWithExpressionsInput"/>
<channel id="aggregatorWithExpressionsOutput"/>
<aggregator id="aggregatorWithExpressions"
input-channel="aggregatorWithExpressionsInput"
output-channel="aggregatorWithExpressionsOutput"
expression="?[payload.startsWith('1')].![payload]"
release-strategy-expression="#root.size()>2"
correlation-strategy-expression="headers['foo']"/>
<channel id="aggregatorWithReferenceAndMethodInput"/>
<aggregator id="aggregatorWithReferenceAndMethod"
ref="adderBean"

12
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

@@ -30,9 +30,9 @@ import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.aggregator.ReleaseStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
import org.springframework.integration.aggregator.CorrelationStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.context.BeanFactoryChannelResolver;
@@ -83,8 +83,8 @@ public class AggregatorAnnotationTests {
final String endpointName = "endpointWithDefaultAnnotationAndCustomReleaseStrategy";
MessageHandler aggregator = this.getAggregator(context, endpointName);
Object ReleaseStrategy = getPropertyValue(aggregator, "releaseStrategy");
Assert.assertTrue(ReleaseStrategy instanceof ReleaseStrategyAdapter);
ReleaseStrategyAdapter releaseStrategyAdapter = (ReleaseStrategyAdapter) ReleaseStrategy;
Assert.assertTrue(ReleaseStrategy instanceof MethodInvokingReleaseStrategy);
MethodInvokingReleaseStrategy releaseStrategyAdapter = (MethodInvokingReleaseStrategy) ReleaseStrategy;
DirectFieldAccessor invokerAccessor = new DirectFieldAccessor(new DirectFieldAccessor(new DirectFieldAccessor(
releaseStrategyAdapter).getPropertyValue("adapter")).getPropertyValue("invoker"));
Object targetObject = invokerAccessor.getPropertyValue("object");
@@ -100,8 +100,8 @@ public class AggregatorAnnotationTests {
final String endpointName = "endpointWithCorrelationStrategy";
MessageHandler aggregator = this.getAggregator(context, endpointName);
Object correlationStrategy = getPropertyValue(aggregator, "correlationStrategy");
Assert.assertTrue(correlationStrategy instanceof CorrelationStrategyAdapter);
CorrelationStrategyAdapter ReleaseStrategyAdapter = (CorrelationStrategyAdapter) correlationStrategy;
Assert.assertTrue(correlationStrategy instanceof MethodInvokingCorrelationStrategy);
MethodInvokingCorrelationStrategy ReleaseStrategyAdapter = (MethodInvokingCorrelationStrategy) correlationStrategy;
DirectFieldAccessor processorAccessor = new DirectFieldAccessor(new DirectFieldAccessor(ReleaseStrategyAdapter)
.getPropertyValue("processor"));
Object targetObject = processorAccessor.getPropertyValue("targetObject");