diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/CompletionStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/CompletionStrategy.java new file mode 100644 index 0000000000..4844b37864 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/CompletionStrategy.java @@ -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 { + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java index 060f8153ee..1a318e040c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java @@ -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 aggregator 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 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; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index 4017572a94..35062f1b95 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -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, MessageHandlerCreator> handlerCreators = - new ConcurrentHashMap, MessageHandlerCreator>(); + private final Map, MessageHandlerCreator> handlerCreators = new ConcurrentHashMap, MessageHandlerCreator>(); private final MessageBus messageBus; - public MessageEndpointAnnotationPostProcessor(MessageBus messageBus) { Assert.notNull(messageBus, "'messageBus' must not be null"); this.messageBus = messageBus; } - - public void setCustomHandlerCreators( - Map, MessageHandlerCreator> customHandlerCreators) { + public void setCustomHandlerCreators(Map, MessageHandlerCreator> customHandlerCreators) { for (Map.Entry, 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 handlers = new ArrayList(); @@ -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 { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index 371d1e3c8d..8e55b36a0e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -234,9 +234,13 @@ + + + + @@ -248,5 +252,17 @@ + + + + + + Defines a completion strategy. + + + + + + \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java index e07020d0af..428473c4bc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java @@ -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 handlers = new CopyOnWriteArrayList(); - /** * 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 getHandlers() { + return Collections.unmodifiableList(handlers); + } + public final Message handle(Message message) { for (MessageHandler next : handlers) { message = next.handle(message); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java b/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java index a161f64ef6..e748e8dc03 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java @@ -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> messages); + Message aggregate(List> messages); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java index 35737f2bfa..de89f84aa1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java @@ -38,45 +38,18 @@ import org.springframework.util.ReflectionUtils; * @author Marius Bogoevici * @author Mark Fisher */ -public class AggregatorAdapter implements Aggregator { - - private final HandlerMethodInvoker 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, 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, this.method.getName()); + super(object, method); } + public AggregatorAdapter(Object object, String methodName) { + super(object, methodName); + } - public Message aggregate(Collection> 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> messages) { + Object returnedValue = this.executeMethod(messages); if (returnedValue == null) { return null; } @@ -86,35 +59,6 @@ public class AggregatorAdapter implements Aggregator { return new GenericMessage(returnedValue); } - private Collection extractPayloadsFromMessages(Collection> messages) { - List payloadList = new ArrayList(); - 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; - } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/CompletionStrategyAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/CompletionStrategyAdapter.java new file mode 100644 index 0000000000..9c58209180 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/CompletionStrategyAdapter.java @@ -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 'aggregator' elements that include a 'method' + * attribute (e.g. <aggregator ref="beanReference" method="methodName"/>). + * + * @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> messages) { + return ((Boolean) executeMethod(messages)).booleanValue(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/MessageListMethodAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/MessageListMethodAdapter.java new file mode 100644 index 0000000000..0a95bee409 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/MessageListMethodAdapter.java @@ -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 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, 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, 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> 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> messages) { + List payloadList = new ArrayList(); + 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; + } + +} \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java index ab5523b532..8ebc74be34 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java @@ -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()); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/Adder.java b/spring-integration-core/src/test/java/org/springframework/integration/config/Adder.java index 322781145d..7cc31adb61 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/Adder.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/Adder.java @@ -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 results) { - long total = 0; + public Long add(List results) { + long total = 0l; for (long partialResult: results) { total += partialResult; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java index c4c37294aa..714a89fd66 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java @@ -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); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index 1e7d2a174e..c9b50c2e48 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -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 Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, MessageChannel replyChannel) { GenericMessage message = new GenericMessage(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); - } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java new file mode 100644 index 0000000000..a1c2edaaee --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java @@ -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; + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregator.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregator.java index d0b8576abe..e17b9b06fe 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregator.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAggregator.java @@ -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> aggregatedMessages = new ConcurrentHashMap>(); - public Message aggregate(Collection> messages) { + public Message aggregate(List> messages) { List> sortableList = new ArrayList>(messages); Collections.sort(sortableList, new MessageSequenceComparator()); StringBuffer buffer = new StringBuffer(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategy.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategy.java new file mode 100644 index 0000000000..28e610013d --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategy.java @@ -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> aggregatedMessages = new ConcurrentHashMap>(); + + @Aggregator + public Message aggregatingMethod(List> messages) { + List> sortableList = new ArrayList>(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> messages) { + return true; + } + + public ConcurrentMap> getAggregatedMessages() { + return aggregatedMessages; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java index 1159c810ae..c5bb2a7b45 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java @@ -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> messages) { + public Message aggregatingMethod(List> messages) { List> sortableList = new ArrayList>(messages); Collections.sort(sortableList, new MessageSequenceComparator()); StringBuffer buffer = new StringBuffer(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java index 3ca344de10..c5106b2178 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java @@ -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> aggregatedMessages = new ConcurrentHashMap>(); @Aggregator - public Message aggregatingMethod(Collection> messages) { + public Message aggregatingMethod(List> messages) { List> sortableList = new ArrayList>(messages); Collections.sort(sortableList, new MessageSequenceComparator()); StringBuffer buffer = new StringBuffer(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml index 49c6e08a6a..66320e89fe 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml @@ -7,29 +7,41 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd"> - + - - - + - - + - - - + + + - + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/completionStrategyMethodWithMissingReference.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/completionStrategyMethodWithMissingReference.xml new file mode 100644 index 0000000000..f65c8f299c --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/completionStrategyMethodWithMissingReference.xml @@ -0,0 +1,22 @@ + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/invalidMethodNameAggregator.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/invalidMethodNameAggregator.xml index ec5293c7f3..ab1a9712ea 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/invalidMethodNameAggregator.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/invalidMethodNameAggregator.xml @@ -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"> - + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml index beb1d3f152..7970222d7a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml @@ -22,6 +22,8 @@ + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml new file mode 100644 index 0000000000..e280fd8fc0 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml @@ -0,0 +1,27 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java index cc753ff51a..566ba00a62 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java @@ -208,7 +208,7 @@ public class AggregatingMessageHandlerTests { private static class TestAggregator implements Aggregator { - public Message aggregate(Collection> messages) { + public Message aggregate(List> messages) { List> sortableList = new ArrayList>(messages); Collections.sort(sortableList, new MessageSequenceComparator()); StringBuffer buffer = new StringBuffer(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatorAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatorAdapterTests.java index 42f00912d5..42bb1b2c06 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatorAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatorAdapterTests.java @@ -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> messages = createCollectionOfMessages(); + public void testAdapterWithNonParameterizedMessageListBasedMethod() { + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnNonParameterizedListOfMessages"); + List> 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> messages = createCollectionOfMessages(); + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfMessagesParametrizedWithWildcard"); + List> 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> messages = createCollectionOfMessages(); + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfMessagesParametrizedWithString"); + List> 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> messages = createCollectionOfMessages(); + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfStrings"); + List> 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> messages = createCollectionOfMessages(); + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfStringsReturningLong"); + List> 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> createCollectionOfMessages() { - Collection> messages = new ArrayList>(); + private static List> createListOfMessages() { + List> messages = new ArrayList>(); messages.add(new GenericMessage("123")); messages.add(new GenericMessage("456")); messages.add(new GenericMessage("789")); @@ -178,7 +177,7 @@ public class AggregatorAdapterTests { } @SuppressWarnings("unchecked") - public Message doAggregationOnNonParameterizedCollectionOfMessages(Collection messages) { + public Message doAggregationOnNonParameterizedListOfMessages(List messages) { this.aggregationPerformed = true; StringBuffer buffer = new StringBuffer(); for (Message message : messages) { @@ -187,7 +186,7 @@ public class AggregatorAdapterTests { return new GenericMessage(buffer.toString()); } - public Message doAggregationOnCollectionOfMessagesParametrizedWithWildcard(Collection> messages) { + public Message doAggregationOnListOfMessagesParametrizedWithWildcard(List> messages) { this.aggregationPerformed = true; StringBuffer buffer = new StringBuffer(); for (Message message : messages) { @@ -196,7 +195,7 @@ public class AggregatorAdapterTests { return new GenericMessage(buffer.toString()); } - public Message doAggregationOnCollectionOfMessagesParametrizedWithString(Collection> messages) { + public Message doAggregationOnListOfMessagesParametrizedWithString(List> messages) { this.aggregationPerformed = true; StringBuffer buffer = new StringBuffer(); for (Message message : messages) { @@ -205,7 +204,7 @@ public class AggregatorAdapterTests { return new GenericMessage(buffer.toString()); } - public Message doAggregationOnCollectionOfStrings(Collection messages) { + public Message doAggregationOnListOfStrings(List messages) { this.aggregationPerformed = true; StringBuffer buffer = new StringBuffer(); for (String payload : messages) { @@ -214,7 +213,7 @@ public class AggregatorAdapterTests { return new GenericMessage(buffer.toString()); } - public Long doAggregationOnCollectionOfStringsReturningLong(Collection messages) { + public Long doAggregationOnListOfStringsReturningLong(List 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; } }