diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java index 536e8eec8d..11b19670a3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java @@ -34,11 +34,19 @@ import org.springframework.integration.router.AggregatingMessageHandler; @Documented @Handler public @interface Aggregator { - String defaultReplyChannel() default ""; - String discardChannel() default ""; - long sendTimeout() default AggregatingMessageHandler.DEFAULT_SEND_TIMEOUT; - long timeout() default AggregatingMessageHandler.DEFAULT_TIMEOUT; - boolean sendPartialResultsOnTimeout() default false; - long reaperInterval() default AggregatingMessageHandler.DEFAULT_REAPER_INTERVAL; - int trackedCorrelationIdCapacity() default AggregatingMessageHandler.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY; + + String defaultReplyChannel() default ""; + + String discardChannel() default ""; + + long sendTimeout() default AggregatingMessageHandler.DEFAULT_SEND_TIMEOUT; + + long timeout() default AggregatingMessageHandler.DEFAULT_TIMEOUT; + + boolean sendPartialResultsOnTimeout() default false; + + long reaperInterval() default AggregatingMessageHandler.DEFAULT_REAPER_INTERVAL; + + int trackedCorrelationIdCapacity() default AggregatingMessageHandler.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY; + } 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 962d35b522..96b430f9c0 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 @@ -76,13 +76,13 @@ public class AggregatorParser implements BeanDefinitionParser { return parseAggregatorElement(element, parserContext, true); } - public static BeanDefinition parseAggregatorElement(Element element, ParserContext parserContext, boolean topLevel) { + + private BeanDefinition parseAggregatorElement(Element element, ParserContext parserContext, boolean topLevel) { final RootBeanDefinition aggregatorDef = new RootBeanDefinition(AggregatingMessageHandler.class); aggregatorDef.setSource(parserContext.extractSource(element)); final String id = element.getAttribute(ID_ATTRIBUTE); final String ref = element.getAttribute(REF_ATTRIBUTE); final String method = element.getAttribute(METHOD_ATTRIBUTE); - if (!StringUtils.hasText(ref)) { throw new MessagingConfigurationException("The 'ref' attribute must be present"); } @@ -101,7 +101,6 @@ public class AggregatorParser implements BeanDefinitionParser { String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDefinition); parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDefinition, adapterBeanName)); aggregatorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(adapterBeanName)); - } IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, COMPLETION_STRATEGY_PROPERTY, element, COMPLETION_STRATEGY_ATTRIBUTE); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java index ed51eb6641..fc4212e002 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java @@ -63,13 +63,14 @@ import org.springframework.util.CollectionUtils; public class AggregatingMessageHandler implements MessageHandler, InitializingBean { public final static long DEFAULT_SEND_TIMEOUT = 1000; - + public final static long DEFAULT_TIMEOUT = 60000; - + public final static long DEFAULT_REAPER_INTERVAL = 1000; - + public final static int DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY = 1000; - + + private final Log logger = LogFactory.getLog(this.getClass()); private final Aggregator aggregator; 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 6b2bc3fc65..5546c00817 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 @@ -21,8 +21,8 @@ import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.util.ArrayList; import java.util.Collection; +import java.util.List; -import org.springframework.integration.MessagingConfigurationException; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.util.SimpleMethodInvoker; @@ -30,10 +30,12 @@ import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; /** - * Aggregator adapter for methods annotated with {@link org.springframework.integration.annotation.Aggregator @Aggregator} and for - * <aggregator ref="beanReference"method="methodName"/> + * Aggregator adapter for methods annotated with {@link org.springframework.integration.annotation.Aggregator @Aggregator} + * and for 'aggregator' elements that include a 'method' attribute + * (e.g. <aggregator ref="beanReference" method="methodName"/>). * * @author Marius Bogoevici + * @author Mark Fisher */ public class AggregatorAdapter implements Aggregator { @@ -41,46 +43,52 @@ public class AggregatorAdapter implements Aggregator { 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."); + Assert.notNull(methodName, "'methodName' must not be null"); this.method = ReflectionUtils.findMethod(object.getClass(), methodName, new Class[] { Collection.class }); - Assert - .notNull(this.method, "Method '" + methodName + "'(Collection args) not found on " - + object.getClass()); - this.invoker = new SimpleMethodInvoker(object, method.getName()); + Assert.notNull(this.method, "Method '" + methodName + "(Collection args)' not found on '" + + object.getClass().getName() + "'."); + this.invoker = new SimpleMethodInvoker(object, this.method.getName()); } 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 MessagingConfigurationException( + Assert.notNull(method, "'method' must not be null"); + if (method.getParameterTypes().length != 1 || !method.getParameterTypes()[0].equals(Collection.class)) { + throw new IllegalArgumentException( "Aggregator method must accept exactly one parameter, and it must be a Collection."); } this.method = method; - this.invoker = new SimpleMethodInvoker(object, method.getName()); + this.invoker = new SimpleMethodInvoker(object, this.method.getName()); } + public Message aggregate(Collection> messages) { Object returnedValue = null; - - if (isMethodParameterParametrized(method) && isHavingActualTypeArguments(method) - && (isActualTypeRawMessage(method) || isActualTypeParametrizedMessage(method))) { - returnedValue = invoker.invokeMethod(messages); + if (isMethodParameterParametrized(this.method) && isHavingActualTypeArguments(this.method) + && (isActualTypeRawMessage(this.method) || isActualTypeParametrizedMessage(this.method))) { + returnedValue = this.invoker.invokeMethod(messages); } else { - returnedValue = invoker.invokeMethod(extractPayloadsFromMessages(messages)); + returnedValue = this.invoker.invokeMethod(extractPayloadsFromMessages(messages)); } if (returnedValue == null) { return null; } - else if (returnedValue instanceof Message) { + if (returnedValue instanceof Message) { return (Message) returnedValue; } - else { - return new GenericMessage(returnedValue); + 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) { @@ -106,12 +114,4 @@ public class AggregatorAdapter implements Aggregator { && method.getGenericParameterTypes()[0] instanceof ParameterizedType; } - private Collection extractPayloadsFromMessages(Collection> messages) { - ArrayList payloadList = new ArrayList(); - for (Message message : messages) { - payloadList.add(message.getPayload()); - } - return payloadList; - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java index 27da5beda7..278e40cd08 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java @@ -26,26 +26,27 @@ import org.springframework.integration.router.AggregatingMessageHandler; import org.springframework.integration.router.AggregatorAdapter; /** - * Creates a {@link AggregatorAdapter AggregatorAdapter} adapter for methods that aggregate messages. + * Creates an {@link AggregatorAdapter AggregatorAdapter} for methods that aggregate messages. * * @author Marius Bogoevici */ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreator { private static final String DEFAULT_REPLY_CHANNEL = "defaultReplyChannel"; - + private static final String DISCARD_CHANNEL = "discardChannel"; - + private static final String SEND_TIMEOUT = "sendTimeout"; - + private static final String SEND_PARTIAL_RESULTS_ON_TIMEOUT = "sendPartialResultsOnTimeout"; - + private static final String REAPER_INTERVAL = "reaperInterval"; - + private static final String TIMEOUT = "timeout"; - + private static final String TRACKED_CORRELATION_ID_CAPACITY = "trackedCorrelationIdCapacity"; + private final MessageBus messageBus; @@ -53,23 +54,31 @@ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreat this.messageBus = messageBus; } + public MessageHandler doCreateHandler(Object object, Method method, Map attributes) { AggregatingMessageHandler messageHandler = new AggregatingMessageHandler(new AggregatorAdapter(object, method)); - if (attributes.containsKey(DEFAULT_REPLY_CHANNEL)) - messageHandler.setDefaultReplyChannel(messageBus.lookupChannel((String)attributes.get(DEFAULT_REPLY_CHANNEL))); - if (attributes.containsKey(DISCARD_CHANNEL)) - messageHandler.setDiscardChannel(messageBus.lookupChannel((String)attributes.get(DISCARD_CHANNEL))); - if (attributes.containsKey(SEND_TIMEOUT)) - messageHandler.setSendTimeout((Long)attributes.get(SEND_TIMEOUT)); - if (attributes.containsKey(SEND_PARTIAL_RESULTS_ON_TIMEOUT)) - messageHandler.setSendPartialResultOnTimeout((Boolean)attributes.get(SEND_PARTIAL_RESULTS_ON_TIMEOUT)); - if (attributes.containsKey(REAPER_INTERVAL)) - messageHandler.setReaperInterval((Long)attributes.get(REAPER_INTERVAL)); - if(attributes.containsKey(TIMEOUT)) - messageHandler.setTimeout((Long)attributes.get(TIMEOUT)); - if(attributes.containsKey(TRACKED_CORRELATION_ID_CAPACITY)) - messageHandler.setTrackedCorrelationIdCapacity((Integer)attributes.get(TRACKED_CORRELATION_ID_CAPACITY)); + if (attributes.containsKey(DEFAULT_REPLY_CHANNEL)) { + messageHandler.setDefaultReplyChannel(this.messageBus.lookupChannel((String) attributes.get(DEFAULT_REPLY_CHANNEL))); + } + if (attributes.containsKey(DISCARD_CHANNEL)) { + messageHandler.setDiscardChannel(this.messageBus.lookupChannel((String) attributes.get(DISCARD_CHANNEL))); + } + if (attributes.containsKey(SEND_TIMEOUT)) { + messageHandler.setSendTimeout((Long) attributes.get(SEND_TIMEOUT)); + } + if (attributes.containsKey(SEND_PARTIAL_RESULTS_ON_TIMEOUT)) { + messageHandler.setSendPartialResultOnTimeout((Boolean) attributes.get(SEND_PARTIAL_RESULTS_ON_TIMEOUT)); + } + if (attributes.containsKey(REAPER_INTERVAL)) { + messageHandler.setReaperInterval((Long) attributes.get(REAPER_INTERVAL)); + } + if (attributes.containsKey(TIMEOUT)) { + messageHandler.setTimeout((Long) attributes.get(TIMEOUT)); + } + if (attributes.containsKey(TRACKED_CORRELATION_ID_CAPACITY)) { + messageHandler.setTrackedCorrelationIdCapacity((Integer) attributes.get(TRACKED_CORRELATION_ID_CAPACITY)); + } return messageHandler; } - -} \ No newline at end of file + +} 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 4831d2bf3b..beb1d3f152 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 @@ -5,22 +5,23 @@ 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-1.0.xsd + 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"> - - - + + + + + + + + + + + - + expression="org\.springframework\.integration\.config\.TestAnnotatedEndpoint.*"/> - - - - - - \ No newline at end of file + 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 360218f59a..bef924558d 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 @@ -17,24 +17,24 @@ package org.springframework.integration.router; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; -import java.util.List; import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import org.springframework.integration.MessagingConfigurationException; + import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; /** * @author Marius Bogoevici + * @author Mark Fisher */ public class AggregatorAdapterTests { private SimpleAggregator simpleAggregator; + @Before public void setUp() { simpleAggregator = new SimpleAggregator(); @@ -87,9 +87,43 @@ public class AggregatorAdapterTests { @Test(expected=IllegalArgumentException.class) public void testAdapterWithWrongMethodName() { - Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "methodThatDoesNotExist"); + new AggregatorAdapter(simpleAggregator, "methodThatDoesNotExist"); } - + + @Test(expected=IllegalArgumentException.class) + public void testInvalidParameterTypeUsingMethodName() { + new AggregatorAdapter(simpleAggregator, "invalidParameterType"); + } + + @Test(expected=IllegalArgumentException.class) + public void testTooManyParametersUsingMethodName() { + new AggregatorAdapter(simpleAggregator, "tooManyParameters"); + } + + @Test(expected=IllegalArgumentException.class) + public void testNotEnoughParametersUsingMethodName() { + new AggregatorAdapter(simpleAggregator, "notEnoughParameters"); + } + + @Test(expected=IllegalArgumentException.class) + public void testInvalidParameterTypeUsingMethodObject() throws SecurityException, NoSuchMethodException { + new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod( + "invalidParameterType", String.class)); + } + + @Test(expected=IllegalArgumentException.class) + public void testTooManyParametersUsingMethodObject() throws SecurityException, NoSuchMethodException { + new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod( + "tooManyParameters", Collection.class, Collection.class)); + } + + @Test(expected=IllegalArgumentException.class) + public void testNotEnoughParametersUsingMethodObject() throws SecurityException, NoSuchMethodException { + new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod( + "notEnoughParameters", null)); + } + + private static Collection> createCollectionOfMessages() { Collection> messages = new ArrayList>(); messages.add(new GenericMessage("123")); @@ -98,17 +132,22 @@ public class AggregatorAdapterTests { return messages; } + private class SimpleAggregator { private volatile boolean aggregationPerformed; public SimpleAggregator() { - aggregationPerformed = false; + this.aggregationPerformed = false; + } + + public boolean isAggregationPerformed() { + return this.aggregationPerformed; } public Message doAggregationOnCollectionOfMessages(Collection messages) { - aggregationPerformed = true; + this.aggregationPerformed = true; StringBuffer buffer = new StringBuffer(); for (Message message : messages) { buffer.append(message.getPayload()); @@ -117,16 +156,16 @@ public class AggregatorAdapterTests { } public Message doAggregationOnCollectionOfMessagesParametrizedWithWildcard(Collection> messages) { - aggregationPerformed = true; + this.aggregationPerformed = true; StringBuffer buffer = new StringBuffer(); for (Message message : messages) { buffer.append(message.getPayload()); } return new GenericMessage(buffer.toString()); } - + public Message doAggregationOnCollectionOfMessagesParametrizedWithString(Collection> messages) { - aggregationPerformed = true; + this.aggregationPerformed = true; StringBuffer buffer = new StringBuffer(); for (Message message : messages) { buffer.append(message.getPayload()); @@ -135,7 +174,7 @@ public class AggregatorAdapterTests { } public Message doAggregationOnCollectionOfStrings(Collection messages) { - aggregationPerformed = true; + this.aggregationPerformed = true; StringBuffer buffer = new StringBuffer(); for (String payload : messages) { buffer.append(payload); @@ -144,7 +183,7 @@ public class AggregatorAdapterTests { } public Long doAggregationOnCollectionOfStringsReturningLong(Collection messages) { - aggregationPerformed = true; + this.aggregationPerformed = true; StringBuffer buffer = new StringBuffer(); for (String payload : messages) { buffer.append(payload); @@ -152,10 +191,17 @@ public class AggregatorAdapterTests { return Long.parseLong(buffer.toString()); } - public boolean isAggregationPerformed() { - return aggregationPerformed; + public Message invalidParameterType(String invalid) { + return null; + } + + public Message tooManyParameters(Collection c1, Collection c2) { + return null; + } + + public Message notEnoughParameters() { + return null; } - } - + }