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

[INT-140] Adds support for @CompletionStrategy annotation and supports a POJO completion strategy via namespace

Browse Source
This commit is contained in:
Marius Bogoevici
2008-04-11 01:56:45 +00:00
parent ec08c33d99
commit 6226d2c06e
25 changed files with 640 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,36 @@
/*
* 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.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;
/**
* Indicates that a method is capable of asserting if a list of messages or
* payload objects is complete.
*
* @author Marius Bogoevici
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@Documented
public @interface CompletionStrategy {
}

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

@@ -16,8 +16,6 @@
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
@@ -27,7 +25,10 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.AggregatorAdapter;
import org.springframework.integration.router.CompletionStrategyAdapter;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;
/**
* Parser for the <em>aggregator</em> element of the integration namespace.
@@ -75,6 +76,9 @@ public class AggregatorParser implements BeanDefinitionParser {
public static final String TIMEOUT = "timeout";
public static final String AGGREGATOR_ELEMENT = "aggregator";
public static final String COMPLETION_STRATEGY_ELEMENT = "completion-strategy";
public BeanDefinition parse(Element element, ParserContext parserContext) {
return parseAggregatorElement(element, parserContext, true);
@@ -86,6 +90,8 @@ public class AggregatorParser implements BeanDefinitionParser {
final String id = element.getAttribute(ID_ATTRIBUTE);
final String ref = element.getAttribute(REF_ATTRIBUTE);
final String method = element.getAttribute(METHOD_ATTRIBUTE);
final String completionStrategyRef = element.getAttribute(COMPLETION_STRATEGY_ATTRIBUTE);
final NodeList completionStrategyChildElements = element.getElementsByTagName(COMPLETION_STRATEGY_ELEMENT);
if (!StringUtils.hasText(ref)) {
throw new ConfigurationException("The 'ref' attribute must be present");
}
@@ -94,19 +100,36 @@ public class AggregatorParser implements BeanDefinitionParser {
"The 'id' attribute is only supported for top-level <aggregator> elements.",
parserContext.extractSource(element));
}
if (completionStrategyChildElements.getLength() > 0 && StringUtils.hasText(completionStrategyRef)) {
parserContext
.getReaderContext()
.error(
"The 'completion-strategy' element is only supported when no 'completion-strategy' attribute is specified.",
parserContext.extractSource(element));
}
if (!StringUtils.hasText(method)) {
aggregatorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(ref));
}
else {
BeanDefinition adapterDefinition = new RootBeanDefinition(AggregatorAdapter.class);
adapterDefinition.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(ref));
adapterDefinition.getConstructorArgumentValues().addGenericArgumentValue(method);
String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDefinition);
parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDefinition, adapterBeanName));
aggregatorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(adapterBeanName));
String adapterBeanName = createAdapterAndReturnBeanName(parserContext, ref, method, AggregatorAdapter.class);
aggregatorDef.getConstructorArgumentValues().addGenericArgumentValue(
new RuntimeBeanReference(adapterBeanName));
}
IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, COMPLETION_STRATEGY_PROPERTY,
element, COMPLETION_STRATEGY_ATTRIBUTE);
if (StringUtils.hasText(completionStrategyRef)) {
aggregatorDef.getPropertyValues().addPropertyValue(COMPLETION_STRATEGY_PROPERTY,
new RuntimeBeanReference(completionStrategyRef));
}
else if (completionStrategyChildElements.getLength() > 0) {
Element completionStrategyElement = (Element) completionStrategyChildElements.item(0);
String childCompletionStrategyReference = completionStrategyElement.getAttribute(REF_ATTRIBUTE);
String childCompletionStrategyMethod = completionStrategyElement.getAttribute(METHOD_ATTRIBUTE);
String adapterBeanName = createAdapterAndReturnBeanName(parserContext, childCompletionStrategyReference,
childCompletionStrategyMethod, CompletionStrategyAdapter.class);
aggregatorDef.getPropertyValues().addPropertyValue(COMPLETION_STRATEGY_PROPERTY,
new RuntimeBeanReference(adapterBeanName));
}
IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, DEFAULT_REPLY_CHANNEL_PROPERTY,
element, DEFAULT_REPLY_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, DISCARD_CHANNEL_PROPERTY, element,
@@ -126,4 +149,14 @@ public class AggregatorParser implements BeanDefinitionParser {
return aggregatorDef;
}
private String createAdapterAndReturnBeanName(ParserContext parserContext, final String ref, final String method,
Class<?> adapterClass) {
BeanDefinition adapterDefinition = new RootBeanDefinition(adapterClass);
adapterDefinition.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(ref));
adapterDefinition.getConstructorArgumentValues().addGenericArgumentValue(method);
String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDefinition);
parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDefinition, adapterBeanName));
return adapterBeanName;
}
}

70
spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java
View File

@@ -26,7 +26,6 @@ import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
@@ -38,7 +37,15 @@ import org.springframework.integration.adapter.DefaultTargetAdapter;
import org.springframework.integration.adapter.MethodInvokingSource;
import org.springframework.integration.adapter.MethodInvokingTarget;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.annotation.*;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.annotation.Concurrency;
import org.springframework.integration.annotation.DefaultOutput;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Polled;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
@@ -50,9 +57,11 @@ import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.handler.config.DefaultMessageHandlerCreator;
import org.springframework.integration.handler.config.MessageHandlerCreator;
import org.springframework.integration.message.Message;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.CompletionStrategyAdapter;
import org.springframework.integration.router.config.AggregatorMessageHandlerCreator;
import org.springframework.integration.router.config.RouterMessageHandlerCreator;
import org.springframework.integration.router.config.SplitterMessageHandlerCreator;
import org.springframework.integration.router.config.AggregatorMessageHandlerCreator;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.Assert;
@@ -64,25 +73,22 @@ import org.springframework.util.StringUtils;
* classes annotated with {@link MessageEndpoint @MessageEndpoint}.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor, InitializingBean {
private final Log logger = LogFactory.getLog(this.getClass());
private final Map<Class<? extends Annotation>, MessageHandlerCreator> handlerCreators =
new ConcurrentHashMap<Class<? extends Annotation>, MessageHandlerCreator>();
private final Map<Class<? extends Annotation>, MessageHandlerCreator> handlerCreators = new ConcurrentHashMap<Class<? extends Annotation>, MessageHandlerCreator>();
private final MessageBus messageBus;
public MessageEndpointAnnotationPostProcessor(MessageBus messageBus) {
Assert.notNull(messageBus, "'messageBus' must not be null");
this.messageBus = messageBus;
}
public void setCustomHandlerCreators(
Map<Class<? extends Annotation>, MessageHandlerCreator> customHandlerCreators) {
public void setCustomHandlerCreators(Map<Class<? extends Annotation>, MessageHandlerCreator> customHandlerCreators) {
for (Map.Entry<Class<? extends Annotation>, MessageHandlerCreator> entry : customHandlerCreators.entrySet()) {
this.handlerCreators.put(entry.getKey(), entry.getValue());
}
@@ -114,8 +120,8 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
this.configureDefaultOutput(bean, beanName, endpointAnnotation, endpoint);
Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(beanClass, Concurrency.class);
if (concurrencyAnnotation != null) {
ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(
concurrencyAnnotation.coreSize(), concurrencyAnnotation.maxSize());
ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(),
concurrencyAnnotation.maxSize());
concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds());
concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity());
endpoint.setConcurrencyPolicy(concurrencyPolicy);
@@ -127,6 +133,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
}
});
}
this.configureCompletionStrategy(bean, endpoint);
this.messageBus.registerEndpoint(beanName + "-endpoint", endpoint);
return bean;
}
@@ -166,8 +173,8 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
});
}
private void configureDefaultOutput(final Object bean, final String beanName,
final MessageEndpoint annotation, final DefaultMessageEndpoint endpoint) {
private void configureDefaultOutput(final Object bean, final String beanName, final MessageEndpoint annotation,
final DefaultMessageEndpoint endpoint) {
String channelName = annotation.defaultOutput();
if (StringUtils.hasText(channelName)) {
endpoint.setDefaultOutputChannelName(channelName);
@@ -175,6 +182,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
}
ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() {
boolean foundDefaultOutput = false;
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, DefaultOutput.class);
if (annotation != null) {
@@ -205,6 +213,38 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
});
}
private void configureCompletionStrategy(final Object bean, final DefaultMessageEndpoint endpoint) {
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class);
if (annotation != null) {
final MessageHandler endpointHandler = endpoint.getHandler();
AggregatingMessageHandler aggregatingMessageHandler = null;
if (endpointHandler != null) {
if (endpointHandler instanceof MessageHandlerChain) {
for (MessageHandler handlerInChain : ((MessageHandlerChain) endpointHandler).getHandlers()) {
if (handlerInChain instanceof AggregatingMessageHandler) {
aggregatingMessageHandler = (AggregatingMessageHandler) handlerInChain;
break;
}
}
}
else if (endpointHandler instanceof AggregatingMessageHandler) {
aggregatingMessageHandler = (AggregatingMessageHandler) endpointHandler;
}
}
if (aggregatingMessageHandler == null) {
throw new ConfigurationException(
"@CompletionStrategy supported only when @Aggregator is present");
}
else {
aggregatingMessageHandler.setCompletionStrategy(new CompletionStrategyAdapter(bean, method));
}
}
}
});
}
@SuppressWarnings("unchecked")
private MessageHandlerChain createHandlerChain(final Object bean) {
final List<MessageHandler> handlers = new ArrayList<MessageHandler>();
@@ -217,8 +257,8 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
MessageHandlerCreator handlerCreator = handlerCreators.get(annotation.annotationType());
if (handlerCreator == null) {
if (logger.isWarnEnabled()) {
logger.warn("No handler creator has been registered for handler annotation '" +
annotation.annotationType() + "'");
logger.warn("No handler creator has been registered for handler annotation '"
+ annotation.annotationType() + "'");
}
}
else {

16
spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -234,9 +234,13 @@
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:sequence>
<xsd:element name="completion-strategy" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="ref" type="xsd:string" use="required"/>
<xsd:attribute name="method" type="xsd:string" use="optional"/>
<xsd:attribute name="completion-strategy" type="xsd:string" use="optional"/>
<xsd:attribute name="completion-strategy-method" type="xsd:string" use="optional"/>
<xsd:attribute name="default-reply-channel" type="xsd:string" use="optional"/>
<xsd:attribute name="discard-channel" type="xsd:string" use="optional"/>
<xsd:attribute name="send-timeout" type="xsd:long" use="optional"/>
@@ -248,5 +252,17 @@
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="completion-strategy">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a completion strategy.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="ref" type="xsd:string" use="required"/>
<xsd:attribute name="method" type="xsd:string" use="optional"/>
</xsd:complexType>
</xsd:element>
</xsd:schema>

9
spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.handler;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -32,7 +33,6 @@ public class MessageHandlerChain implements MessageHandler {
private final List<MessageHandler> handlers = new CopyOnWriteArrayList<MessageHandler>();
/**
* Add a handler to the end of the chain.
*/
@@ -52,6 +52,13 @@ public class MessageHandlerChain implements MessageHandler {
this.handlers.addAll(handlers);
}
/**
* Get an immutable list of handlers
*/
public List<MessageHandler> getHandlers() {
return Collections.unmodifiableList(handlers);
}
public final Message<?> handle(Message<?> message) {
for (MessageHandler next : handlers) {
message = next.handle(message);

4
spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java
View File

@@ -16,7 +16,7 @@
package org.springframework.integration.router;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.message.Message;
@@ -28,6 +28,6 @@ import org.springframework.integration.message.Message;
*/
public interface Aggregator {
Message<?> aggregate(Collection<Message<?>> messages);
Message<?> aggregate(List<Message<?>> messages);
}

72
spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java
View File

@@ -38,45 +38,18 @@ import org.springframework.util.ReflectionUtils;
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class AggregatorAdapter implements Aggregator {
private final HandlerMethodInvoker<Object> invoker;
private final Method method;
public AggregatorAdapter(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<?>[] { Collection.class });
if (this.method == null) {
throw new ConfigurationException("Method '" + methodName +
"(Collection<?> args)' not found on '" + object.getClass().getName() + "'.");
}
this.invoker = new HandlerMethodInvoker<Object>(object, this.method.getName());
}
public class AggregatorAdapter extends MessageListMethodAdapter implements Aggregator {
public AggregatorAdapter(Object object, Method method) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(method, "'method' must not be null");
if (method.getParameterTypes().length != 1 || !method.getParameterTypes()[0].equals(Collection.class)) {
throw new ConfigurationException(
"Aggregator method must accept exactly one parameter, and it must be a Collection.");
}
this.method = method;
this.invoker = new HandlerMethodInvoker<Object>(object, this.method.getName());
super(object, method);
}
public AggregatorAdapter(Object object, String methodName) {
super(object, methodName);
}
public Message<?> aggregate(Collection<Message<?>> messages) {
Object returnedValue = null;
if (isMethodParameterParametrized(this.method) && isHavingActualTypeArguments(this.method)
&& (isActualTypeRawMessage(this.method) || isActualTypeParametrizedMessage(this.method))) {
returnedValue = this.invoker.invokeMethod(messages);
}
else {
returnedValue = this.invoker.invokeMethod(extractPayloadsFromMessages(messages));
}
public Message<?> aggregate(List<Message<?>> messages) {
Object returnedValue = this.executeMethod(messages);
if (returnedValue == null) {
return null;
}
@@ -86,35 +59,6 @@ public class AggregatorAdapter implements Aggregator {
return new GenericMessage<Object>(returnedValue);
}
private Collection<?> extractPayloadsFromMessages(Collection<Message<?>> messages) {
List<Object> payloadList = new ArrayList<Object>();
for (Message<?> message : messages) {
payloadList.add(message.getPayload());
}
return payloadList;
}
private static boolean isActualTypeParametrizedMessage(Method method) {
return getCollectionActualType(method) instanceof ParameterizedType
&& Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) getCollectionActualType(method))
.getRawType());
}
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 isMethodParameterParametrized(Method method) {
return method.getGenericParameterTypes().length == 1
&& method.getGenericParameterTypes()[0] instanceof ParameterizedType;
}
}

58
spring-integration-core/src/main/java/org/springframework/integration/router/CompletionStrategyAdapter.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.router;
import java.lang.reflect.Method;
import java.util.List;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.message.Message;
/**
* Aggregator adapter for methods annotated with
* {@link org.springframework.integration.annotation.CompletionStrategy @CompletionStrategy}
* and for '<code>aggregator</code>' elements that include a '<code>method</code>'
* attribute (e.g. &lt;aggregator ref="beanReference" method="methodName"/&gt;).
*
* @author Marius Bogoevici
*/
public class CompletionStrategyAdapter extends MessageListMethodAdapter implements CompletionStrategy {
public CompletionStrategyAdapter(Object object, Method method) {
super(object, method);
assertMethodReturnsBoolean();
}
public CompletionStrategyAdapter(Object object, String methodName) {
super(object, methodName);
assertMethodReturnsBoolean();
}
private void assertMethodReturnsBoolean() {
if (!Boolean.class.equals(this.getMethod().getReturnType())
&& !boolean.class.equals(this.getMethod().getReturnType())) {
throw new ConfigurationException("Method " + getMethod().getName()
+ " does not return a boolean value");
}
}
public boolean isComplete(List<Message<?>> messages) {
return ((Boolean) executeMethod(messages)).booleanValue();
}
}

115
spring-integration-core/src/main/java/org/springframework/integration/router/MessageListMethodAdapter.java Normal file
View File

@@ -0,0 +1,115 @@
/*
* 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.router;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.handler.HandlerMethodInvoker;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* Base class for implementing adapters for methods which take as an argument a
* list of {@link Message Message} instances or payloads.
*
* @author Marius Bogoevici
*/
public abstract class MessageListMethodAdapter {
private final HandlerMethodInvoker<Object> invoker;
protected Method method;
public MessageListMethodAdapter(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 });
if (this.method == null) {
throw new ConfigurationException("Method '" + methodName +
"(List<?> args)' not found on '" + object.getClass().getName() + "'.");
}
this.invoker = new HandlerMethodInvoker<Object>(object, this.method.getName());
}
public MessageListMethodAdapter(Object object, Method method) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(method, "'method' must not be null");
if (method.getParameterTypes().length != 1 || !method.getParameterTypes()[0].equals(List.class)) {
throw new ConfigurationException(
"Method must accept exactly one parameter, and it must be a Collection.");
}
this.method = method;
this.invoker = new HandlerMethodInvoker<Object>(object, this.method.getName());
}
private static boolean isActualTypeParametrizedMessage(Method method) {
return getCollectionActualType(method) instanceof ParameterizedType
&& Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) getCollectionActualType(method))
.getRawType());
}
protected final Object executeMethod(List<Message<?>> messages) {
if (isMethodParameterParametrized(this.method) && isHavingActualTypeArguments(this.method)
&& (isActualTypeRawMessage(this.method) || isActualTypeParametrizedMessage(this.method))) {
return this.invoker.invokeMethod(messages);
}
else {
return this.invoker.invokeMethod(extractPayloadsFromMessages(messages));
}
}
private List<?> extractPayloadsFromMessages(List<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 isMethodParameterParametrized(Method method) {
return method.getGenericParameterTypes().length == 1
&& method.getGenericParameterTypes()[0] instanceof ParameterizedType;
}
public Method getMethod() {
return method;
}
public void setMethod(Method method) {
this.method = method;
}
}

9
spring-integration-core/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java
View File

@@ -22,11 +22,12 @@ import org.springframework.integration.message.Message;
import org.springframework.util.CollectionUtils;
/**
* An implementation of {@link CompletionStrategy} that simply
* compares the current size of the message list to the expected 'sequenceSize'
* according to the first {@link Message} in the list.
* An implementation of {@link CompletionStrategy} that simply compares the
* current size of the message list to the expected 'sequenceSize' according to
* the first {@link Message} in the list.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class SequenceSizeCompletionStrategy implements CompletionStrategy {
@@ -34,7 +35,7 @@ public class SequenceSizeCompletionStrategy implements CompletionStrategy {
if (CollectionUtils.isEmpty(messages)) {
return false;
}
return (messages.size() >= messages.get(0).getHeader().getSequenceSize());
return messages.size() != 0 && (messages.size() >= messages.get(0).getHeader().getSequenceSize());
}
}

6
spring-integration-core/src/test/java/org/springframework/integration/config/Adder.java
View File

@@ -16,15 +16,15 @@
package org.springframework.integration.config;
import java.util.Collection;
import java.util.List;
/**
* @author Marius Bogoevici
*/
public class Adder {
public Long add(Collection<Long> results) {
long total = 0;
public Long add(List<Long> results) {
long total = 0l;
for (long partialResult: results) {
total += partialResult;
}

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

@@ -90,6 +90,8 @@ public class AggregatorAnnotationTests {
DirectFieldAccessor aggregatingMessageHandlerAccessor = new DirectFieldAccessor(aggregatingMessageHandler);
return aggregatingMessageHandlerAccessor;
}
private MessageBus getMessageBus(ApplicationContext context) {
MessageBus messageBus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);

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

@@ -16,25 +16,25 @@
package org.springframework.integration.config;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.HandlerMethodInvoker;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.Aggregator;
import org.springframework.integration.router.CompletionStrategy;
import org.springframework.util.ReflectionUtils;
import org.springframework.integration.router.CompletionStrategyAdapter;
/**
* @author Marius Bogoevici
@@ -75,32 +75,28 @@ public class AggregatorParserTests {
CompletionStrategy completionStrategy = (CompletionStrategy) context.getBean("completionStrategy");
MessageChannel defaultReplyChannel = (MessageChannel) context.getBean("replyChannel");
MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel");
DirectFieldAccessor messageHandlerFieldAccessor = new DirectFieldAccessor(completeAggregatingMessageHandler);
Assert.assertEquals("The AggregatingMessageHandler is not injected with the appropriate Aggregator instance",
testAggregator, getPropertyValue(completeAggregatingMessageHandler, "aggregator", Aggregator.class));
testAggregator, messageHandlerFieldAccessor.getPropertyValue("aggregator"));
Assert.assertEquals(
"The AggregatingMessageHandler is not injected with the appropriate CompletionStrategy instance",
completionStrategy, getPropertyValue(completeAggregatingMessageHandler, "completionStrategy",
CompletionStrategy.class));
completionStrategy, messageHandlerFieldAccessor.getPropertyValue("completionStrategy"));
Assert.assertEquals("The AggregatingMessageHandler is not injected with the appropriate default reply channel",
defaultReplyChannel, getPropertyValue(completeAggregatingMessageHandler, "defaultReplyChannel",
MessageChannel.class));
defaultReplyChannel, messageHandlerFieldAccessor.getPropertyValue("defaultReplyChannel"));
Assert.assertEquals("The AggregatingMessageHandler is not injected with the appropriate discard channel",
discardChannel, getPropertyValue(completeAggregatingMessageHandler, "discardChannel",
MessageChannel.class));
discardChannel, messageHandlerFieldAccessor.getPropertyValue("discardChannel"));
Assert.assertEquals("The AggregatingMessageHandler is not set with the appropriate timeout value", 86420000l,
getPropertyValue(completeAggregatingMessageHandler, "sendTimeout", long.class));
messageHandlerFieldAccessor.getPropertyValue("sendTimeout"));
Assert.assertEquals(
"The AggregatingMessageHandler is not configured with the appropriate 'send partial results on timeout' flag",
true, getPropertyValue(completeAggregatingMessageHandler, "sendPartialResultOnTimeout",
boolean.class));
true, messageHandlerFieldAccessor.getPropertyValue("sendPartialResultOnTimeout"));
Assert.assertEquals("The AggregatingMessageHandler is not configured with the appropriate reaper interval",
135l, getPropertyValue(completeAggregatingMessageHandler, "reaperInterval", long.class));
135l, messageHandlerFieldAccessor.getPropertyValue("reaperInterval"));
Assert.assertEquals(
"The AggregatingMessageHandler is not configured with the appropriate tracked correlationId capacity",
99, getPropertyValue(completeAggregatingMessageHandler, "trackedCorrelationIdCapacity", int.class));
99, messageHandlerFieldAccessor.getPropertyValue("trackedCorrelationIdCapacity"));
Assert.assertEquals("The AggregatingMessageHandler is not configured with the appropriate timeout",
42l, getPropertyValue(completeAggregatingMessageHandler, "timeout", long.class));
42l, messageHandlerFieldAccessor.getPropertyValue("timeout"));
}
@Test
@@ -122,7 +118,39 @@ public class AggregatorParserTests {
public void testMissingMethodOnAggregator() {
context = new ClassPathXmlApplicationContext("invalidMethodNameAggregator.xml", this.getClass());
}
@Test(expected=BeanDefinitionParsingException.class)
public void testDuplicateCompletionStrategyDefinition() {
context = new ClassPathXmlApplicationContext("completionStrategyMethodWithMissingReference.xml", this.getClass());
}
@Test
public void testAggregatorWithPojoCompletionStrategy(){
AggregatingMessageHandler aggregatorWithPojoCompletionStrategy = (AggregatingMessageHandler) context.getBean("aggregatorWithPojoCompletionStrategy");
CompletionStrategy completionStrategy = (CompletionStrategy)new DirectFieldAccessor(aggregatorWithPojoCompletionStrategy).getPropertyValue("completionStrategy");
Assert.assertTrue(completionStrategy instanceof CompletionStrategyAdapter);
DirectFieldAccessor completionStrategyAccessor = new DirectFieldAccessor(completionStrategy);
HandlerMethodInvoker<?> invoker = (HandlerMethodInvoker<?>)completionStrategyAccessor.getPropertyValue("invoker");
Assert.assertTrue(new DirectFieldAccessor(invoker).getPropertyValue("object") instanceof MaxValueCompletionStrategy);
Assert.assertTrue(((Method)completionStrategyAccessor.getPropertyValue("method")).getName().equals("checkCompleteness"));
aggregatorWithPojoCompletionStrategy.handle(createMessage(1l, "id1", 0 , 0, null));
aggregatorWithPojoCompletionStrategy.handle(createMessage(2l, "id1", 0 , 0, null));
aggregatorWithPojoCompletionStrategy.handle(createMessage(3l, "id1", 0 , 0, null));
MessageChannel replyChannel = (MessageChannel) context.getBean("replyChannel");
Message<?> reply = replyChannel.receive(0);
Assert.assertNull(reply);
aggregatorWithPojoCompletionStrategy.handle(createMessage(5l, "id1", 0 , 0, null));
reply = replyChannel.receive(0);
Assert.assertNotNull(reply);
Assert.assertEquals(11l, reply.getPayload());
}
@Test(expected=BeanDefinitionParsingException.class)
public void testAggregatorWithDuplicateCompletionStrategy() {
context = new ClassPathXmlApplicationContext("duplicateCompletionStrategy.xml", this.getClass());
}
private static <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
MessageChannel replyChannel) {
GenericMessage<T> message = new GenericMessage<T>(payload);
@@ -133,17 +161,4 @@ public class AggregatorParserTests {
return message;
}
/**
* Reading private fields through reflection, since they don't have setters
* @param beanUnderTest
* @param fieldName
* @return the value of the field
* @throws Exception
*/
private static Object getPropertyValue(Object beanUnderTest, String fieldName, Class<?> type) throws Exception {
Field field = ReflectionUtils.findField(beanUnderTest.getClass(), fieldName, type);
ReflectionUtils.makeAccessible(field);
return field.get(beanUnderTest);
}
}

58
spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java Normal file
View File

@@ -0,0 +1,58 @@
package org.springframework.integration.config;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.ConcurrentHandler;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.CompletionStrategyAdapter;
public class CompletionStrategyAnnotationTests {
@Test
public void testAnnotationWithDefaultSettings() {
ApplicationContext context = new ClassPathXmlApplicationContext(
new String[] { "classpath:/org/springframework/integration/config/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithDefaultAnnotationAndCustomCompletionStrategy";
DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context,
endpointName);
Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof CompletionStrategyAdapter);
DirectFieldAccessor invokerAccessor = new DirectFieldAccessor(new DirectFieldAccessor(
aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy")).getPropertyValue("invoker"));
Assert.assertSame(context.getBean(endpointName), invokerAccessor.getPropertyValue("object"));
Assert.assertEquals("completionChecker", invokerAccessor.getPropertyValue("method"));
}
@Test(expected=BeanCreationException.class)
public void testInvalidAnnotation() {
ApplicationContext context = new ClassPathXmlApplicationContext(
new String[] { "classpath:/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml" });
}
@SuppressWarnings("unchecked")
private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context,
final String endpointName) {
MessageBus messageBus = getMessageBus(context);
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) messageBus
.lookupEndpoint(endpointName + "-endpoint");
MessageHandlerChain messageHandlerChain = (MessageHandlerChain) endpoint.getHandler();
AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) ((List) new DirectFieldAccessor(
messageHandlerChain).getPropertyValue("handlers")).get(0);
DirectFieldAccessor aggregatingMessageHandlerAccessor = new DirectFieldAccessor(aggregatingMessageHandler);
return aggregatingMessageHandlerAccessor;
}
private MessageBus getMessageBus(ApplicationContext context) {
MessageBus messageBus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
return messageBus;
}
}

3
spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregator.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.config;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
@@ -35,7 +34,7 @@ public class TestAggregator implements Aggregator {
private final ConcurrentMap<Object, Message<?>> aggregatedMessages = new ConcurrentHashMap<Object, Message<?>>();
public Message<?> aggregate(Collection<Message<?>> messages) {
public Message<?> aggregate(List<Message<?>> messages) {
List<Message<?>> sortableList = new ArrayList<Message<?>>(messages);
Collections.sort(sortableList, new MessageSequenceComparator());
StringBuffer buffer = new StringBuffer();

68
spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategy.java Normal file
View File

@@ -0,0 +1,68 @@
/*
* Copyright 2002-2007 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.config;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.router.MessageSequenceComparator;
import org.springframework.stereotype.Component;
/**
* @author Marius Bogoevici
*/
@MessageEndpoint(input="inputChannel")
@Component("endpointWithDefaultAnnotationAndCustomCompletionStrategy")
public class TestAnnotatedEndpointWithCompletionStrategy {
private final ConcurrentMap<Object, Message<?>> aggregatedMessages = new ConcurrentHashMap<Object, Message<?>>();
@Aggregator
public Message<?> aggregatingMethod(List<Message<?>> messages) {
List<Message<?>> sortableList = new ArrayList<Message<?>>(messages);
Collections.sort(sortableList, new MessageSequenceComparator());
StringBuffer buffer = new StringBuffer();
Object correlationId = null;
for (Message<?> message : sortableList) {
buffer.append(message.getPayload().toString());
if (null == correlationId) {
correlationId = message.getHeader().getCorrelationId();
}
}
Message<?> returnedMessage = new StringMessage(buffer.toString());
aggregatedMessages.put(correlationId, returnedMessage);
return returnedMessage;
}
@CompletionStrategy
public boolean completionChecker(List<Message<?>> messages) {
return true;
}
public ConcurrentMap<Object, Message<?>> getAggregatedMessages() {
return aggregatedMessages;
}
}

3
spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.config;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
@@ -42,7 +41,7 @@ public class TestAnnotatedEndpointWithCustomizedAggregator {
@Aggregator(defaultReplyChannel = "replyChannel", discardChannel = "discardChannel",
reaperInterval = 1234, sendPartialResultsOnTimeout = true,
sendTimeout = 98765432, timeout = 4567890, trackedCorrelationIdCapacity = 42)
public Message<?> aggregatingMethod(Collection<Message<?>> messages) {
public Message<?> aggregatingMethod(List<Message<?>> messages) {
List<Message<?>> sortableList = new ArrayList<Message<?>>(messages);
Collections.sort(sortableList, new MessageSequenceComparator());
StringBuffer buffer = new StringBuffer();

3
spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.config;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
@@ -40,7 +39,7 @@ public class TestAnnotatedEndpointWithDefaultAggregator {
private final ConcurrentMap<Object, Message<?>> aggregatedMessages = new ConcurrentHashMap<Object, Message<?>>();
@Aggregator
public Message<?> aggregatingMethod(Collection<Message<?>> messages) {
public Message<?> aggregatingMethod(List<Message<?>> messages) {
List<Message<?>> sortableList = new ArrayList<Message<?>>(messages);
Collections.sort(sortableList, new MessageSequenceComparator());
StringBuffer buffer = new StringBuffer();

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

@@ -7,29 +7,41 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<aggregator id="aggregatorWithReference" ref="aggregatorBean"/>
<aggregator id="aggregatorWithReference" ref="aggregatorBean" />
<aggregator id="completelyDefinedAggregator"
ref="aggregatorBean"
completion-strategy="completionStrategy"
default-reply-channel="replyChannel"
discard-channel="discardChannel"
send-timeout="86420000"
send-partial-result-on-timeout="true"
reaper-interval="135"
tracked-correlation-id-capacity="99"
timeout="42"/>
<aggregator id="aggregatorWithReferenceAndMethod" ref="adderBean" method="add" default-reply-channel="replyChannel"/>
<aggregator id="completelyDefinedAggregator" ref="aggregatorBean"
completion-strategy="completionStrategy"
default-reply-channel="replyChannel" discard-channel="discardChannel"
send-timeout="86420000" send-partial-result-on-timeout="true"
reaper-interval="135" tracked-correlation-id-capacity="99"
timeout="42" />
<channel id="replyChannel"/>
<channel id="discardChannel"/>
<aggregator id="aggregatorWithReferenceAndMethod" ref="adderBean"
method="add" default-reply-channel="replyChannel" />
<beans:bean id="aggregatorBean" class="org.springframework.integration.config.TestAggregator"/>
<beans:bean id="adderBean" class="org.springframework.integration.config.Adder"/>
<aggregator id="aggregatorWithPojoCompletionStrategy"
ref="adderBean" method="add" default-reply-channel="replyChannel">
<completion-strategy ref="pojoCompletionStrategy"
method="checkCompleteness" />
</aggregator>
<beans:bean id="completionStrategy" class="org.springframework.integration.config.TestCompletionStrategy"/>
<channel id="replyChannel" />
<channel id="discardChannel" />
<beans:bean id="aggregatorBean"
class="org.springframework.integration.config.TestAggregator" />
<beans:bean id="adderBean"
class="org.springframework.integration.config.Adder" />
<beans:bean id="completionStrategy"
class="org.springframework.integration.config.TestCompletionStrategy" />
<beans:bean id="pojoCompletionStrategy"
class="org.springframework.integration.config.MaxValueCompletionStrategy">
<beans:constructor-arg value="10" />
</beans:bean>
</beans:beans>

22
spring-integration-core/src/test/java/org/springframework/integration/config/completionStrategyMethodWithMissingReference.xml Normal file
View File

@@ -0,0 +1,22 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<message-bus/>
<aggregator id="aggregator" ref="adderBean" method="add" completion-strategy="testCompletionStrategy" default-reply-channel="replyChannel">
<completion-strategy ref="testCompletionStrategy"/>
</aggregator>
<channel id="replyChannel"/>
<beans:bean id="adderBean" class="org.springframework.integration.config.Adder"/>
<beans:bean id="completionStrategyBean" class="org.springframework.integration.config.TestCompletionStrategy"></beans:bean>
</beans:beans>

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

@@ -6,7 +6,9 @@
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<message-bus/>
<aggregator id="aggregatorWithReferenceAndMethod" ref="adderBean" method="substract" default-reply-channel="replyChannel"/>
<channel id="replyChannel"/>

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

@@ -22,6 +22,8 @@
<context:component-scan base-package="org.springframework.integration.config" use-default-filters="false">
<context:include-filter type="regex"
expression="org\.springframework\.integration\.config\.TestAnnotatedEndpoint.*"/>
<context:exclude-filter type="regex"
expression="org\.springframework\.integration\.config\.TestAnnotatedEndpointWithCompletionStrategyOnly"/>
</context:component-scan>
</beans:beans>

27
spring-integration-core/src/test/java/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml Normal file
View File

@@ -0,0 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-2.5.xsd">
<message-bus/>
<annotation-driven/>
<channel id="inputChannel"/>
<channel id="replyChannel"/>
<channel id="discardChannel"/>
<context:component-scan base-package="org.springframework.integration.config" use-default-filters="false">
<context:include-filter type="regex"
expression="org\.springframework\.integration\.config\.TestAnnotatedEndpointWithCompletionStrategyOnly"/>
</context:component-scan>
</beans:beans>

2
spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java
View File

@@ -208,7 +208,7 @@ public class AggregatingMessageHandlerTests {
private static class TestAggregator implements Aggregator {
public Message<?> aggregate(Collection<Message<?>> messages) {
public Message<?> aggregate(List<Message<?>> messages) {
List<Message<?>> sortableList = new ArrayList<Message<?>>(messages);
Collections.sort(sortableList, new MessageSequenceComparator());
StringBuffer buffer = new StringBuffer();

53
spring-integration-core/src/test/java/org/springframework/integration/router/AggregatorAdapterTests.java
View File

@@ -18,13 +18,12 @@ package org.springframework.integration.router;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
@@ -44,9 +43,9 @@ public class AggregatorAdapterTests {
}
@Test
public void testAdapterWithNonParameterizedMessageCollectionBasedMethod() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnNonParameterizedCollectionOfMessages");
Collection<Message<?>> messages = createCollectionOfMessages();
public void testAdapterWithNonParameterizedMessageListBasedMethod() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnNonParameterizedListOfMessages");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
Assert.assertEquals("123456789", returnedMessge.getPayload());
@@ -54,8 +53,8 @@ public class AggregatorAdapterTests {
@Test
public void testAdapterWithWildcardParametrizedMessageBasedMethod() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnCollectionOfMessagesParametrizedWithWildcard");
Collection<Message<?>> messages = createCollectionOfMessages();
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfMessagesParametrizedWithWildcard");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
Assert.assertEquals("123456789", returnedMessge.getPayload());
@@ -63,8 +62,8 @@ public class AggregatorAdapterTests {
@Test
public void testAdapterWithTypeParametrizedMessageBasedMethod() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnCollectionOfMessagesParametrizedWithString");
Collection<Message<?>> messages = createCollectionOfMessages();
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfMessagesParametrizedWithString");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
Assert.assertEquals("123456789", returnedMessge.getPayload());
@@ -72,8 +71,8 @@ public class AggregatorAdapterTests {
@Test
public void testAdapterWithPojoBasedMethod() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnCollectionOfStrings");
Collection<Message<?>> messages = createCollectionOfMessages();
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfStrings");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
Assert.assertEquals("123456789", returnedMessge.getPayload());
@@ -81,8 +80,8 @@ public class AggregatorAdapterTests {
@Test
public void testAdapterWithPojoBasedMethodReturningObject() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnCollectionOfStringsReturningLong");
Collection<Message<?>> messages = createCollectionOfMessages();
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfStringsReturningLong");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
Assert.assertEquals(123456789l, returnedMessge.getPayload());
@@ -109,8 +108,8 @@ public class AggregatorAdapterTests {
}
@Test(expected=ConfigurationException.class)
public void testCollectionSubclassParameterUsingMethodName() {
new AggregatorAdapter(simpleAggregator, "collectionSubclassParameter");
public void testListSubclassParameterUsingMethodName() {
new AggregatorAdapter(simpleAggregator, "ListSubclassParameter");
}
@Test(expected=ConfigurationException.class)
@@ -122,7 +121,7 @@ public class AggregatorAdapterTests {
@Test(expected=ConfigurationException.class)
public void testTooManyParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod(
"tooManyParameters", Collection.class, Collection.class));
"tooManyParameters", List.class, List.class));
}
@Test(expected=ConfigurationException.class)
@@ -132,9 +131,9 @@ public class AggregatorAdapterTests {
}
@Test(expected= ConfigurationException.class)
public void testCollectionSubclassParameterUsingMethodObject() throws SecurityException, NoSuchMethodException {
public void testListSubclassParameterUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod(
"collectionSubclassParameter", new Class[] {List.class} ));
"listSubclassParameter", new Class[] {LinkedList.class} ));
}
@Test(expected=IllegalArgumentException.class)
@@ -155,8 +154,8 @@ public class AggregatorAdapterTests {
}
private static Collection<Message<?>> createCollectionOfMessages() {
Collection<Message<?>> messages = new ArrayList<Message<?>>();
private static List<Message<?>> createListOfMessages() {
List<Message<?>> messages = new ArrayList<Message<?>>();
messages.add(new GenericMessage<String>("123"));
messages.add(new GenericMessage<String>("456"));
messages.add(new GenericMessage<String>("789"));
@@ -178,7 +177,7 @@ public class AggregatorAdapterTests {
}
@SuppressWarnings("unchecked")
public Message<?> doAggregationOnNonParameterizedCollectionOfMessages(Collection<Message> messages) {
public Message<?> doAggregationOnNonParameterizedListOfMessages(List<Message> messages) {
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (Message<?> message : messages) {
@@ -187,7 +186,7 @@ public class AggregatorAdapterTests {
return new GenericMessage<String>(buffer.toString());
}
public Message<?> doAggregationOnCollectionOfMessagesParametrizedWithWildcard(Collection<Message<?>> messages) {
public Message<?> doAggregationOnListOfMessagesParametrizedWithWildcard(List<Message<?>> messages) {
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (Message<?> message : messages) {
@@ -196,7 +195,7 @@ public class AggregatorAdapterTests {
return new GenericMessage<String>(buffer.toString());
}
public Message<?> doAggregationOnCollectionOfMessagesParametrizedWithString(Collection<Message<String>> messages) {
public Message<?> doAggregationOnListOfMessagesParametrizedWithString(List<Message<String>> messages) {
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (Message<String> message : messages) {
@@ -205,7 +204,7 @@ public class AggregatorAdapterTests {
return new GenericMessage<String>(buffer.toString());
}
public Message<?> doAggregationOnCollectionOfStrings(Collection<String> messages) {
public Message<?> doAggregationOnListOfStrings(List<String> messages) {
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (String payload : messages) {
@@ -214,7 +213,7 @@ public class AggregatorAdapterTests {
return new GenericMessage<String>(buffer.toString());
}
public Long doAggregationOnCollectionOfStringsReturningLong(Collection<String> messages) {
public Long doAggregationOnListOfStringsReturningLong(List<String> messages) {
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (String payload : messages) {
@@ -227,7 +226,7 @@ public class AggregatorAdapterTests {
return null;
}
public Message<?> tooManyParameters(Collection<?> c1, Collection<?> c2) {
public Message<?> tooManyParameters(List<?> c1, List<?> c2) {
return null;
}
@@ -235,7 +234,7 @@ public class AggregatorAdapterTests {
return null;
}
public Message<?> collectionSubclassParameter(List<?> l1){
public Message<?> listSubclassParameter(LinkedList<?> l1){
return null;
}
}