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

INT-538, INT-665, INT-818: @Header, @Headers and (new) @Payloads recognised in message list POJOs

Browse Source 
* All method scanning and expression building is consolidated in one place
* MessageProcessor and MessageListProcessor implementations share commmon delegate helper
This commit is contained in:
David Syer
2010-08-10 10:48:25 +00:00
parent 7df44e7da3
commit e6b508ca87
34 changed files with 1705 additions and 1312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

152
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
View File

@@ -13,33 +13,24 @@
package org.springframework.integration.aggregator;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.core.MessageBuilder;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.splitter.AbstractMessageSplitter;
import org.springframework.integration.store.MessageGroup;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* Base class for MessageGroupProcessor implementations that aggregate the group of Messages into a single Message.
@@ -47,22 +38,28 @@ import org.springframework.util.ReflectionUtils;
* @author Iwein Fuld
* @author Alexander Peters
* @author Mark Fisher
* @author Dave Syer
*
* @since 2.0
*/
public abstract class AbstractAggregatingMessageGroupProcessor implements MessageGroupProcessor {
private final Log logger = LogFactory.getLog(this.getClass());
@SuppressWarnings("unchecked")
public final void processAndSend(MessageGroup group, MessagingTemplate channelTemplate, MessageChannel outputChannel) {
public final void processAndSend(MessageGroup group, MessagingOperations messagingTemplate, MessageChannel outputChannel) {
Assert.notNull(group, "MessageGroup must not be null");
Assert.notNull(outputChannel, "'outputChannel' must not be null");
Object payload = this.aggregatePayloads(group);
Map<String, Object> headers = this.aggregateHeaders(group);
MessageBuilder<?> builder = (payload instanceof Message) ? MessageBuilder.fromMessage((Message<?>) payload)
: MessageBuilder.withPayload(payload);
Message<?> message = builder.copyHeadersIfAbsent(headers).build();
channelTemplate.send(outputChannel, message);
Object payload = this.aggregatePayloads(group, headers);
MessageBuilder<?> builder;
if (payload instanceof Message<?>) {
builder = MessageBuilder.fromMessage((Message<?>) payload);
}
else {
builder = MessageBuilder.withPayload(payload).copyHeadersIfAbsent(headers);
}
Message<?> message = builder.build();
messagingTemplate.send(outputChannel, message);
}
/**
@@ -120,125 +117,6 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
return aggregatedHeaders;
}
protected abstract Object aggregatePayloads(MessageGroup group);
protected MessageListProcessor getAdapter(Object candidate, Class<? extends Annotation> annotationType) {
Method method = findAggregatorMethod(candidate, annotationType);
if (method == null) {
return null;
}
return new MethodInvokingMessageListProcessor(candidate, method);
}
private Method findAggregatorMethod(Object candidate, Class<? extends Annotation> annotationType) {
Class<?> targetClass = AopUtils.getTargetClass(candidate);
if (targetClass == null) {
targetClass = candidate.getClass();
}
Method method = this.findAnnotatedMethod(targetClass, annotationType);
if (method == null) {
method = this.findSinglePublicMethod(targetClass);
}
return method;
}
private Method findAnnotatedMethod(final Class<?> targetClass, final Class<? extends Annotation> annotationType) {
final AtomicReference<Method> annotatedMethod = new AtomicReference<Method>();
ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.findAnnotation(method, annotationType);
if (annotation != null) {
Assert.isNull(annotatedMethod.get(), "found more than one method on target class [" + targetClass
+ "] with the annotation type [" + annotationType.getName() + "]");
annotatedMethod.set(method);
}
}
});
return annotatedMethod.get();
}
private Method findSinglePublicMethod(Class<?> targetClass) {
Set<Method> methods = new HashSet<Method>();
for (Method method : targetClass.getMethods()) {
if (!method.getDeclaringClass().equals(Object.class)) {
methods.add(method);
}
}
removeListIncompatibleMethodsFrom(methods);
removeVoidMethodsFrom(methods);
removeUnfittingFrom(methods);
if (methods.size() > 1) {
throw new IllegalArgumentException("Class [" + targetClass + "] contains more than one public Method.");
}
return methods.isEmpty() ? null : methods.iterator().next();
}
private void removeListIncompatibleMethodsFrom(Set<Method> candidates) {
removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
int found = 0;
for (Class<?> parameterClass : method.getParameterTypes()) {
if (Collection.class.isAssignableFrom(parameterClass)) {
found++;
}
}
return found != 1;
}
});
}
private void removeVoidMethodsFrom(Set<Method> candidates) {
removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
return method.getReturnType().getName().equals("void");
}
});
}
private Set<Method> removeUnfittingFrom(Set<Method> candidates) {
return removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
Annotation[][] parameterAnnotations = method.getParameterAnnotations();
Class<?>[] parameterTypes = method.getParameterTypes();
return (!isFittinglyAnnotated(parameterTypes, parameterAnnotations));
}
});
}
private boolean isFittinglyAnnotated(Class<?>[] parameterTypes, Annotation[][] parameterAnnotations) {
int candidateParametersFound = 0;
for (int i = 0; i < parameterTypes.length; i++) {
Class<?> parameterType = parameterTypes[i];
if (Collection.class.isAssignableFrom(parameterType)) {
boolean headerAnnotationFound = false;
for (Annotation annotation : parameterAnnotations[i]) {
if (annotation instanceof Header) {
headerAnnotationFound = true;
}
}
if (!headerAnnotationFound) {
candidateParametersFound++;
}
}
}
return candidateParametersFound == 1;
}
private Set<Method> removeMethodsMatchingSelector(Set<Method> candidates, MethodSelector selector) {
Set<Method> removed = new HashSet<Method>();
Iterator<Method> iterator = candidates.iterator();
while (iterator.hasNext()) {
Method method = iterator.next();
if (selector.select(method)) {
iterator.remove();
removed.add(method);
}
}
return removed;
}
private interface MethodSelector {
boolean select(Method method);
}
protected abstract Object aggregatePayloads(MessageGroup group, Map<String, Object> defaultHeaders);
}

3
spring-integration-core/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.aggregator;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import org.springframework.integration.Message;
import org.springframework.integration.store.MessageGroup;
@@ -36,7 +37,7 @@ import org.springframework.util.Assert;
public class DefaultAggregatingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor {
@Override
protected final Object aggregatePayloads(MessageGroup group) {
protected final Object aggregatePayloads(MessageGroup group, Map<String, Object> headers) {
Collection<Message<?>> messages = group.getUnmarked();
Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups");
List<Object> payloads = new ArrayList<Object>(messages.size());

6
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessor.java
View File

@@ -1,5 +1,7 @@
package org.springframework.integration.aggregator;
import java.util.Map;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.convert.ConversionService;
@@ -37,10 +39,10 @@ public class ExpressionEvaluatingMessageGroupProcessor extends AbstractAggregati
/**
* Evaluate the expression provided on the unmarked messages (a collection) in the group, and delegate to the
* {@link MessagingTemplate} to send dowstream.
* {@link MessagingTemplate} to send downstream.
*/
@Override
protected Object aggregatePayloads(MessageGroup group) {
protected Object aggregatePayloads(MessageGroup group, Map<String, Object> headers) {
return processor.process(group.getUnmarked());
}

63
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageListProcessor.java
View File

@@ -18,22 +18,14 @@ 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.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.context.SimpleBeanResolver;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.transformer.MessageTransformationException;
import org.springframework.integration.util.AbstractExpressionEvaluator;
/**
* A base class for aggregators that evaluates a SpEL expression with the message list as the root object within the
@@ -42,29 +34,14 @@ import org.springframework.integration.transformer.MessageTransformationExceptio
* @author Dave Syer
* @since 2.0
*/
public class ExpressionEvaluatingMessageListProcessor implements BeanFactoryAware, MessageListProcessor {
public class ExpressionEvaluatingMessageListProcessor extends AbstractExpressionEvaluator implements MessageListProcessor {
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 ExpressionEvaluatingMessageListProcessor(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.
*/
@@ -72,37 +49,13 @@ public class ExpressionEvaluatingMessageListProcessor implements BeanFactoryAwar
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 SimpleBeanResolver(beanFactory));
}
}
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) {
public ExpressionEvaluatingMessageListProcessor(String expression) {
try {
return (expectedType != null) ? expression.getValue(this.evaluationContext, messages, expectedType)
: expression.getValue(this.evaluationContext, messages);
this.expression = parser.parseExpression(expression);
this.getEvaluationContext().addPropertyAccessor(new MapAccessor());
}
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);
catch (ParseException e) {
throw new IllegalArgumentException("Failed to parse expression.", e);
}
}

8
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java
View File

@@ -14,7 +14,7 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
/**
@@ -26,11 +26,11 @@ import org.springframework.integration.store.MessageGroup;
public interface MessageGroupProcessor {
/**
* Process the given group and send the resulting message(s) to the output channel using the messaging template.
* Implementations are free to send as little or as many messages based on the invocation as needed. For example an
* Process the given group and send the resulting message(s) to the output channel using the messaging operations.
* Implementations are free to send as few or as many messages based on the invocation as needed. For example an
* aggregating processor will send only a single message representing the group, where a resequencing strategy will
* send all messages in the group individually.
*/
void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate, MessageChannel outputChannel);
void processAndSend(MessageGroup group, MessagingOperations messagingTemplate, MessageChannel outputChannel);
}

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

@@ -18,11 +18,13 @@ package org.springframework.integration.aggregator;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Map;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.Message;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.store.MessageGroup;
import org.springframework.util.Assert;
/**
* MessageGroupProcessor that serves as an adapter for the invocation of a POJO method.
@@ -34,7 +36,7 @@ import org.springframework.util.Assert;
*/
public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor {
private final MessageListProcessor adapter;
private final MethodInvokingMessageListProcessor processor;
/**
* Creates a wrapper around the object passed in. This constructor will look for a method that can process
@@ -43,8 +45,7 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
* @param target the object to wrap
*/
public MethodInvokingMessageGroupProcessor(Object target) {
this.adapter = getAdapter(target, Aggregator.class);
Assert.notNull(this.adapter, "No aggregator method could be found for object of type: "+target.getClass());
this.processor = new MethodInvokingMessageListProcessor(target, Aggregator.class);
}
/**
@@ -55,7 +56,7 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
* @param methodName the name of the method to invoke
*/
public MethodInvokingMessageGroupProcessor(Object target, String methodName) {
this.adapter = new MethodInvokingMessageListProcessor(target, methodName);
this.processor = new MethodInvokingMessageListProcessor(target, methodName);
}
/**
@@ -65,13 +66,21 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
* @param method the method to invoke
*/
public MethodInvokingMessageGroupProcessor(Object target, Method method) {
this.adapter = new MethodInvokingMessageListProcessor(target, method);
this.processor = new MethodInvokingMessageListProcessor(target, method);
}
public void setConversionService(ConversionService conversionService) {
processor.setConversionService(conversionService);
}
public void setBeanFactory(BeanFactory beanFactory) {
processor.setBeanFactory(beanFactory);
}
@Override
protected final Object aggregatePayloads(MessageGroup group) {
protected final Object aggregatePayloads(MessageGroup group, Map<String, Object> headers) {
final Collection<Message<?>> messagesUpForProcessing = group.getUnmarked();
Object result = this.adapter.process(messagesUpForProcessing);
Object result = this.processor.process(messagesUpForProcessing, headers);
return result;
}

125
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* 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.
@@ -16,106 +16,61 @@
package org.springframework.integration.aggregator;
import java.lang.reflect.InvocationTargetException;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.util.DefaultMethodInvoker;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.integration.util.AbstractExpressionEvaluator;
import org.springframework.integration.util.MessagingMethodInvokerHelper;
/**
* Base class for implementing adapters for methods which take as an argument a
* list of {@link Message Message} instances or payloads.
*
* @author Marius Bogoevici
* @author Iwein Fuld
* A MessageListProcessor implementation that invokes a method on a target POJO.
*
* @author Dave Syer
* @since 2.0
*/
public class MethodInvokingMessageListProcessor implements MessageListProcessor {
public class MethodInvokingMessageListProcessor extends AbstractExpressionEvaluator {
private final DefaultMethodInvoker invoker;
private final MessagingMethodInvokerHelper delegate;
protected final Method method;
public MethodInvokingMessageListProcessor(Object targetObject, Method method, Class<?> expectedType) {
delegate = new MessagingMethodInvokerHelper(targetObject, method, expectedType, true);
}
public MethodInvokingMessageListProcessor(Object object, String methodName) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(methodName, "'methodName' must not be null");
this.method = ReflectionUtils.findMethod(object.getClass(), methodName, new Class<?>[]{List.class});
Assert.notNull(this.method, "Method '" + methodName +
"(List<?> args)' not found on '" + object.getClass().getName() + "'.");
this.invoker = new DefaultMethodInvoker(object, this.method);
}
public MethodInvokingMessageListProcessor(Object targetObject, Method method) {
delegate = new MessagingMethodInvokerHelper(targetObject, method, true);
}
public MethodInvokingMessageListProcessor(Object object, Method method) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(method, "'method' must not be null");
Assert.isTrue(method.getParameterTypes().length == 1
&& method.getParameterTypes()[0].equals(List.class),
"Method " + method + " does not accept exactly one parameter, of type List.");
this.method = method;
this.invoker = new DefaultMethodInvoker(object, this.method);
}
public MethodInvokingMessageListProcessor(Object targetObject, String methodName, Class<?> expectedType) {
delegate = new MessagingMethodInvokerHelper(targetObject, methodName,
expectedType, true);
}
public MethodInvokingMessageListProcessor(Object targetObject, String methodName) {
delegate = new MessagingMethodInvokerHelper(targetObject, methodName, true);
}
public Method getMethod() {
return method;
}
public MethodInvokingMessageListProcessor(Object targetObject, Class<? extends Annotation> annotationType) {
delegate = new MessagingMethodInvokerHelper(targetObject, annotationType, Object.class, true);
}
/* (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))) {
return this.invoker.invokeMethod(messages);
}
return this.invoker.invokeMethod(extractPayloadsFromMessages(messages));
}
catch (InvocationTargetException e) {
throw new MessagingException(
"Method '" + this.method + "' threw an Exception.", e.getTargetException());
}
catch (Exception e) {
throw new MessagingException("Failed to invoke method '" + this.method + "'.");
}
}
public String toString() {
return delegate.toString();
}
private static boolean isActualTypeParameterizedMessage(Method method) {
return (getCollectionActualType(method) instanceof ParameterizedType)
&& Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) getCollectionActualType(method)).getRawType());
}
private List<?> extractPayloadsFromMessages(Collection<? extends Message<?>> messages) {
List<Object> payloadList = new ArrayList<Object>();
for (Message<?> message : messages) {
payloadList.add(message.getPayload());
}
return payloadList;
}
private static boolean isActualTypeRawMessage(Method method) {
return getCollectionActualType(method).equals(Message.class);
}
private static Type getCollectionActualType(Method method) {
return ((ParameterizedType) method.getGenericParameterTypes()[0]).getActualTypeArguments()[0];
}
private static boolean isHavingActualTypeArguments(Method method) {
return ((ParameterizedType) method.getGenericParameterTypes()[0]).getActualTypeArguments().length == 1;
}
private static boolean isMethodParameterParameterized(Method method) {
return method.getGenericParameterTypes().length == 1
&& method.getGenericParameterTypes()[0] instanceof ParameterizedType;
}
public Object process(Collection<? extends Message<?>> messages, Map<String, Object> aggregateHeaders) {
try {
return delegate.process(new ArrayList<Message<?>>(messages), aggregateHeaders);
}
catch (RuntimeException e) {
throw e;
}
catch (Exception e) {
throw new IllegalStateException("Failed to process message list", e);
}
}
}

29
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java
View File

@@ -18,14 +18,12 @@ package org.springframework.integration.aggregator;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.store.MessageGroup;
import org.springframework.util.Assert;
/**
* Adapter for methods annotated with
* {@link org.springframework.integration.annotation.ReleaseStrategy @ReleaseStrategy}
* and for '<code>release-strategy</code>' elements that include a '<code>method</code>'
* attribute (e.g. &lt;release-strategy ref="beanReference" method="methodName"/&gt;).
* A {@link ReleaseStrategy} that invokes a method on a plain old Java object.
*
* @author Marius Bogoevici
* @author Dave Syer
@@ -35,24 +33,23 @@ public class MethodInvokingReleaseStrategy implements ReleaseStrategy {
private final MethodInvokingMessageListProcessor adapter;
public MethodInvokingReleaseStrategy(Object object, Method method) {
adapter = new MethodInvokingMessageListProcessor(object, method);
this.assertMethodReturnsBoolean();
adapter = new MethodInvokingMessageListProcessor(object, method, Boolean.class);
}
public MethodInvokingReleaseStrategy(Object object, String methodName) {
adapter = new MethodInvokingMessageListProcessor(object, methodName);
this.assertMethodReturnsBoolean();
adapter = new MethodInvokingMessageListProcessor(object, methodName, Boolean.class);
}
public void setBeanFactory(BeanFactory beanFactory) {
adapter.setBeanFactory(beanFactory);
}
public void setConversionService(ConversionService conversionService) {
adapter.setConversionService(conversionService);
}
public boolean canRelease(MessageGroup messages) {
return ((Boolean) adapter.process(messages.getUnmarked())).booleanValue();
}
private void assertMethodReturnsBoolean() {
Assert.isTrue(Boolean.class.equals(adapter.getMethod().getReturnType())
|| boolean.class.equals(adapter.getMethod().getReturnType()),
"Method '" + adapter.getMethod().getName() + "' does not return a boolean value");
return (Boolean) adapter.process(messages.getUnmarked(), null);
}
}

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

@@ -15,7 +15,7 @@ package org.springframework.integration.aggregator;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
/**
@@ -28,7 +28,7 @@ import org.springframework.integration.store.MessageGroup;
*/
public class PassThroughMessageGroupProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate, MessageChannel outputChannel) {
public void processAndSend(MessageGroup group, MessagingOperations messagingTemplate, MessageChannel outputChannel) {
for (Message<?> message : group.getUnmarked()) {
messagingTemplate.send(outputChannel, message);
}

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

@@ -21,7 +21,7 @@ import java.util.List;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
/**
@@ -45,7 +45,7 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
this.comparator = comparator;
}
public void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate, MessageChannel outputChannel) {
public void processAndSend(MessageGroup group, MessagingOperations messagingTemplate, MessageChannel outputChannel) {
Collection<Message<?>> messages = group.getUnmarked();
if (messages.size() > 0) {
List<Message<?>> sorted = new ArrayList<Message<?>>(messages);

45
spring-integration-core/src/main/java/org/springframework/integration/annotation/Payloads.java Normal file
View File

@@ -0,0 +1,45 @@
/*
* 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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* This annotation marks a method parameter as being a list of message payloads, for POJO handlers that deal with lists
* of messages (e.g. aggregators and release strategies).
* <p>
* Example: void foo(@Payloads("city.name") List<String> cityName) - will map the value of the 'name' property of the 'city'
* property of all the payload objects in the input list.
*
* @author Dave Syer
* @since 2.0
*/
@Target( { ElementType.PARAMETER, ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Payloads {
/**
* Expression for matching against nested properties of the payloads.
*/
String value() default "";
}

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

@@ -19,7 +19,6 @@ package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
@@ -30,7 +29,6 @@ import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.NamespaceHandler;
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.ObjectUtils;
/**
* Base class for NamespaceHandlers that registers a BeanFactoryPostProcessor

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

@@ -22,7 +22,6 @@ import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;

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

@@ -16,58 +16,16 @@
package org.springframework.integration.handler;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.convert.ConversionService;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.util.BeanFactoryTypeConverter;
import org.springframework.integration.util.AbstractExpressionEvaluator;
/**
* @author Mark Fisher
* @author Dave Syer
* @since 2.0
*/
public abstract class AbstractMessageProcessor implements MessageProcessor, BeanFactoryAware {
public abstract class AbstractMessageProcessor extends AbstractExpressionEvaluator implements MessageProcessor {
private final StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
private final BeanFactoryTypeConverter typeConverter = new BeanFactoryTypeConverter();
public AbstractMessageProcessor() {
evaluationContext.setTypeConverter(typeConverter);
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
typeConverter.setBeanFactory(beanFactory);
}
public void setConversionService(ConversionService conversionService) {
if (conversionService != null) {
typeConverter.setConversionService(conversionService);
}
}
protected StandardEvaluationContext getEvaluationContext() {
return this.evaluationContext;
}
protected Object evaluateExpression(Expression expression, Message<?> message, Class<?> expectedType) {
try {
return (expectedType != null)
? expression.getValue(this.evaluationContext, message, expectedType)
: expression.getValue(this.evaluationContext, message);
}
catch (EvaluationException e) {
Throwable cause = e.getCause();
throw new MessageHandlingException(message, "Expression evaluation failed: "+expression.getExpressionString(), cause==null ? e : cause);
}
catch (Exception e) {
throw new MessageHandlingException(message, "Expression evaluation failed: "+expression.getExpressionString(), e);
}
}
abstract public Object processMessage(Message<?> message);
}

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

@@ -16,8 +16,6 @@
package org.springframework.integration.handler;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.context.expression.MapAccessor;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
@@ -25,7 +23,6 @@ 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.integration.context.SimpleBeanResolver;
import org.springframework.util.Assert;
/**
@@ -35,7 +32,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @since 2.0
*/
public class ExpressionEvaluatingMessageProcessor extends AbstractMessageProcessor implements BeanFactoryAware {
public class ExpressionEvaluatingMessageProcessor extends AbstractMessageProcessor {
private final ExpressionParser parser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
@@ -66,16 +63,6 @@ public class ExpressionEvaluatingMessageProcessor extends AbstractMessageProcess
this.expectedType = expectedType;
}
/**
* Specify a BeanFactory in order to enable resolution via <code>@beanName</code> in the expression.
*/
public void setBeanFactory(final BeanFactory beanFactory) {
super.setBeanFactory(beanFactory);
if (beanFactory != null) {
this.getEvaluationContext().setBeanResolver(new SimpleBeanResolver(beanFactory));
}
}
/**
* Processes the Message by evaluating the expression with that Message as the
* root object. The expression evaluation result Object will be returned.

447
spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java
View File

@@ -18,452 +18,61 @@ package org.springframework.integration.handler;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.context.expression.MapAccessor;
import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
import org.springframework.core.MethodParameter;
import org.springframework.core.ParameterNameDiscoverer;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.annotation.Headers;
import org.springframework.integration.annotation.Payload;
import org.springframework.integration.util.ClassUtils;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.util.ReflectionUtils.MethodCallback;
import org.springframework.util.ReflectionUtils.MethodFilter;
import org.springframework.integration.util.MessagingMethodInvokerHelper;
/**
* A MessageProcessor implementation that invokes a method on a target Object.
* The Method instance or method name may be provided as a constructor argument.
* If a method name is provided, and more than one declared method has that name,
* the method-selection will be dynamic, based on the underlying SpEL method
* resolution. Alternatively, an annotation type may be provided so that the
* candidates for SpEL's method resolution are determined by the presence of that
* A MessageProcessor implementation that invokes a method on a target Object. The Method instance or method name may be
* provided as a constructor argument. If a method name is provided, and more than one declared method has that name,
* the method-selection will be dynamic, based on the underlying SpEL method resolution. Alternatively, an annotation
* type may be provided so that the candidates for SpEL's method resolution are determined by the presence of that
* annotation rather than the method name.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Dave Syer
*
* @since 2.0
*/
public class MethodInvokingMessageProcessor extends AbstractMessageProcessor {
private final Log logger = LogFactory.getLog(this.getClass());
private final Object targetObject;
private volatile String displayString;
private volatile boolean requiresReply;
private final Map<Class<?>, HandlerMethod> handlerMethods;
private final MessagingMethodInvokerHelper delegate;
public MethodInvokingMessageProcessor(Object targetObject, Method method) {
this(targetObject, null, method);
delegate = new MessagingMethodInvokerHelper(targetObject, method, false);
}
public MethodInvokingMessageProcessor(Object targetObject, String methodName) {
this(targetObject, null, methodName);
delegate = new MessagingMethodInvokerHelper(targetObject, methodName, false);
}
public MethodInvokingMessageProcessor(Object targetObject, String methodName, boolean requiresReply) {
this(targetObject, null, methodName, requiresReply);
delegate = new MessagingMethodInvokerHelper(targetObject, methodName, Object.class, false);
}
public MethodInvokingMessageProcessor(Object targetObject, Class<? extends Annotation> annotationType) {
this(targetObject, annotationType, (String) null);
delegate = new MessagingMethodInvokerHelper(targetObject, annotationType, false);
}
/*
* Private constructors for internal use
*/
private MethodInvokingMessageProcessor(Object targetObject, Class<? extends Annotation> annotationType, Method method) {
Assert.notNull(method, "method must not be null");
HandlerMethod handlerMethod = new HandlerMethod(method);
Assert.notNull(targetObject, "targetObject must not be null");
this.targetObject = targetObject;
this.handlerMethods = Collections.<Class<?>, HandlerMethod>singletonMap(handlerMethod.getTargetParameterType(), handlerMethod);
this.prepareEvaluationContext(this.getEvaluationContext(), method, annotationType);
this.setDisplayString(targetObject, method);
@Override
public void setConversionService(ConversionService conversionService) {
super.setConversionService(conversionService);
delegate.setConversionService(conversionService);
}
private MethodInvokingMessageProcessor(Object targetObject, Class<? extends Annotation> annotationType, String methodName) {
this(targetObject, annotationType, methodName, false);
}
private MethodInvokingMessageProcessor(Object targetObject, Class<? extends Annotation> annotationType, String methodName, boolean requiresReply) {
Assert.notNull(targetObject, "targetObject must not be null");
this.targetObject = targetObject;
this.requiresReply = requiresReply;
this.handlerMethods = this.findHandlerMethodsForTarget(targetObject, annotationType, methodName, requiresReply);
this.prepareEvaluationContext(this.getEvaluationContext(), methodName, annotationType);
this.setDisplayString(targetObject, methodName);
}
private void setDisplayString(Object targetObject, Object targetMethod) {
StringBuilder sb = new StringBuilder(targetObject.getClass().getName());
if (targetMethod instanceof Method) {
sb.append("." + ((Method) targetMethod).getName());
}
else if (targetMethod instanceof String) {
sb.append("." + (String) targetMethod);
}
this.displayString = sb.toString() + "]";
}
private void prepareEvaluationContext(StandardEvaluationContext context, Object method, Class<? extends Annotation> annotationType) {
Class<?> targetType = AopUtils.getTargetClass(this.targetObject);
if (method instanceof Method) {
context.registerMethodFilter(targetType, new FixedHandlerMethodFilter((Method) method));
}
else if (method == null || method instanceof String) {
context.registerMethodFilter(targetType,
new HandlerMethodFilter(annotationType, (String) method, this.requiresReply));
}
context.addPropertyAccessor(new MapAccessor());
context.setVariable("target", targetObject);
}
public String toString() {
return this.displayString;
@Override
public void setBeanFactory(BeanFactory beanFactory) {
super.setBeanFactory(beanFactory);
delegate.setBeanFactory(beanFactory);
}
public Object processMessage(Message<?> message) {
Throwable evaluationException = null;
List<HandlerMethod> candidates = this.findHandlerMethodsForMessage(message);
for (HandlerMethod candidate : candidates) {
try {
Expression expression = candidate.getExpression();
Class<?> expectedType = candidate.method.getReturnType();
Object result = this.evaluateExpression(expression, message, expectedType);
if (this.requiresReply) {
Assert.notNull(result, "Expression evaluation result was null, but this processor requires a reply.");
}
return result;
}
catch (MessageHandlingException e) {
if (evaluationException == null) {
// keep the first exception
evaluationException = e.getCause();
}
}
}
throw new MessageHandlingException(message, "Failed to process Message.", evaluationException);
}
private Map<Class<?>, HandlerMethod> findHandlerMethodsForTarget(final Object targetObject,
final Class<? extends Annotation> annotationType, final String methodName, final boolean requiresReply) {
final Map<Class<?>, HandlerMethod> candidateMethods = new HashMap<Class<?>, HandlerMethod>();
final Map<Class<?>, HandlerMethod> fallbackMethods = new HashMap<Class<?>, HandlerMethod>();
final AtomicReference<Class<?>> ambiguousFallbackType = new AtomicReference<Class<?>>();
final Class<?> targetClass = this.getTargetClass(targetObject);
MethodFilter methodFilter = new UniqueMethodFilter(targetClass);
ReflectionUtils.doWithMethods(targetClass, new MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
boolean matchesAnnotation = false;
if (method.isBridge()) {
return;
}
if (isMethodDefinedOnObjectClass(method)) {
return;
}
if (!Modifier.isPublic(method.getModifiers())) {
return;
}
if (requiresReply && void.class.equals(method.getReturnType())) {
return;
}
if (methodName != null && !methodName.equals(method.getName())) {
return;
}
if (annotationType != null && AnnotationUtils.findAnnotation(method, annotationType) != null) {
matchesAnnotation = true;
}
HandlerMethod handlerMethod = null;
try {
handlerMethod = new HandlerMethod(method);
}
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug("Method [" + method + "] is not eligible for Message handling.", e);
}
return;
}
Class<?> targetParameterType = handlerMethod.getTargetParameterType();
if (matchesAnnotation || annotationType == null) {
Assert.isTrue(!candidateMethods.containsKey(targetParameterType),
"Found more than one method match for type [" + targetParameterType + "]");
candidateMethods.put(targetParameterType, handlerMethod);
}
else {
if (fallbackMethods.containsKey(targetParameterType)) {
// we need to check for duplicate type matches,
// but only if we end up falling back
// and we'll only keep track of the first one
ambiguousFallbackType.compareAndSet(null, targetParameterType);
}
fallbackMethods.put(targetParameterType, handlerMethod);
}
}
}, methodFilter);
if (!candidateMethods.isEmpty()) {
return candidateMethods;
}
Assert.notEmpty(fallbackMethods, "Target object of type [" + this.targetObject.getClass() +
"] has no eligible methods for handling Messages.");
Assert.isNull(ambiguousFallbackType.get(),
"Found more than one method match for type [" + ambiguousFallbackType + "]");
return fallbackMethods;
}
private Class<?> getTargetClass(Object targetObject) {
Class<?> targetClass = targetObject.getClass();
if (AopUtils.isAopProxy(targetObject)) {
targetClass = AopUtils.getTargetClass(targetObject);
}
else if(AopUtils.isCglibProxyClass(targetClass)) {
Class<?> superClass = targetObject.getClass().getSuperclass();
if (!Object.class.equals(superClass)) {
targetClass = superClass;
}
}
return targetClass;
}
private List<HandlerMethod> findHandlerMethodsForMessage(Message<?> message) {
final Class<?> payloadType = message.getPayload().getClass();
HandlerMethod closestMatch = this.findClosestMatch(payloadType);
if (closestMatch != null) {
return Collections.singletonList(closestMatch);
}
return new ArrayList<HandlerMethod>(this.handlerMethods.values());
}
private HandlerMethod findClosestMatch(Class<?> payloadType) {
Set<Class<?>> candidates = this.handlerMethods.keySet();
Class<?> match = null;
if (candidates != null && !candidates.isEmpty()) {
match = ClassUtils.findClosestMatch(payloadType, candidates, true);
}
return (match != null) ? this.handlerMethods.get(match) : null;
}
private static boolean isMethodDefinedOnObjectClass(Method method) {
if (method == null) {
return false;
}
if (method.getDeclaringClass().equals(Object.class)) {
return true;
}
if (ReflectionUtils.isEqualsMethod(method) ||
ReflectionUtils.isHashCodeMethod(method) ||
ReflectionUtils.isToStringMethod(method) ||
AopUtils.isFinalizeMethod(method)) {
return true;
}
return (method.getName().equals("clone") && method.getParameterTypes().length == 0);
}
/**
* Helper class for generating and exposing metadata for a candidate handler method.
* The metadata includes the SpEL expression and the expected payload type.
*/
private static class HandlerMethod {
private static final SpelExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
private static final ParameterNameDiscoverer PARAMETER_NAME_DISCOVERER =
new LocalVariableTableParameterNameDiscoverer();
private final Method method;
private final Expression expression;
private volatile Class<?> targetParameterType;
HandlerMethod(Method method) {
this.method = method;
this.expression = this.generateExpression(method);
}
Expression getExpression() {
return this.expression;
}
Class<?> getTargetParameterType() {
return this.targetParameterType;
}
public String toString() {
return this.method.toString();
}
private Expression generateExpression(Method method) {
StringBuilder sb = new StringBuilder("#target." + method.getName() + "(");
Class<?>[] parameterTypes = method.getParameterTypes();
Annotation[][] parameterAnnotations = method.getParameterAnnotations();
boolean hasUnqualifiedMapParameter = false;
for (int i = 0; i < parameterTypes.length; i++) {
if (i != 0) {
sb.append(", ");
}
Class<?> parameterType = parameterTypes[i];
Annotation mappingAnnotation = findMappingAnnotation(parameterAnnotations[i]);
if (mappingAnnotation != null) {
Class<? extends Annotation> annotationType = mappingAnnotation.annotationType();
if (annotationType.equals(Payload.class)) {
sb.append("payload");
String qualifierExpression = ((Payload) mappingAnnotation).value();
if (StringUtils.hasText(qualifierExpression)) {
sb.append("." + qualifierExpression);
}
if (!StringUtils.hasText(qualifierExpression)) {
this.setExclusiveTargetParameterType(parameterType);
}
}
else if (annotationType.equals(Headers.class)) {
Assert.isTrue(Map.class.isAssignableFrom(parameterType),
"The @Headers annotation can only be applied to a Map-typed parameter.");
sb.append("headers");
}
else if (annotationType.equals(Header.class)) {
Header headerAnnotation = (Header) mappingAnnotation;
sb.append(this.determineHeaderExpression(headerAnnotation, new MethodParameter(method, i)));
}
}
else if (Message.class.isAssignableFrom(parameterType)) {
sb.append("#root");
this.setExclusiveTargetParameterType(Message.class);
}
else if (Map.class.isAssignableFrom(parameterType)) {
if (Properties.class.isAssignableFrom(parameterType)) {
sb.append("payload instanceof T(java.util.Map) or " +
"(payload instanceof T(String) and payload.contains('=')) ? payload : headers");
}
else {
sb.append("(payload instanceof T(java.util.Map) ? payload : headers)");
}
Assert.isTrue(!hasUnqualifiedMapParameter,
"Found more than one Map typed parameter without any qualification. " +
"Consider using @Payload or @Headers on at least one of the parameters.");
hasUnqualifiedMapParameter = true;
}
else {
sb.append("payload");
this.setExclusiveTargetParameterType(parameterType);
}
}
if (hasUnqualifiedMapParameter) {
if (targetParameterType != null && Map.class.isAssignableFrom(this.targetParameterType)) {
throw new IllegalArgumentException(
"Unable to determine payload matching parameter due to ambiguous Map typed parameters. " +
"Consider adding the @Payload and or @Headers annotations as appropriate.");
}
}
sb.append(")");
if (this.targetParameterType == null) {
this.targetParameterType = Message.class;
}
return EXPRESSION_PARSER.parseExpression(sb.toString());
}
private Annotation findMappingAnnotation(Annotation[] annotations) {
if (annotations == null || annotations.length == 0) {
return null;
}
Annotation match = null;
for (Annotation annotation : annotations) {
Class<? extends Annotation> type = annotation.annotationType();
if (type.equals(Payload.class) || type.equals(Header.class) || type.equals(Headers.class)) {
if (match != null) {
throw new MessagingException("At most one parameter annotation can be provided for message mapping, " +
"but found two: [" + match.annotationType().getName() + "] and [" + annotation.annotationType().getName() + "]");
}
match = annotation;
}
}
return match;
}
private String determineHeaderExpression(Header headerAnnotation, MethodParameter methodParameter) {
methodParameter.initParameterNameDiscovery(PARAMETER_NAME_DISCOVERER);
String headerName = null;
String relativeExpression = "";
String valueAttribute = headerAnnotation.value();
if (!StringUtils.hasText(valueAttribute)) {
headerName = methodParameter.getParameterName();
}
else if (valueAttribute.indexOf('.') != -1) {
String tokens[] = valueAttribute.split("\\.", 2);
headerName = tokens[0];
if (StringUtils.hasText(tokens[1])) {
relativeExpression = "." + tokens[1];
}
}
else {
headerName = valueAttribute;
}
Assert.notNull(headerName, "Cannot determine header name. Possible reasons: -debug is " +
"disabled or header name is not explicitly provided via @Header annotation.");
String headerExpression = "headers." + headerName + relativeExpression;
return (headerAnnotation.required()) ? headerExpression
: "headers['" + headerName + "'] != null ? " + headerExpression + " : null";
}
private synchronized void setExclusiveTargetParameterType(Class<?> targetParameterType) {
Assert.isNull(this.targetParameterType, "Found more than one parameter type candidate: [" +
this.targetParameterType + "] and [" + targetParameterType + "]");
this.targetParameterType = targetParameterType;
}
}
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
private static class UniqueMethodFilter implements MethodFilter {
private List<Method> uniqueMethods = new ArrayList<Method>();
public UniqueMethodFilter(Class<?> targetClass) {
ArrayList<Method> allMethods = new ArrayList<Method>(Arrays.asList(targetClass.getMethods()));
for (Method method : allMethods) {
uniqueMethods.add(org.springframework.util.ClassUtils.getMostSpecificMethod(method, targetClass));
}
}
public boolean matches(Method method) {
return uniqueMethods.contains(method);
try {
return delegate.process(message);
} catch (Exception e) {
throw new MessageHandlingException(message, e);
}
}

84
spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java Normal file
View File

@@ -0,0 +1,84 @@
/*
* 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.util;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.convert.ConversionService;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.context.SimpleBeanResolver;
/**
* @author Mark Fisher
* @since 2.0
*/
public abstract class AbstractExpressionEvaluator implements BeanFactoryAware {
private final StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
private final BeanFactoryTypeConverter typeConverter = new BeanFactoryTypeConverter();
public AbstractExpressionEvaluator() {
evaluationContext.setTypeConverter(typeConverter);
}
/**
* Specify a BeanFactory in order to enable resolution via <code>@beanName</code> in the expression.
*/
public void setBeanFactory(final BeanFactory beanFactory) {
if (beanFactory != null) {
typeConverter.setBeanFactory(beanFactory);
this.getEvaluationContext().setBeanResolver(new SimpleBeanResolver(beanFactory));
}
}
public void setConversionService(ConversionService conversionService) {
if (conversionService != null) {
typeConverter.setConversionService(conversionService);
}
}
// TODO: does this need to be public?
public StandardEvaluationContext getEvaluationContext() {
return this.evaluationContext;
}
protected Object evaluateExpression(Expression expression, Message<?> message, Class<?> expectedType) {
try {
return evaluateExpression(expression, (Object) message, expectedType);
}
catch (EvaluationException e) {
Throwable cause = e.getCause();
throw new MessageHandlingException(message, "Expression evaluation failed: "
+ expression.getExpressionString(), cause == null ? e : cause);
}
catch (Exception e) {
throw new MessageHandlingException(message, "Expression evaluation failed: "
+ expression.getExpressionString(), e);
}
}
protected Object evaluateExpression(Expression expression, Object message, Class<?> expectedType) {
return (expectedType != null) ? expression.getValue(this.evaluationContext, message, expectedType) : expression
.getValue(this.evaluationContext, message);
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/handler/HandlerMethodFilter.java → spring-integration-core/src/main/java/org/springframework/integration/util/AnnotatedMethodFilter.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.handler;
package org.springframework.integration.util;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
@@ -37,7 +37,7 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
* @since 2.0
*/
class HandlerMethodFilter implements MethodFilter {
public class AnnotatedMethodFilter implements MethodFilter {
private final Class<? extends Annotation> annotationType;
@@ -46,7 +46,7 @@ class HandlerMethodFilter implements MethodFilter {
private final boolean requiresReply;
public HandlerMethodFilter(Class<? extends Annotation> annotationType, String methodName, boolean requiresReply) {
public AnnotatedMethodFilter(Class<? extends Annotation> annotationType, String methodName, boolean requiresReply) {
this.annotationType = annotationType;
this.methodName = methodName;
this.requiresReply = requiresReply;

6
spring-integration-core/src/main/java/org/springframework/integration/handler/FixedHandlerMethodFilter.java → spring-integration-core/src/main/java/org/springframework/integration/util/FixedMethodFilter.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.handler;
package org.springframework.integration.util;
import java.lang.reflect.Method;
import java.util.Collections;
@@ -32,12 +32,12 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @since 2.0
*/
class FixedHandlerMethodFilter implements MethodFilter {
public class FixedMethodFilter implements MethodFilter {
private final Method method;
public FixedHandlerMethodFilter(Method method) {
public FixedMethodFilter(Method method) {
Assert.notNull(method, "method must not be null");
this.method = method;
}

621
spring-integration-core/src/main/java/org/springframework/integration/util/MessagingMethodInvokerHelper.java Normal file
View File

@@ -0,0 +1,621 @@
/*
* 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.util;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.context.expression.MapAccessor;
import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
import org.springframework.core.MethodParameter;
import org.springframework.core.ParameterNameDiscoverer;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.expression.TypeConverter;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.annotation.Headers;
import org.springframework.integration.annotation.Payload;
import org.springframework.integration.annotation.Payloads;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.util.ReflectionUtils.MethodCallback;
import org.springframework.util.ReflectionUtils.MethodFilter;
/**
* A helper class for processors that invoke a method on a target Object using a combination of message payload(s) and
* headers as arguments. The Method instance or method name may be provided as a constructor argument. If a method name
* is provided, and more than one declared method has that name, the method-selection will be dynamic, based on the
* underlying SpEL method resolution. Alternatively, an annotation type may be provided so that the candidates for
* SpEL's method resolution are determined by the presence of that annotation rather than the method name.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Dave Syer
*
* @since 2.0
*/
public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator {
private final Log logger = LogFactory.getLog(this.getClass());
private final Object targetObject;
private volatile String displayString;
private volatile boolean requiresReply;
private final Map<Class<?>, HandlerMethod> handlerMethods;
private final Class<?> expectedType;
private final boolean canProcessMessageList;
public MessagingMethodInvokerHelper(Object targetObject, Method method, Class<?> expectedType,
boolean canProcessMessageList) {
this(targetObject, null, method, expectedType, canProcessMessageList);
}
public MessagingMethodInvokerHelper(Object targetObject, Method method, boolean canProcessMessageList) {
this(targetObject, method, null, canProcessMessageList);
}
public MessagingMethodInvokerHelper(Object targetObject, String methodName, Class<?> expectedType,
boolean canProcessMessageList) {
this(targetObject, null, methodName, expectedType, canProcessMessageList);
}
public MessagingMethodInvokerHelper(Object targetObject, String methodName, boolean canProcessMessageList) {
this(targetObject, methodName, null, canProcessMessageList);
}
public MessagingMethodInvokerHelper(Object targetObject, Class<? extends Annotation> annotationType,
boolean canProcessMessageList) {
this(targetObject, annotationType, null, canProcessMessageList);
}
public MessagingMethodInvokerHelper(Object targetObject, Class<? extends Annotation> annotationType, Class<?> expectedType,
boolean canProcessMessageList) {
this(targetObject, annotationType, (String) null, expectedType, canProcessMessageList);
}
public Object process(Message<?> message) throws Exception {
ParametersWrapper parameters = new ParametersWrapper(message);
return processInternal(parameters);
}
public Object process(Collection<Message<?>> messages, Map<String, ?> headers) throws Exception {
ParametersWrapper parameters = new ParametersWrapper(messages, headers);
return processInternal(parameters);
}
public String toString() {
return this.displayString;
}
/*
* Private constructors for internal use
*/
private MessagingMethodInvokerHelper(Object targetObject, Class<? extends Annotation> annotationType,
Method method, Class<?> expectedType, boolean canProcessMessageList) {
this.canProcessMessageList = canProcessMessageList;
Assert.notNull(method, "method must not be null");
this.expectedType = expectedType;
this.requiresReply = expectedType != null;
if (expectedType != null) {
Assert.isTrue(method.getReturnType() != Void.class && method.getReturnType() != Void.TYPE,
"method must have a return type");
}
HandlerMethod handlerMethod = new HandlerMethod(method, canProcessMessageList);
Assert.notNull(targetObject, "targetObject must not be null");
this.targetObject = targetObject;
this.handlerMethods = Collections.<Class<?>, HandlerMethod> singletonMap(handlerMethod.getTargetParameterType()
.getObjectType(), handlerMethod);
this.prepareEvaluationContext(this.getEvaluationContext(), method, annotationType);
this.setDisplayString(targetObject, method);
}
private MessagingMethodInvokerHelper(Object targetObject, Class<? extends Annotation> annotationType,
String methodName, Class<?> expectedType, boolean canProcessMessageList) {
this.canProcessMessageList = canProcessMessageList;
Assert.notNull(targetObject, "targetObject must not be null");
this.expectedType = expectedType;
this.targetObject = targetObject;
this.requiresReply = expectedType != null;
this.handlerMethods = this.findHandlerMethodsForTarget(targetObject, annotationType, methodName, requiresReply);
this.prepareEvaluationContext(this.getEvaluationContext(), methodName, annotationType);
this.setDisplayString(targetObject, methodName);
}
private void setDisplayString(Object targetObject, Object targetMethod) {
StringBuilder sb = new StringBuilder(targetObject.getClass().getName());
if (targetMethod instanceof Method) {
sb.append("." + ((Method) targetMethod).getName());
}
else if (targetMethod instanceof String) {
sb.append("." + (String) targetMethod);
}
this.displayString = sb.toString() + "]";
}
private void prepareEvaluationContext(StandardEvaluationContext context, Object method,
Class<? extends Annotation> annotationType) {
Class<?> targetType = AopUtils.getTargetClass(this.targetObject);
if (method instanceof Method) {
context.registerMethodFilter(targetType, new FixedMethodFilter((Method) method));
if (expectedType != null) {
Assert.state(context.getTypeConverter().canConvert(((Method) method).getReturnType(), expectedType),
"Cannot convert to expected type (" + expectedType + ") from " + method);
}
}
else if (method == null || method instanceof String) {
AnnotatedMethodFilter filter = new AnnotatedMethodFilter(annotationType, (String) method,
this.requiresReply);
Assert.state(canReturnExpectedType(filter, targetType, context.getTypeConverter()),
"Cannot convert to expected type (" + expectedType + ") from " + method);
context.registerMethodFilter(targetType, filter);
}
context.addPropertyAccessor(new MapAccessor());
context.setVariable("target", targetObject);
}
private boolean canReturnExpectedType(AnnotatedMethodFilter filter, Class<?> targetType, TypeConverter typeConverter) {
if (expectedType == null) {
return true;
}
List<Method> methods = filter.filter(Arrays.asList(ReflectionUtils.getAllDeclaredMethods(targetType)));
for (Method method : methods) {
if (typeConverter.canConvert(method.getReturnType(), expectedType)) {
return true;
}
}
return false;
}
private Object processInternal(ParametersWrapper parameters) throws Exception {
Throwable evaluationException = null;
List<HandlerMethod> candidates = this.findHandlerMethodsForParameters(parameters);
Assert.state(!candidates.isEmpty(), "No candidate methods found for messages.");
for (HandlerMethod candidate : candidates) {
try {
Expression expression = candidate.getExpression();
Class<?> expectedType = this.expectedType != null ? this.expectedType : candidate.method
.getReturnType();
Object result = this.evaluateExpression(expression, parameters, expectedType);
if (this.requiresReply) {
Assert.notNull(result,
"Expression evaluation result was null, but this processor requires a reply.");
}
return result;
}
// keep the first exception
catch (EvaluationException e) {
if (evaluationException == null) {
evaluationException = e.getCause();
}
if (evaluationException == null) {
evaluationException = e;
}
}
catch (MessageHandlingException e) {
if (evaluationException == null) {
evaluationException = e.getCause();
}
if (evaluationException == null) {
evaluationException = e;
}
}
catch (Exception e) {
if (evaluationException == null) {
evaluationException = e;
}
}
}
if (evaluationException instanceof Exception) {
throw (Exception) evaluationException;
}
else if (evaluationException instanceof Error) {
throw (Error) evaluationException;
}
else {
throw new IllegalStateException("Cannot process message", evaluationException);
}
}
private Map<Class<?>, HandlerMethod> findHandlerMethodsForTarget(final Object targetObject,
final Class<? extends Annotation> annotationType, final String methodName, final boolean requiresReply) {
final Map<Class<?>, HandlerMethod> candidateMethods = new HashMap<Class<?>, HandlerMethod>();
final Map<Class<?>, HandlerMethod> fallbackMethods = new HashMap<Class<?>, HandlerMethod>();
final AtomicReference<Class<?>> ambiguousFallbackType = new AtomicReference<Class<?>>();
final Class<?> targetClass = this.getTargetClass(targetObject);
MethodFilter methodFilter = new UniqueMethodFilter(targetClass);
ReflectionUtils.doWithMethods(targetClass, new MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
boolean matchesAnnotation = false;
if (method.isBridge()) {
return;
}
if (isMethodDefinedOnObjectClass(method)) {
return;
}
if (!Modifier.isPublic(method.getModifiers())) {
return;
}
if (requiresReply && void.class.equals(method.getReturnType())) {
return;
}
if (methodName != null && !methodName.equals(method.getName())) {
return;
}
if (annotationType != null && AnnotationUtils.findAnnotation(method, annotationType) != null) {
matchesAnnotation = true;
}
HandlerMethod handlerMethod = null;
try {
handlerMethod = new HandlerMethod(method, canProcessMessageList);
}
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug("Method [" + method + "] is not eligible for Message handling.", e);
}
return;
}
Class<?> targetParameterType = handlerMethod.getTargetParameterType().getObjectType();
if (matchesAnnotation || annotationType == null) {
Assert.isTrue(!candidateMethods.containsKey(targetParameterType),
"Found more than one method match for type [" + targetParameterType + "]");
candidateMethods.put(targetParameterType, handlerMethod);
}
else {
if (fallbackMethods.containsKey(targetParameterType)) {
// we need to check for duplicate type matches,
// but only if we end up falling back
// and we'll only keep track of the first one
ambiguousFallbackType.compareAndSet(null, targetParameterType);
}
fallbackMethods.put(targetParameterType, handlerMethod);
}
}
}, methodFilter);
if (!candidateMethods.isEmpty()) {
return candidateMethods;
}
Assert.notEmpty(fallbackMethods, "Target object of type [" + this.targetObject.getClass()
+ "] has no eligible methods for handling Messages.");
Assert.isNull(ambiguousFallbackType.get(), "Found ambiguous parameter type [" + ambiguousFallbackType
+ "] for method match: " + fallbackMethods.values());
return fallbackMethods;
}
private Class<?> getTargetClass(Object targetObject) {
Class<?> targetClass = targetObject.getClass();
if (AopUtils.isAopProxy(targetObject)) {
targetClass = AopUtils.getTargetClass(targetObject);
}
else if (AopUtils.isCglibProxyClass(targetClass)) {
Class<?> superClass = targetObject.getClass().getSuperclass();
if (!Object.class.equals(superClass)) {
targetClass = superClass;
}
}
return targetClass;
}
private List<HandlerMethod> findHandlerMethodsForParameters(ParametersWrapper parameters) {
final Class<?> payloadType = parameters.getFirstParameterType();
HandlerMethod closestMatch = this.findClosestMatch(payloadType);
if (closestMatch != null) {
return Collections.singletonList(closestMatch);
}
return new ArrayList<HandlerMethod>(this.handlerMethods.values());
}
private HandlerMethod findClosestMatch(Class<?> payloadType) {
Set<Class<?>> candidates = this.handlerMethods.keySet();
Class<?> match = null;
if (candidates != null && !candidates.isEmpty()) {
match = ClassUtils.findClosestMatch(payloadType, candidates, true);
}
return (match != null) ? this.handlerMethods.get(match) : null;
}
private static boolean isMethodDefinedOnObjectClass(Method method) {
if (method == null) {
return false;
}
if (method.getDeclaringClass().equals(Object.class)) {
return true;
}
if (ReflectionUtils.isEqualsMethod(method) || ReflectionUtils.isHashCodeMethod(method)
|| ReflectionUtils.isToStringMethod(method) || AopUtils.isFinalizeMethod(method)) {
return true;
}
return (method.getName().equals("clone") && method.getParameterTypes().length == 0);
}
/**
* Helper class for generating and exposing metadata for a candidate handler method. The metadata includes the SpEL
* expression and the expected payload type.
*/
private static class HandlerMethod {
private static final SpelExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
private static final ParameterNameDiscoverer PARAMETER_NAME_DISCOVERER = new LocalVariableTableParameterNameDiscoverer();
private final Method method;
private final Expression expression;
private volatile TypeDescriptor targetParameterType;
private static final TypeDescriptor messageTypeDescriptor = TypeDescriptor.valueOf(Message.class);
private static final TypeDescriptor messageListTypeDescriptor = new TypeDescriptor(ReflectionUtils.findField(
HandlerMethod.class, "dummyMessages"));
private static final TypeDescriptor messageArrayTypeDescriptor = TypeDescriptor.valueOf(Message[].class);
@SuppressWarnings("unused")
private static final List<Message<?>> dummyMessages = Collections.emptyList();
private final boolean canProcessMessageList;
HandlerMethod(Method method, boolean canProcessMessageList) {
this.method = method;
this.canProcessMessageList = canProcessMessageList;
this.expression = this.generateExpression(method);
}
Expression getExpression() {
return this.expression;
}
TypeDescriptor getTargetParameterType() {
return this.targetParameterType;
}
public String toString() {
return this.method.toString();
}
private Expression generateExpression(Method method) {
StringBuilder sb = new StringBuilder("#target." + method.getName() + "(");
Class<?>[] parameterTypes = method.getParameterTypes();
Annotation[][] parameterAnnotations = method.getParameterAnnotations();
boolean hasUnqualifiedMapParameter = false;
TypeDescriptor defaultParameterTypeDescriptor = TypeDescriptor.valueOf(List.class);
for (int i = 0; i < parameterTypes.length; i++) {
if (i != 0) {
sb.append(", ");
}
TypeDescriptor parameterTypeDescriptor = new TypeDescriptor(new MethodParameter(method, i));
defaultParameterTypeDescriptor = parameterTypeDescriptor;
Class<?> parameterType = parameterTypeDescriptor.getObjectType();
Annotation mappingAnnotation = findMappingAnnotation(parameterAnnotations[i]);
if (mappingAnnotation != null) {
Class<? extends Annotation> annotationType = mappingAnnotation.annotationType();
if (annotationType.equals(Payload.class)) {
sb.append("payload");
String qualifierExpression = ((Payload) mappingAnnotation).value();
if (StringUtils.hasText(qualifierExpression)) {
sb.append("." + qualifierExpression);
}
if (!StringUtils.hasText(qualifierExpression)) {
this.setExclusiveTargetParameterType(parameterTypeDescriptor);
}
}
if (annotationType.equals(Payloads.class)) {
sb.append("messages.![payload");
String qualifierExpression = ((Payloads) mappingAnnotation).value();
if (StringUtils.hasText(qualifierExpression)) {
sb.append("." + qualifierExpression);
}
sb.append("]");
if (!StringUtils.hasText(qualifierExpression)) {
this.setExclusiveTargetParameterType(parameterTypeDescriptor);
}
}
else if (annotationType.equals(Headers.class)) {
Assert.isTrue(Map.class.isAssignableFrom(parameterType),
"The @Headers annotation can only be applied to a Map-typed parameter.");
sb.append("headers");
}
else if (annotationType.equals(Header.class)) {
Header headerAnnotation = (Header) mappingAnnotation;
sb.append(this.determineHeaderExpression(headerAnnotation, new MethodParameter(method, i)));
}
}
else if (parameterTypeDescriptor != null
&& parameterTypeDescriptor.isAssignableTo(messageTypeDescriptor)) {
sb.append("message");
this.setExclusiveTargetParameterType(parameterTypeDescriptor);
}
else if (parameterTypeDescriptor != null
&& (parameterTypeDescriptor.isAssignableTo(messageListTypeDescriptor) || parameterTypeDescriptor
.isAssignableTo(messageArrayTypeDescriptor))) {
sb.append("messages");
this.setExclusiveTargetParameterType(parameterTypeDescriptor);
}
else if (Collection.class.isAssignableFrom(parameterType) || parameterType.isArray()) {
if (canProcessMessageList) {
sb.append("messages.![payload]");
} else {
sb.append("payload");
}
this.setExclusiveTargetParameterType(parameterTypeDescriptor);
}
else if (Map.class.isAssignableFrom(parameterType)) {
if (Properties.class.isAssignableFrom(parameterType)) {
sb.append("payload instanceof T(java.util.Map) or "
+ "(payload instanceof T(String) and payload.contains('=')) ? payload : headers");
}
else {
sb.append("(payload instanceof T(java.util.Map) ? payload : headers)");
}
Assert.isTrue(!hasUnqualifiedMapParameter,
"Found more than one Map typed parameter without any qualification. "
+ "Consider using @Payload or @Headers on at least one of the parameters.");
hasUnqualifiedMapParameter = true;
}
else {
sb.append("payload");
this.setExclusiveTargetParameterType(parameterTypeDescriptor);
}
}
if (hasUnqualifiedMapParameter) {
if (targetParameterType != null && Map.class.isAssignableFrom(this.targetParameterType.getObjectType())) {
throw new IllegalArgumentException(
"Unable to determine payload matching parameter due to ambiguous Map typed parameters. "
+ "Consider adding the @Payload and or @Headers annotations as appropriate.");
}
}
sb.append(")");
if (this.targetParameterType == null) {
this.targetParameterType = defaultParameterTypeDescriptor;
}
return EXPRESSION_PARSER.parseExpression(sb.toString());
}
private Annotation findMappingAnnotation(Annotation[] annotations) {
if (annotations == null || annotations.length == 0) {
return null;
}
Annotation match = null;
for (Annotation annotation : annotations) {
Class<? extends Annotation> type = annotation.annotationType();
if (type.equals(Payload.class) || type.equals(Header.class) || type.equals(Headers.class)) {
if (match != null) {
throw new MessagingException(
"At most one parameter annotation can be provided for message mapping, "
+ "but found two: [" + match.annotationType().getName() + "] and ["
+ annotation.annotationType().getName() + "]");
}
match = annotation;
}
}
return match;
}
private String determineHeaderExpression(Header headerAnnotation, MethodParameter methodParameter) {
methodParameter.initParameterNameDiscovery(PARAMETER_NAME_DISCOVERER);
String headerName = null;
String relativeExpression = "";
String valueAttribute = headerAnnotation.value();
if (!StringUtils.hasText(valueAttribute)) {
headerName = methodParameter.getParameterName();
}
else if (valueAttribute.indexOf('.') != -1) {
String tokens[] = valueAttribute.split("\\.", 2);
headerName = tokens[0];
if (StringUtils.hasText(tokens[1])) {
relativeExpression = "." + tokens[1];
}
}
else {
headerName = valueAttribute;
}
Assert.notNull(headerName, "Cannot determine header name. Possible reasons: -debug is "
+ "disabled or header name is not explicitly provided via @Header annotation.");
String headerExpression = "headers." + headerName + relativeExpression;
return (headerAnnotation.required()) ? headerExpression : "headers['" + headerName + "'] != null ? "
+ headerExpression + " : null";
}
private synchronized void setExclusiveTargetParameterType(TypeDescriptor targetParameterType) {
Assert.isNull(this.targetParameterType, "Found more than one parameter type candidate: ["
+ this.targetParameterType + "] and [" + targetParameterType + "]");
this.targetParameterType = targetParameterType;
}
}
@SuppressWarnings("unused")
private static class ParametersWrapper {
private final Object payload;
private final Collection<Message<?>> messages;
private final Map<String, ?> headers;
private final Message<?> message;
public ParametersWrapper(Message<?> message) {
this.message = message;
this.payload = message.getPayload();
this.headers = message.getHeaders();
this.messages = null;
}
public ParametersWrapper(Collection<Message<?>> messages, Map<String, ?> headers) {
this.payload = null;
this.messages = messages;
this.headers = headers;
this.message = null;
}
public Object getPayload() {
Assert.state(payload != null, "Invalid method parameter for payload: was expecting collection.");
return payload;
}
public Collection<Message<?>> getMessages() {
Assert.state(messages != null, "Invalid method parameter for messages: was expecting a single payload.");
return messages;
}
public Map<String, ?> getHeaders() {
return headers;
}
public Message<?> getMessage() {
return message;
}
public Class<?> getFirstParameterType() {
if (payload != null) {
return payload.getClass();
}
return Collection.class;
}
}
}

28
spring-integration-core/src/main/java/org/springframework/integration/util/UniqueMethodFilter.java Normal file
View File

@@ -0,0 +1,28 @@
package org.springframework.integration.util;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.springframework.util.ReflectionUtils.MethodFilter;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
public class UniqueMethodFilter implements MethodFilter {
private List<Method> uniqueMethods = new ArrayList<Method>();
public UniqueMethodFilter(Class<?> targetClass) {
ArrayList<Method> allMethods = new ArrayList<Method>(Arrays.asList(targetClass.getMethods()));
for (Method method : allMethods) {
uniqueMethods.add(org.springframework.util.ClassUtils.getMostSpecificMethod(method, targetClass));
}
}
public boolean matches(Method method) {
return uniqueMethods.contains(method);
}
}

6
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java
View File

@@ -31,7 +31,7 @@ import org.springframework.integration.MessageHeaders;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageBuilder;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageStore;
@@ -231,7 +231,7 @@ public class AggregatorTests {
}
private class MultiplyingProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate,
public void processAndSend(MessageGroup group, MessagingOperations messagingTemplate,
MessageChannel outputChannel) {
Integer product = 1;
for (Message<?> message : group.getUnmarked()) {
@@ -242,7 +242,7 @@ public class AggregatorTests {
}
private class NullReturningMessageProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group, MessagingTemplate messagingTemplate,
public void processAndSend(MessageGroup group, MessagingOperations messagingTemplate,
MessageChannel outputChannel) {
// noop
}

6
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java
View File

@@ -37,7 +37,7 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageBuilder;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
@@ -344,7 +344,7 @@ public class ConcurrentAggregatorTests {
private class MultiplyingProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group,
MessagingTemplate messagingTemplate,
MessagingOperations messagingTemplate,
MessageChannel outputChannel) {
Integer product = 1;
for (Message<?> message : group.getUnmarked()) {
@@ -356,7 +356,7 @@ public class ConcurrentAggregatorTests {
private class NullReturningMessageProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group,
MessagingTemplate messagingTemplate,
MessagingOperations messagingTemplate,
MessageChannel outputChannel) {
// noop
}

2
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java → spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java
View File

@@ -34,7 +34,7 @@ import static org.mockito.Mockito.*;
* @author Iwein Fuld
*/
@RunWith(MockitoJUnitRunner.class)
public class CorrelatingMessageBarrierTest {
public class CorrelatingMessageBarrierTests {
private CorrelatingMessageBarrier barrier;
@Mock

474
spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java
View File

@@ -25,29 +25,38 @@ import static org.mockito.Matchers.isA;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.ConversionServiceFactory;
import org.springframework.core.convert.support.GenericConversionService;
import org.springframework.integration.Message;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.annotation.Headers;
import org.springframework.integration.annotation.Payloads;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageBuilder;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.StringMessage;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageGroup;
@RunWith(MockitoJUnitRunner.class)
public class MethodInvokingMessageGroupProcessorTests {
@@ -70,27 +79,28 @@ public class MethodInvokingMessageGroupProcessorTests {
messagesUpForProcessing.add(MessageBuilder.withPayload(4).build());
}
@SuppressWarnings("unused")
private class AnnotatedAggregatorMethod {
@Aggregator
public Integer and(List<Integer> flags) {
int result = 0;
for (Integer flag : flags) {
result = result | flag;
}
return result;
}
public String know(List<Integer> flags) {
return "I'm not the one ";
}
}
@SuppressWarnings("unchecked")
@Test
public void shouldFindAnnotatedAggregatorMethod() throws Exception {
@SuppressWarnings("unused")
class AnnotatedAggregatorMethod {
@Aggregator
public Integer and(List<Integer> flags) {
int result = 0;
for (Integer flag : flags) {
result = result | flag;
}
return result;
}
public String know(List<Integer> flags) {
return "I'm not the one ";
}
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new AnnotatedAggregatorMethod());
@SuppressWarnings("unchecked")
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
@@ -100,21 +110,22 @@ public class MethodInvokingMessageGroupProcessorTests {
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
}
@SuppressWarnings("unused")
private class SimpleAggregator {
public Integer and(List<Integer> flags) {
int result = 0;
for (Integer flag : flags) {
result = result | flag;
}
return result;
}
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethod() throws Exception {
@SuppressWarnings("unused")
class SimpleAggregator {
public Integer and(List<Integer> flags) {
int result = 0;
for (Integer flag : flags) {
result = result | flag;
}
return result;
}
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
@SuppressWarnings("unchecked")
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
@@ -124,31 +135,169 @@ public class MethodInvokingMessageGroupProcessorTests {
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
}
@SuppressWarnings("unused")
private class UnannotatedAggregator {
public Integer and(List<Integer> flags) {
int result = 0;
for (Integer flag : flags) {
result = result | flag;
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethodForMessages() throws Exception {
@SuppressWarnings("unused")
class SimpleAggregator {
public Integer and(List<Message<Integer>> flags) {
int result = 0;
for (Message<Integer> flag : flags) {
result = result | flag.getPayload();
}
return result;
}
return result;
}
public void voidMethodShouldBeIgnored(List<Integer> flags) {
fail("this method should not be invoked");
}
public String methodAcceptingNoCollectionShouldBeIgnored(@Header String irrelevant) {
fail("this method should not be invoked");
return null;
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindAnnotatedPayloads() throws Exception {
@SuppressWarnings("unused")
class SimpleAggregator {
public String and(@Payloads List<Integer> flags, @Header("foo") List<Integer> header) {
List<Integer> result = new ArrayList<Integer>();
for (int flag : flags) {
result.add(flag);
}
for (int flag : header) {
result.add(flag);
}
return result.toString();
}
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
messagesUpForProcessing.add(MessageBuilder.withPayload(3).setHeader("foo", Arrays.asList(101, 102)).build());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((String) messageCaptor.getValue().getPayload(), is("[1, 2, 4, 3, 101, 102]"));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethodWithCollection() throws Exception {
@SuppressWarnings("unused")
class SimpleAggregator {
public Integer and(Collection<Integer> flags) {
int result = 0;
for (Integer flag : flags) {
result = result | flag;
}
return result;
}
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethodWithArray() throws Exception {
@SuppressWarnings("unused")
class SimpleAggregator {
public Integer and(int[] flags) {
int result = 0;
for (int flag : flags) {
result = result | flag;
}
return result;
}
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
}
@Ignore("INT-938: it probably should work if there is a converter registered, but maybe a SpEL bug?")
@Test
@SuppressWarnings("unchecked")
public void shouldFindSimpleAggregatorMethodWithIterator() throws Exception {
@SuppressWarnings("unused")
class SimpleAggregator {
public Integer and(Iterator<Integer> flags) {
int result = 0;
for (int flag = flags.next(); flags.hasNext();) {
result = result | flag;
}
return result;
}
}
MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
GenericConversionService conversionService = ConversionServiceFactory.createDefaultConversionService();
conversionService.addConverter(new Converter<ArrayList<?>, Iterator<?>>() {
public Iterator<?> convert(ArrayList<?> source) {
return source.iterator();
}
});
processor.setConversionService(conversionService);
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
}
@Test
@SuppressWarnings("unchecked")
public void shouldFindFittingMethodAmongMultipleUnannotated() {
@SuppressWarnings("unused")
class UnannotatedAggregator {
public Integer and(List<Integer> flags) {
int result = 0;
for (Integer flag : flags) {
result = result | flag;
}
return result;
}
public void voidMethodShouldBeIgnored(List<Integer> flags) {
fail("this method should not be invoked");
}
public String methodAcceptingNoCollectionShouldBeIgnored(String irrelevant) {
fail("this method should not be invoked");
return null;
}
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new UnannotatedAggregator());
@SuppressWarnings("unchecked")
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
@@ -159,69 +308,168 @@ public class MethodInvokingMessageGroupProcessorTests {
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
}
@SuppressWarnings("unused")
private class AnnotatedParametersAggregator {
public Integer and(List<Integer> flags) {
int result = 0;
for (Integer flag : flags) {
result = result | flag;
@Test(expected = IllegalArgumentException.class)
public void testTwoMethodsWithSameParameterTypesAmbiguous() {
@SuppressWarnings("unused")
class AnnotatedParametersAggregator {
public Integer and(List<Integer> flags) {
int result = 0;
for (Integer flag : flags) {
result = result | flag;
}
return result;
}
public String listHeaderShouldBeIgnored(@Header List<Integer> flags) {
fail("this method should not be invoked");
return "";
}
return result;
}
public String listHeaderShouldBeIgnored(@Header List<Integer> flags) {
fail("this method should not be invoked");
return "";
}
}
new MethodInvokingMessageGroupProcessor(new AnnotatedParametersAggregator());
@Test
public void shouldFindFittingMethodAmongMultipleWithAnnotatedParameters() {
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new AnnotatedParametersAggregator());
@SuppressWarnings("unchecked")
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
when(outputChannel.send(isA(Message.class))).thenReturn(true);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
processor.processAndSend(messageGroupMock, messagingTemplate, outputChannel);
// verify
verify(messagingTemplate).send(eq(outputChannel), messageCaptor.capture());
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
}
@Test
public void singleAnnotation() throws Exception {
@SuppressWarnings("unused")
class SingleAnnotationTestBean {
@Aggregator
public String method1(List<String> input) {
return input.get(0);
}
public String method2(List<String> input) {
return input.get(1);
}
}
SingleAnnotationTestBean bean = new SingleAnnotationTestBean();
MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(bean);
Method method = this.getMethod(aggregator);
Method expected = SingleAnnotationTestBean.class.getMethod("method1", new Class[] { List.class });
assertEquals(expected, method);
SimpleMessageGroup group = new SimpleMessageGroup("FOO");
group.add(new StringMessage("foo"));
group.add(new StringMessage("bar"));
assertEquals("foo", aggregator.aggregatePayloads(group, null));
}
@Test
public void testHeaderParameters() throws Exception {
@SuppressWarnings("unused")
class SingleAnnotationTestBean {
@Aggregator
public String method1(List<String> input, @Header("foo") String foo) {
return input.get(0) + foo;
}
}
SingleAnnotationTestBean bean = new SingleAnnotationTestBean();
MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(bean);
SimpleMessageGroup group = new SimpleMessageGroup("FOO");
group.add(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build());
group.add(MessageBuilder.withPayload("bar").setHeader("foo", "bar").build());
assertEquals("foobar", aggregator.aggregatePayloads(group, aggregator.aggregateHeaders(group)));
}
@Test
public void testHeadersParameters() throws Exception {
@SuppressWarnings("unused")
class SingleAnnotationTestBean {
@Aggregator
public String method1(List<String> input, @Headers Map<String, ?> map) {
return input.get(0) + map.get("foo");
}
}
SingleAnnotationTestBean bean = new SingleAnnotationTestBean();
MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(bean);
SimpleMessageGroup group = new SimpleMessageGroup("FOO");
group.add(MessageBuilder.withPayload("foo").setHeader("foo", "bar").build());
group.add(MessageBuilder.withPayload("bar").setHeader("foo", "bar").build());
assertEquals("foobar", aggregator.aggregatePayloads(group, aggregator.aggregateHeaders(group)));
}
@Test(expected = IllegalArgumentException.class)
public void multipleAnnotations() {
@SuppressWarnings("unused")
class MultipleAnnotationTestBean {
@Aggregator
public String method1(List<String> input) {
return input.get(0);
}
@Aggregator
public String method2(List<String> input) {
return input.get(0);
}
}
MultipleAnnotationTestBean bean = new MultipleAnnotationTestBean();
new MethodInvokingMessageGroupProcessor(bean);
}
@Test
public void noAnnotations() throws Exception {
@SuppressWarnings("unused")
class NoAnnotationTestBean {
public String method1(List<String> input) {
return input.get(0);
}
String method2(List<String> input) {
return input.get(1);
}
}
NoAnnotationTestBean bean = new NoAnnotationTestBean();
MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(bean);
Method method = this.getMethod(aggregator);
Method expected = NoAnnotationTestBean.class.getMethod("method1", new Class[] { List.class });
assertEquals(expected, method);
SimpleMessageGroup group = new SimpleMessageGroup("FOO");
group.add(new StringMessage("foo"));
group.add(new StringMessage("bar"));
assertEquals("foo", aggregator.aggregatePayloads(group, null));
}
@Test(expected = IllegalArgumentException.class)
public void multiplePublicMethods() {
@SuppressWarnings("unused")
class MultiplePublicMethodTestBean {
public String upperCase(String s) {
return s.toUpperCase();
}
public String lowerCase(String s) {
return s.toLowerCase();
}
}
MultiplePublicMethodTestBean bean = new MultiplePublicMethodTestBean();
new MethodInvokingMessageGroupProcessor(bean);
}
@Test(expected = IllegalArgumentException.class)
public void noPublicMethods() {
@SuppressWarnings("unused")
class NoPublicMethodTestBean {
String lowerCase(String s) {
return s.toLowerCase();
}
}
NoPublicMethodTestBean bean = new NoPublicMethodTestBean();
new MethodInvokingMessageGroupProcessor(bean);
}
@@ -264,74 +512,8 @@ public class MethodInvokingMessageGroupProcessorTests {
assertEquals("hello proxy", output.receive(0).getPayload());
}
private Method getMethod(MethodInvokingMessageGroupProcessor aggregator) {
Object invoker = new DirectFieldAccessor(aggregator).getPropertyValue("adapter");
return (Method) new DirectFieldAccessor(invoker).getPropertyValue("method");
}
@SuppressWarnings("unused")
private static class SingleAnnotationTestBean {
@Aggregator
public String method1(List<String> input) {
return input.get(0);
}
public String method2(List<String> input) {
return input.get(0);
}
}
@SuppressWarnings("unused")
private static class MultipleAnnotationTestBean {
@Aggregator
public String method1(List<String> input) {
return input.get(0);
}
@Aggregator
public String method2(List<String> input) {
return input.get(0);
}
}
@SuppressWarnings("unused")
private static class NoAnnotationTestBean {
public String method1(List<String> input) {
return input.get(0);
}
String method2(List<String> input) {
return input.get(0);
}
}
@SuppressWarnings("unused")
private static class MultiplePublicMethodTestBean {
public String upperCase(String s) {
return s.toUpperCase();
}
public String lowerCase(String s) {
return s.toLowerCase();
}
}
@SuppressWarnings("unused")
private static class NoPublicMethodTestBean {
String lowerCase(String s) {
return s.toLowerCase();
}
}
public interface GreetingService {
String sayHello(List<String> names);
}
public static class GreetingBean implements GreetingService {

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

@@ -0,0 +1,287 @@
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aggregator;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.core.convert.ConversionFailedException;
import org.springframework.integration.Message;
import org.springframework.integration.core.GenericMessage;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageGroup;
/**
* @author Marius Bogoevici
* @author Dave Syer
*/
public class MethodInvokingReleaseStrategyTests {
@Test
public void testTrueConvertedProperly() {
MethodInvokingReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new AlwaysTrueReleaseStrategy(),
"checkCompleteness");
Assert.assertTrue(adapter.canRelease(createListOfMessages(0)));
}
@Test
public void testFalseConvertedProperly() {
MethodInvokingReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new AlwaysFalseReleaseStrategy(),
"checkCompleteness");
Assert.assertTrue(!adapter.canRelease(createListOfMessages(0)));
}
@Test
public void testAdapterWithNonParameterizedMessageListBasedMethod() {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean checkCompletenessOnNonParameterizedListOfMessages(List<Message<?>> messages) {
Assert.assertTrue(messages.size() > 0);
return messages.size() > messages.iterator().next().getHeaders().getSequenceSize();
}
}
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new TestReleaseStrategy(),
"checkCompletenessOnNonParameterizedListOfMessages");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test
public void testAdapterWithWildcardParametrizedMessageBasedMethod() {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean checkCompletenessOnListOfMessagesParametrizedWithWildcard(List<Message<?>> messages) {
Assert.assertTrue(messages.size() > 0);
return messages.size() > messages.iterator().next().getHeaders().getSequenceSize();
}
}
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new TestReleaseStrategy(),
"checkCompletenessOnListOfMessagesParametrizedWithWildcard");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test
public void testAdapterWithTypeParametrizedMessageBasedMethod() {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean checkCompletenessOnListOfMessagesParametrizedWithString(List<Message<String>> messages) {
Assert.assertTrue(messages.size() > 0);
return messages.size() > messages.iterator().next().getHeaders().getSequenceSize();
}
}
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new TestReleaseStrategy(),
"checkCompletenessOnListOfMessagesParametrizedWithString");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test
public void testAdapterWithPojoBasedMethod() {
class TestReleaseStrategy {
@SuppressWarnings("unused")
// Example for the case when completeness is checked on the structure of
// the data
public boolean checkCompletenessOnListOfStrings(List<String> messages) {
StringBuffer buffer = new StringBuffer();
for (String content : messages) {
buffer.append(content);
}
return buffer.length() >= 9;
}
}
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new TestReleaseStrategy(),
"checkCompletenessOnListOfStrings");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test
public void testAdapterWithPojoBasedMethodReturningObject() {
class TestReleaseStrategy {
@SuppressWarnings("unused")
// Example for the case when completeness is checked on the structure of
// the data
public boolean checkCompletenessOnListOfStrings(List<String> messages) {
StringBuffer buffer = new StringBuffer();
for (String content : messages) {
buffer.append(content);
}
return buffer.length() >= 9;
}
}
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new TestReleaseStrategy(),
"checkCompletenessOnListOfStrings");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test(expected = IllegalArgumentException.class)
public void testAdapterWithWrongMethodName() {
class TestReleaseStrategy {
}
new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), "methodThatDoesNotExist");
}
@Test(expected = IllegalStateException.class)
public void testInvalidParameterTypeUsingMethodName() {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean invalidParameterType(Date invalid) {
return true;
}
}
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), "invalidParameterType");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test(expected = IllegalArgumentException.class)
public void testTooManyParametersUsingMethodName() {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean tooManyParameters(List<?> c1, List<?> c2) {
return false;
}
}
new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), "tooManyParameters");
}
@Test
public void testNotEnoughParametersUsingMethodName() {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean notEnoughParameters() {
return false;
}
}
// TODO: this is stupid, but maybe it should be illegal?
new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), "notEnoughParameters");
}
@Test
public void testNotEnoughParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean notEnoughParameters() {
return false;
}
}
// TODO: this is stupid, but maybe it should be illegal?
new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), TestReleaseStrategy.class.getMethod(
"notEnoughParameters", new Class[] {}));
}
@Test
public void testListSubclassParameterUsingMethodName() {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean listSubclassParameter(LinkedList<?> l1) {
return true;
}
}
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), "listSubclassParameter");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
// TODO: should this be MessageHandlingException?
@Test(expected = ConversionFailedException.class)
public void testWrongReturnType() throws SecurityException, NoSuchMethodError {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public String wrongReturnType(List<Message<?>> messages) {
return "foo";
}
}
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), "wrongReturnType");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test(expected = IllegalArgumentException.class)
public void testTooManyParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean tooManyParameters(List<?> c1, List<?> c2) {
return false;
}
}
new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), TestReleaseStrategy.class.getMethod(
"tooManyParameters", List.class, List.class));
}
@Test
public void testListSubclassParameterUsingMethodObject() throws SecurityException, NoSuchMethodException {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public boolean listSubclassParameter(LinkedList<?> l1) {
return true;
}
}
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new TestReleaseStrategy(),
TestReleaseStrategy.class.getMethod("listSubclassParameter", new Class[] { LinkedList.class }));
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
// TODO: review exception type here
@Test(expected = IllegalStateException.class)
public void testWrongReturnTypeUsingMethodObject() throws SecurityException, NoSuchMethodException {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public int wrongReturnType(List<Message<?>> message) {
return 0;
}
}
new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), TestReleaseStrategy.class.getMethod(
"wrongReturnType", new Class[] { List.class }));
}
private static MessageGroup createListOfMessages(int size) {
List<Message<?>> messages = new ArrayList<Message<?>>();
if (size > 0) {
messages.add(new GenericMessage<String>("123"));
}
if (size > 1) {
messages.add(new GenericMessage<String>("456"));
}
if (size > 2) {
messages.add(new GenericMessage<String>("789"));
}
return new SimpleMessageGroup(messages, "ABC");
}
@SuppressWarnings("unused")
private static class AlwaysTrueReleaseStrategy {
public boolean checkCompleteness(List<Message<?>> messages) {
return true;
}
}
@SuppressWarnings("unused")
private static class AlwaysFalseReleaseStrategy {
public boolean checkCompleteness(List<Message<?>> messages) {
return false;
}
}
}

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

@@ -1,226 +0,0 @@
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aggregator;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.core.GenericMessage;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageGroup;
/**
* @author Marius Bogoevici
*/
public class ReleaseStrategyAdapterTests {
private SimpleReleaseStrategy simpleReleaseStrategy;
@Before
public void setUp() {
simpleReleaseStrategy = new SimpleReleaseStrategy();
}
@Test
public void testTrueConvertedProperly() {
MethodInvokingReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new AlwaysTrueReleaseStrategy(),
"checkCompleteness");
Assert.assertTrue(adapter.canRelease(createListOfMessages(0)));
}
@Test
public void testFalseConvertedProperly() {
MethodInvokingReleaseStrategy adapter = new MethodInvokingReleaseStrategy(new AlwaysFalseReleaseStrategy(),
"checkCompleteness");
Assert.assertTrue(!adapter.canRelease(createListOfMessages(0)));
}
@Test
public void testAdapterWithNonParameterizedMessageListBasedMethod() {
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy,
"checkCompletenessOnNonParameterizedListOfMessages");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test
public void testAdapterWithWildcardParametrizedMessageBasedMethod() {
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy,
"checkCompletenessOnListOfMessagesParametrizedWithWildcard");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test
public void testAdapterWithTypeParametrizedMessageBasedMethod() {
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy,
"checkCompletenessOnListOfMessagesParametrizedWithString");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test
public void testAdapterWithPojoBasedMethod() {
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "checkCompletenessOnListOfStrings");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test
public void testAdapterWithPojoBasedMethodReturningObject() {
ReleaseStrategy adapter = new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "checkCompletenessOnListOfStrings");
MessageGroup messages = createListOfMessages(3);
Assert.assertTrue(adapter.canRelease(messages));
}
@Test(expected = IllegalArgumentException.class)
public void testAdapterWithWrongMethodName() {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "methodThatDoesNotExist");
}
@Test(expected = IllegalArgumentException.class)
public void testInvalidParameterTypeUsingMethodName() {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "invalidParameterType");
}
@Test(expected = IllegalArgumentException.class)
public void testTooManyParametersUsingMethodName() {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "tooManyParameters");
}
@Test(expected = IllegalArgumentException.class)
public void testNotEnoughParametersUsingMethodName() {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "notEnoughParameters");
}
@Test(expected = IllegalArgumentException.class)
public void testListSubclassParameterUsingMethodName() {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "ListSubclassParameter");
}
@Test(expected = IllegalArgumentException.class)
public void testWrongReturnType() throws SecurityException, NoSuchMethodError {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, "wrongReturnType");
}
@Test(expected = IllegalArgumentException.class)
public void testTooManyParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod(
"tooManyParameters", List.class, List.class));
}
@Test(expected = IllegalArgumentException.class)
public void testNotEnoughParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod(
"notEnoughParameters", new Class[] {}));
}
@Test(expected = IllegalArgumentException.class)
public void testListSubclassParameterUsingMethodObject() throws SecurityException, NoSuchMethodException {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod(
"ListSubclassParameter", new Class[] { LinkedList.class }));
}
@Test(expected = IllegalArgumentException.class)
public void testWrongReturnTypeUsingMethodObject() throws SecurityException, NoSuchMethodException {
new MethodInvokingReleaseStrategy(simpleReleaseStrategy, simpleReleaseStrategy.getClass().getMethod("wrongReturnType",
new Class[] { List.class }));
}
private static MessageGroup createListOfMessages(int size) {
List<Message<?>> messages = new ArrayList<Message<?>>();
if (size > 0) {
messages.add(new GenericMessage<String>("123"));
}
if (size > 1) {
messages.add(new GenericMessage<String>("456"));
}
if (size > 2) {
messages.add(new GenericMessage<String>("789"));
}
return new SimpleMessageGroup(messages, "ABC");
}
@SuppressWarnings("unused")
private static class AlwaysTrueReleaseStrategy {
public boolean checkCompleteness(List<Message<?>> messages) {
return true;
}
}
@SuppressWarnings("unused")
private static class AlwaysFalseReleaseStrategy {
public boolean checkCompleteness(List<Message<?>> messages) {
return false;
}
}
@SuppressWarnings("unused")
private static class SimpleReleaseStrategy {
public boolean checkCompletenessOnNonParameterizedListOfMessages(List<Message<?>> messages) {
Assert.assertTrue(messages.size() > 0);
return messages.size() > messages.iterator().next().getHeaders().getSequenceSize();
}
public boolean checkCompletenessOnListOfMessagesParametrizedWithWildcard(List<Message<?>> messages) {
Assert.assertTrue(messages.size() > 0);
return messages.size() > messages.iterator().next().getHeaders().getSequenceSize();
}
public boolean checkCompletenessOnListOfMessagesParametrizedWithString(List<Message<String>> messages) {
Assert.assertTrue(messages.size() > 0);
return messages.size() > messages.iterator().next().getHeaders().getSequenceSize();
}
// Example for the case when completeness is checked on the structure of
// the data
public boolean checkCompletenessOnListOfStrings(List<String> messages) {
StringBuffer buffer = new StringBuffer();
for (String content : messages) {
buffer.append(content);
}
return buffer.length() >= 9;
}
public String wrongReturnType(List<Message<?>> message) {
return "";
}
public boolean invalidParameterType(String invalid) {
return false;
}
public boolean tooManyParameters(List<?> c1, List<?> c2) {
return false;
}
public boolean notEnoughParameters() {
return false;
}
public boolean ListSubclassParameter(LinkedList<?> l1) {
return false;
}
}
}

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

@@ -50,7 +50,7 @@ public class AggregatorIntegrationTests {
@Qualifier("output")
private PollableChannel output;
@Test(timeout=5000)
@Test//(timeout=5000)
public void testVanillaAggregation() throws Exception {
for (int i = 0; i < 5; i++) {
Map<String, Object> headers = stubHeaders(i, 5, 1);

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

@@ -19,10 +19,11 @@ package org.springframework.integration.config;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Assert;
@@ -38,9 +39,8 @@ 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.MethodInvokingMessageListProcessor;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.core.MessageBuilder;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHandler;
@@ -48,7 +48,6 @@ 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;
/**
* @author Marius Bogoevici
@@ -99,7 +98,8 @@ public class AggregatorParserTests {
for (Message<?> message : outboundMessages) {
input.send(message);
}
assertEquals("The aggregated message payload is not correct", "[123]", aggregatedMessage.get().getPayload().toString());
assertEquals("The aggregated message payload is not correct", "[123]", aggregatedMessage.get().getPayload()
.toString());
}
@Test
@@ -112,10 +112,14 @@ public class AggregatorParserTests {
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, ((MethodInvokingMessageListProcessor) new DirectFieldAccessor(accessor
.getPropertyValue("outputProcessor")).getPropertyValue("adapter")).getMethod());
Map<?, ?> map = (Map<?, ?>) new DirectFieldAccessor(new DirectFieldAccessor(new DirectFieldAccessor(accessor
.getPropertyValue("outputProcessor")).getPropertyValue("processor")).getPropertyValue("delegate"))
.getPropertyValue("handlerMethods");
assertEquals("The MethodInvokingAggregator is not injected with the appropriate aggregation method", 1, map
.size());
assertEquals("The release strategy is not injected with the appropriate method", 1, map.size());
assertTrue("Handler methods do not contain correct method: " + map, map.toString().contains(
"createSingleMessageFromGroup"));
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",
@@ -163,13 +167,12 @@ public class AggregatorParserTests {
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"));
DirectFieldAccessor releaseStrategyAccessor = new DirectFieldAccessor(new DirectFieldAccessor(new DirectFieldAccessor(releaseStrategy)
.getPropertyValue("adapter")).getPropertyValue("delegate"));
Map<?, ?> map = (Map<?, ?>) releaseStrategyAccessor.getPropertyValue("handlerMethods");
assertEquals("The release strategy is not injected with the appropriate method", 1, map.size());
assertTrue("Handler methods do not contain correct method: " + map, map.toString()
.contains("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));

4
spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregatorBean.java
View File

@@ -48,7 +48,9 @@ public class TestAggregatorBean {
}
}
Message<?> returnedMessage = new StringMessage(buffer.toString());
aggregatedMessages.put(correlationId, returnedMessage);
if (correlationId!=null) {
aggregatedMessages.put(correlationId, returnedMessage);
}
return returnedMessage;
}

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

@@ -82,15 +82,16 @@ public class AggregatorAnnotationTests {
new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithDefaultAnnotationAndCustomReleaseStrategy";
MessageHandler aggregator = this.getAggregator(context, endpointName);
Object ReleaseStrategy = getPropertyValue(aggregator, "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");
assertSame(context.getBean(endpointName), targetObject);
Method completionCheckerMethod = (Method) invokerAccessor.getPropertyValue("method");
assertEquals("completionChecker", completionCheckerMethod.getName());
Object releaseStrategy = getPropertyValue(aggregator, "releaseStrategy");
Assert.assertTrue(releaseStrategy instanceof MethodInvokingReleaseStrategy);
MethodInvokingReleaseStrategy releaseStrategyAdapter = (MethodInvokingReleaseStrategy) releaseStrategy;
Map<?, ?> map = (Map<?, ?>) new DirectFieldAccessor(new DirectFieldAccessor(new DirectFieldAccessor(releaseStrategyAdapter)
.getPropertyValue("adapter")).getPropertyValue("delegate")).getPropertyValue("handlerMethods");
assertEquals("The MethodInvokingAggregator is not injected with the appropriate aggregation method", 1, map
.size());
assertEquals("The release strategy is not injected with the appropriate method", 1, map.size());
assertTrue("Handler methods do not contain correct method: " + map, map.toString()
.contains("completionChecker"));
}
@Test
@@ -101,9 +102,9 @@ public class AggregatorAnnotationTests {
MessageHandler aggregator = this.getAggregator(context, endpointName);
Object correlationStrategy = getPropertyValue(aggregator, "correlationStrategy");
Assert.assertTrue(correlationStrategy instanceof MethodInvokingCorrelationStrategy);
MethodInvokingCorrelationStrategy ReleaseStrategyAdapter = (MethodInvokingCorrelationStrategy) correlationStrategy;
DirectFieldAccessor processorAccessor = new DirectFieldAccessor(new DirectFieldAccessor(ReleaseStrategyAdapter)
.getPropertyValue("processor"));
MethodInvokingCorrelationStrategy releaseStrategyAdapter = (MethodInvokingCorrelationStrategy) correlationStrategy;
DirectFieldAccessor processorAccessor = new DirectFieldAccessor(new DirectFieldAccessor(new DirectFieldAccessor(releaseStrategyAdapter)
.getPropertyValue("processor")).getPropertyValue("delegate"));
Object targetObject = processorAccessor.getPropertyValue("targetObject");
assertSame(context.getBean(endpointName), targetObject);
Map<?, ?> handlerMethods = (Map<?, ?>) processorAccessor.getPropertyValue("handlerMethods");

2
spring-integration-core/src/test/java/org/springframework/integration/config/xml/PNamespaceTests.java
View File

@@ -85,7 +85,7 @@ public class PNamespaceTests {
private TestBean prepare(EventDrivenConsumer edc) {
return TestUtils.getPropertyValue(serviceActivator,
"handler.processor.targetObject", TestBean.class);
"handler.processor.delegate.targetObject", TestBean.class);
}

37
spring-integration-core/src/test/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessorTests.java
View File

@@ -27,6 +27,7 @@ import org.junit.internal.matchers.TypeSafeMatcher;
import org.junit.rules.ExpectedException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.core.io.Resource;
import org.springframework.expression.EvaluationException;
@@ -55,6 +56,12 @@ public class ExpressionEvaluatingMessageProcessorTests {
@Test
public void testProcessMessageWithParameterCoercion() {
@SuppressWarnings("unused")
class TestTarget {
public String stringify(int number) {
return number+"";
}
}
ExpressionEvaluatingMessageProcessor processor = new ExpressionEvaluatingMessageProcessor("#target.stringify(payload)");
processor.getEvaluationContext().setVariable("target", new TestTarget());
assertEquals("2", processor.processMessage(new StringMessage("2")));
@@ -62,6 +69,11 @@ public class ExpressionEvaluatingMessageProcessorTests {
@Test
public void testProcessMessageWithVoidResult() {
@SuppressWarnings("unused")
class TestTarget {
public void ping(String input) {
}
}
ExpressionEvaluatingMessageProcessor processor = new ExpressionEvaluatingMessageProcessor("#target.ping(payload)");
processor.getEvaluationContext().setVariable("target", new TestTarget());
assertEquals(null, processor.processMessage(new StringMessage("2")));
@@ -69,7 +81,15 @@ public class ExpressionEvaluatingMessageProcessorTests {
@Test
public void testProcessMessageWithParameterCoercionToNonPrimitive() {
class TestTarget {
@SuppressWarnings("unused")
public String find(Resource[] resources) {
return Arrays.asList(resources).toString();
}
}
ExpressionEvaluatingMessageProcessor processor = new ExpressionEvaluatingMessageProcessor("#target.find(payload)");
processor.setBeanFactory(new GenericApplicationContext().getBeanFactory());
processor.getEvaluationContext().setVariable("target", new TestTarget());
String result = (String) processor.processMessage(new StringMessage("classpath:*.properties"));
assertTrue("Wrong result: "+result, result.contains("log4j.properties"));
@@ -187,23 +207,6 @@ public class ExpressionEvaluatingMessageProcessorTests {
}
}
@SuppressWarnings("unused")
private static class TestTarget {
public String stringify(int number) {
return number+"";
}
public String find(Resource[] resources) {
return Arrays.asList(resources).toString();
}
public void ping(String input) {
}
}
@SuppressWarnings("serial")
private static final class CheckedException extends Exception {
public CheckedException(String string) {

187
spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java
View File

@@ -49,7 +49,7 @@ import org.springframework.integration.core.StringMessage;
public class MethodInvokingMessageProcessorTests {
private static final Log logger = LogFactory.getLog(MethodInvokingMessageProcessorTests.class);
@Rule
public ExpectedException expected = ExpectedException.none();
@@ -58,14 +58,16 @@ public class MethodInvokingMessageProcessorTests {
class A {
@SuppressWarnings("unused")
public Message<String> myMethod(final Message<String> msg) {
return MessageBuilder.fromMessage(msg).setHeader("A", "A").build();
}
return MessageBuilder.fromMessage(msg).setHeader("A", "A").build();
}
}
class B extends A {}
class B extends A {
}
@SuppressWarnings("unused")
class C extends B {}
class C extends B {
}
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new B(), "myMethod");
Message<?> message = (Message<?>) processor.processMessage(new StringMessage(""));
@@ -77,18 +79,19 @@ public class MethodInvokingMessageProcessorTests {
class A {
@SuppressWarnings("unused")
public Message<String> myMethod(Message<String> msg) {
return MessageBuilder.fromMessage(msg).setHeader("A", "A").build();
}
return MessageBuilder.fromMessage(msg).setHeader("A", "A").build();
}
}
class B extends A {
public Message<String> myMethod(Message<String> msg) {
return MessageBuilder.fromMessage(msg).setHeader("B", "B").build();
}
return MessageBuilder.fromMessage(msg).setHeader("B", "B").build();
}
}
@SuppressWarnings("unused")
class C extends B {}
class C extends B {
}
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new B(), "myMethod");
Message<?> message = (Message<?>) processor.processMessage(new StringMessage(""));
@@ -99,20 +102,20 @@ public class MethodInvokingMessageProcessorTests {
class A {
@SuppressWarnings("unused")
public Message<String> myMethod(Message<String> msg) {
return MessageBuilder.fromMessage(msg).setHeader("A", "A").build();
}
return MessageBuilder.fromMessage(msg).setHeader("A", "A").build();
}
}
class B extends A {
public Message<String> myMethod(Message<String> msg) {
return MessageBuilder.fromMessage(msg).setHeader("B", "B").build();
}
return MessageBuilder.fromMessage(msg).setHeader("B", "B").build();
}
}
class C extends B {
public Message<String> myMethod(Message<String> msg) {
return MessageBuilder.fromMessage(msg).setHeader("C", "C").build();
}
return MessageBuilder.fromMessage(msg).setHeader("C", "C").build();
}
}
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new C(), "myMethod");
@@ -123,17 +126,18 @@ public class MethodInvokingMessageProcessorTests {
public void testHandlerInheritanceMethodImplInSubClassAndSuper() {
class A {
@SuppressWarnings("unused")
public Message<String> myMethod(Message<String> msg){
return MessageBuilder.fromMessage(msg).setHeader("A", "A").build();
}
public Message<String> myMethod(Message<String> msg) {
return MessageBuilder.fromMessage(msg).setHeader("A", "A").build();
}
}
class B extends A {}
class B extends A {
}
class C extends B {
public Message<String> myMethod(Message<String> msg) {
return MessageBuilder.fromMessage(msg).setHeader("C", "C").build();
}
return MessageBuilder.fromMessage(msg).setHeader("C", "C").build();
}
}
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new C(), "myMethod");
@@ -143,92 +147,102 @@ public class MethodInvokingMessageProcessorTests {
@Test
public void payloadAsMethodParameterAndObjectAsReturnValue() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(
new TestBean(), "acceptPayloadAndReturnObject");
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(),
"acceptPayloadAndReturnObject");
Object result = processor.processMessage(new StringMessage("testing"));
assertEquals("testing-1", result);
}
@Test
public void testPayloadCoercedToString() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(
new TestBean(), "acceptPayloadAndReturnObject");
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(),
"acceptPayloadAndReturnObject");
Object result = processor.processMessage(new GenericMessage<Integer>(123456789));
assertEquals("123456789-1", result);
}
@Test
public void payloadAsMethodParameterAndMessageAsReturnValue() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(
new TestBean(), "acceptPayloadAndReturnMessage");
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(),
"acceptPayloadAndReturnMessage");
Message<?> result = (Message<?>) processor.processMessage(new StringMessage("testing"));
assertEquals("testing-2", result.getPayload());
}
@Test
public void messageAsMethodParameterAndObjectAsReturnValue() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(
new TestBean(), "acceptMessageAndReturnObject");
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(),
"acceptMessageAndReturnObject");
Object result = processor.processMessage(new StringMessage("testing"));
assertEquals("testing-3", result);
}
@Test
public void messageAsMethodParameterAndMessageAsReturnValue() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(
new TestBean(), "acceptMessageAndReturnMessage");
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(),
"acceptMessageAndReturnMessage");
Message<?> result = (Message<?>) processor.processMessage(new StringMessage("testing"));
assertEquals("testing-4", result.getPayload());
}
@Test
public void messageSubclassAsMethodParameterAndMessageAsReturnValue() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(
new TestBean(), "acceptMessageSubclassAndReturnMessage");
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(),
"acceptMessageSubclassAndReturnMessage");
Message<?> result = (Message<?>) processor.processMessage(new StringMessage("testing"));
assertEquals("testing-5", result.getPayload());
}
@Test
public void messageSubclassAsMethodParameterAndMessageSubclassAsReturnValue() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(
new TestBean(), "acceptMessageSubclassAndReturnMessageSubclass");
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(),
"acceptMessageSubclassAndReturnMessageSubclass");
Message<?> result = (Message<?>) processor.processMessage(new StringMessage("testing"));
assertEquals("testing-6", result.getPayload());
}
@Test
public void payloadAndHeaderAnnotationMethodParametersAndObjectAsReturnValue() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(
new TestBean(), "acceptPayloadAndHeaderAndReturnObject");
Message<?> request = MessageBuilder.withPayload("testing")
.setHeader("number", new Integer(123)).build();
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(),
"acceptPayloadAndHeaderAndReturnObject");
Message<?> request = MessageBuilder.withPayload("testing").setHeader("number", new Integer(123)).build();
Object result = processor.processMessage(request);
assertEquals("testing-123", result);
}
@Test
public void testVoidMethodsIncludedbyDefault() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(), "testVoidReturningMethods");
assertNull(processor.processMessage(MessageBuilder.withPayload("Something").build()));
assertEquals(12, processor.processMessage(MessageBuilder.withPayload(12).build()));
}
@Test
public void testVoidMethodsIncludedbyDefault() {
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(),
"testVoidReturningMethods");
assertNull(processor.processMessage(MessageBuilder.withPayload("Something").build()));
assertEquals(12, processor.processMessage(MessageBuilder.withPayload(12).build()));
}
@Test
public void testVoidMethodsExcludedByFlag() {
Exception exception = null;
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new TestBean(), "testVoidReturningMethods", true);
assertEquals(12, processor.processMessage(MessageBuilder.withPayload(12).build()));
try {
processor.processMessage(MessageBuilder.withPayload("not_a_number").build());
fail();
}
catch(MessageHandlingException ex) {
// the only void method expects a number
exception = ex;
}
assertNotNull(exception);
}
@Test
public void testVoidMethodsExcludedByFlag() {
@SuppressWarnings("unused")
class VoidMethodsBean {
public void testVoidReturningMethods(String s) {
// do nothing
}
public int testVoidReturningMethods(int i) {
return i;
}
}
Exception exception = null;
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new VoidMethodsBean(),
"testVoidReturningMethods", true);
assertEquals(12, processor.processMessage(MessageBuilder.withPayload(12).build()));
try {
processor.processMessage(MessageBuilder.withPayload("not_a_number").build());
fail();
}
catch (MessageHandlingException ex) {
// the only void method expects a number
exception = ex;
}
assertNotNull(exception);
}
@Test
public void messageOnlyWithAnnotatedMethod() throws Exception {
@@ -267,6 +281,7 @@ public class MethodInvokingMessageProcessorTests {
@Test
public void testProcessMessageBadExpression() throws Exception {
// TODO: should this be MessageHandlingException or NumberFormatException?
expected.expect(new ExceptionCauseMatcher(NumberFormatException.class));
AnnotatedTestService service = new AnnotatedTestService();
Method method = service.getClass().getMethod("integerMethod", Integer.class);
@@ -297,8 +312,7 @@ public class MethodInvokingMessageProcessorTests {
AnnotatedTestService service = new AnnotatedTestService();
Method method = service.getClass().getMethod("messageAndHeader", Message.class, Integer.class);
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, method);
Message<String> message = MessageBuilder.withPayload("foo")
.setHeader("number", 42).build();
Message<String> message = MessageBuilder.withPayload("foo").setHeader("number", 42).build();
Object result = processor.processMessage(message);
assertEquals("foo-42", result);
}
@@ -308,9 +322,8 @@ public class MethodInvokingMessageProcessorTests {
AnnotatedTestService service = new AnnotatedTestService();
Method method = service.getClass().getMethod("twoHeaders", String.class, Integer.class);
MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(service, method);
Message<String> message = MessageBuilder.withPayload("foo")
.setHeader("prop", "bar")
.setHeader("number", 42).build();
Message<String> message = MessageBuilder.withPayload("foo").setHeader("prop", "bar").setHeader("number", 42)
.build();
Object result = processor.processMessage(message);
assertEquals("bar-42", result);
}
@@ -334,7 +347,7 @@ public class MethodInvokingMessageProcessorTests {
assertEquals(String.class, bean.lastArg.getClass());
assertEquals("true", bean.lastArg);
}
@Test
public void testOverloadedNonVoidReturningMethodsWithExactMatchForType() {
AmbiguousMethodBean bean = new AmbiguousMethodBean();
@@ -344,20 +357,24 @@ public class MethodInvokingMessageProcessorTests {
assertEquals(String.class, bean.lastArg.getClass());
assertEquals("true", bean.lastArg);
}
private static class ExceptionCauseMatcher extends TypeSafeMatcher<Exception> {
private Throwable cause;
private Class<? extends Exception> type;
public ExceptionCauseMatcher(Class<? extends Exception> type) {
this.type = type;
}
@Override
public boolean matchesSafely(Exception item) {
logger.debug(item);
cause = item.getCause();
assertNotNull("There is no cause for "+item, cause);
assertNotNull("There is no cause for " + item, cause);
return type.isAssignableFrom(cause.getClass());
}
public void describeTo(Description description) {
description.appendText("cause to be ").appendValue(type).appendText("but was ").appendValue(cause);
}
@@ -368,11 +385,12 @@ public class MethodInvokingMessageProcessorTests {
public String error(String input) {
throw new UnsupportedOperationException("Expected test exception");
}
public String checked(String input) throws Exception {
throw new CheckedException("Expected test exception");
}
}
@SuppressWarnings("serial")
public static final class CheckedException extends Exception {
public CheckedException(String string) {
@@ -411,17 +429,16 @@ public class MethodInvokingMessageProcessorTests {
return s + "-" + n;
}
public void testVoidReturningMethods(String s) {
// do nothing
}
public void testVoidReturningMethods(String s) {
// do nothing
}
public int testVoidReturningMethods(int i) {
return i;
}
public int testVoidReturningMethods(int i) {
return i;
}
}
@SuppressWarnings("unused")
private static class AnnotatedTestService {
@@ -437,15 +454,16 @@ public class MethodInvokingMessageProcessorTests {
return prop + "-" + num.toString();
}
public Integer optionalHeader(@Header(required=false) Integer num) {
public Integer optionalHeader(@Header(required = false) Integer num) {
return num;
}
public Integer requiredHeader(@Header(value="num", required=true) Integer num) {
public Integer requiredHeader(@Header(value = "num", required = true) Integer num) {
return num;
}
public String optionalAndRequiredHeader(@Header(required=false) String prop, @Header(value="num", required=true) Integer num) {
public String optionalAndRequiredHeader(@Header(required = false) String prop,
@Header(value = "num", required = true) Integer num) {
return prop + num;
}
@@ -464,10 +482,9 @@ public class MethodInvokingMessageProcessorTests {
}
/**
* Method names create ambiguities, but the MethodResolver implementation
* should filter out based on the annotation or the 'requiresReply' flag.
* Method names create ambiguities, but the MethodResolver implementation should filter out based on the annotation
* or the 'requiresReply' flag.
*/
@SuppressWarnings("unused")
private static class AmbiguousMethodBean {
@@ -491,8 +508,8 @@ public class MethodInvokingMessageProcessorTests {
}
/**
* Method names create ambiguities, but the MethodResolver implementation
* should filter out based on the annotation or the 'requiresReply' flag.
* Method names create ambiguities, but the MethodResolver implementation should filter out based on the annotation
* or the 'requiresReply' flag.
*/
@SuppressWarnings("unused")
private static class OverloadedMethodBean {