diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractExpressionEvaluatingMessageListProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractExpressionEvaluatingMessageListProcessor.java new file mode 100644 index 0000000000..470b2b7958 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractExpressionEvaluatingMessageListProcessor.java @@ -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 @beanName 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> 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> messages) { + return this.evaluateExpression(this.expression, messages, this.expectedType); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.java new file mode 100644 index 0000000000..69fcb2cd40 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategy.java @@ -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); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessor.java new file mode 100644 index 0000000000..71b950b287 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessor.java @@ -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()); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java new file mode 100644 index 0000000000..1bd11f6bdd --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java @@ -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(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java index ecc3fded5f..9702ab611f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java @@ -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> messages) { + /* (non-Javadoc) + * @see org.springframework.integration.aggregator.MessageListProcessor#executeMethod(java.util.Collection) + */ + public final Object process(Collection> messages) { try { if (isMethodParameterParameterized(this.method) && isHavingActualTypeArguments(this.method) && (isActualTypeRawMessage(this.method) || isActualTypeParameterizedMessage(this.method))) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java index 75d09a65c9..6c148073c1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java @@ -38,7 +38,7 @@ import org.springframework.util.ReflectionUtils; */ public class MessageListMethodAdapterHelper { - public MessageListMethodAdapter getAdapter(Object candidate, Class annotationType) { + public MessageListProcessor getAdapter(Object candidate, Class annotationType) { Method method = findAggregatorMethod(candidate, annotationType); if (method == null) { return null; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListProcessor.java new file mode 100644 index 0000000000..e04fb081c8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListProcessor.java @@ -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> messages); + +} \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java similarity index 87% rename from spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java rename to spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java index c5ecaee8f0..e3e86ca52b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java @@ -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"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java index 93ccfda272..833aa61f96 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java @@ -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> messagesUpForProcessing = group.getUnmarked(); - Object result = this.adapter.executeMethod(messagesUpForProcessing); + Object result = this.adapter.process(messagesUpForProcessing); return result; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java similarity index 84% rename from spring-integration-core/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java rename to spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java index 36cf75bfbe..f75fb01db1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java @@ -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() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java index f9a520c55a..e455386c37 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java @@ -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 } 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); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java index 65ac190c54..47634bcbc5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java @@ -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 reference = new AtomicReference(); + private MethodInvokingReleaseStrategy getReleaseStrategy(final Object bean) { + final AtomicReference reference = new AtomicReference(); 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 reference = new AtomicReference(); + private MethodInvokingCorrelationStrategy getCorrelationStrategy(final Object bean) { + final AtomicReference reference = new AtomicReference(); 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)); } } }); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java index 8c4793ede4..ff6466939b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java @@ -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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java index d720ca20b0..53686e4261 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java @@ -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"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProcessor.java index 8a4384958c..e1e55393f2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProcessor.java @@ -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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java index 0249fc6aa3..3950087ac0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java @@ -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()); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 5dc15fb1a5..e9a5ed0644 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -1950,6 +1950,13 @@ Name of the header whose value to use. + + + + A SpEL expression to be evaluated against the input message list as its root object. + + + @@ -1968,6 +1975,11 @@ Name of the header whose value to use. + + + An expression to apply to the message group + + @@ -1986,6 +1998,11 @@ Name of the header whose value to use. + + + An expression to apply to the message group + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java index 06ee9cc7ec..5f3a068c38 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java @@ -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)); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategyTests.java new file mode 100644 index 0000000000..c52b3793bb --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingCorrelationStrategyTests.java @@ -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("bla")); + assertThat(correlationKey, is(String.class)); + assertThat((String) correlationKey, is("b")); + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java new file mode 100644 index 0000000000..5a92e859c5 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java @@ -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> messages = new ArrayList>(); + + @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 values) { + int result = 0; + for (Integer value : values) { + result += value; + } + return result; + } + + private static class PayloadMatcher extends TypeSafeMatcher> { + + 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 Matcher> hasPayload(Matcher payloadMatcher) { + return new PayloadMatcher(payloadMatcher); + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategyTests.java new file mode 100644 index 0000000000..89393bab53 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategyTests.java @@ -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)); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java index c6d61d1452..6b52f5eada 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java @@ -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 })); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java index a50c977740..c6aa5571a2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java @@ -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 headers = stubHeaders(i, 5, 1); input.send(new GenericMessage(i, headers)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index 56e9dde76f..98827e2a15 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -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> outboundMessages = new ArrayList>(); + outboundMessages.add(createMessage("123", "id1", 3, 1, null)); + outboundMessages.add(createMessage("789", "id1", 3, 3, null)); + outboundMessages.add(createMessage("456", "id1", 3, 2, null)); + for (Message message : outboundMessages) { + 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> outboundMessages = new ArrayList>(); - outboundMessages.add(createMessage("123", "id1", 3, 1, null)); - outboundMessages.add(createMessage("789", "id1", 3, 3, null)); - outboundMessages.add(createMessage("456", "id1", 3, 2, null)); - for (Message message : outboundMessages) { - 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> aggregatedMessage = new AtomicReference>(); + outputChannel.subscribe(new MessageHandler() { + public void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, + MessageDeliveryException { + aggregatedMessage.set(message); + } + }); + List> outboundMessages = new ArrayList>(); + 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> outboundMessages = new ArrayList>(); - 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 Message 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 Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, + MessageChannel outputChannel) { + return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize) + .setSequenceNumber(sequenceNumber).setReplyChannel(outputChannel).build(); + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index cc4fa8f1c9..7575dc5b84 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -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()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml index 6d2d14c863..425441b4e9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml @@ -29,6 +29,15 @@ send-timeout="86420000" send-partial-result-on-expiry="true"/> + + + +