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 2aca549b2e..140a435045 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.List; +import java.util.Collection; import org.springframework.integration.message.Message; @@ -28,6 +28,6 @@ import org.springframework.integration.message.Message; */ public interface Aggregator { - Message aggregate(List> messages); + Message aggregate(Collection> 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 new file mode 100644 index 0000000000..b08f81292c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java @@ -0,0 +1,121 @@ +/* + * 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.router; + +import java.lang.reflect.Method; +import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.Collection; + +import org.springframework.integration.MessagingConfigurationException; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.util.SimpleMethodInvoker; +import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; + +/** + * Aggregator adapter for methods used through + * + * @author Marius Bogoevici + */ +public class AggregatorAdapter implements Aggregator { + + private final SimpleMethodInvoker 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 }); + Assert + .notNull(this.method, "Method '" + methodName + "'(Collection args) not found on " + + object.getClass()); + this.invoker = new SimpleMethodInvoker(object, method.getName()); + } + + public AggregatorAdapter(Object object, Method method) { + Assert.notNull(object, "'object' must not be null"); + Assert.notNull(method, "'method' must not be null."); + assertMethodCanBeAdapted(); + this.method = method; + this.invoker = new SimpleMethodInvoker(object, method.getName()); + } + + private void assertMethodCanBeAdapted() { + if (this.method.getParameterTypes().length != 1 && this.method.getParameterTypes()[0].equals(Collection.class)) { + throw new MessagingConfigurationException( + "Aggregator method must accept exactly one parameter, and it must be a Collection."); + } + } + + public Message aggregate(Collection> messages) { + Object returnedValue = null; + + if (isMethodParameterParametrized(method) && isHavingActualTypeArguments(method) + && (isActualTypeRawMessage(method) || isActualTypeParametrizedMessage(method))) { + returnedValue = invoker.invokeMethod(messages); + } + else { + returnedValue = invoker.invokeMethod(extractPayloadsFromMessages(messages)); + } + if (returnedValue == null) { + return null; + } + else if (returnedValue instanceof Message) { + return (Message) returnedValue; + } + else { + return new GenericMessage(returnedValue); + } + } + + 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; + } + + 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/test/java/org/springframework/integration/config/Adder.java b/spring-integration-core/src/test/java/org/springframework/integration/config/Adder.java new file mode 100644 index 0000000000..322781145d --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/Adder.java @@ -0,0 +1,34 @@ +/* + * 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.Collection; + +/** + * @author Marius Bogoevici + */ +public class Adder { + + public Long add(Collection results) { + long total = 0; + for (long partialResult: results) { + total += partialResult; + } + return total; + } + +} 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 52537aebb7..7af5e824c4 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 @@ -24,9 +24,11 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.integration.router.AggregatingMessageHandler; @@ -41,7 +43,6 @@ public class AggregatorParserTests { private ApplicationContext context; - @Before public void setUp() { context = new ClassPathXmlApplicationContext("aggregatorParserTests.xml", this.getClass()); @@ -100,10 +101,29 @@ public class AggregatorParserTests { 99, getPropertyValue(completeAggregatingMessageHandler, "trackedCorrelationIdCapacity", int.class)); } + @Test + public void testSimpleJavaBeanAggregator() { + AggregatingMessageHandler addingAggregator = (AggregatingMessageHandler) context.getBean("aggregatorWithReferenceAndMethod"); + List> outboundMessages = new ArrayList>(); + outboundMessages.add(createMessage(1l, "id1", 3, 1, null)); + outboundMessages.add(createMessage(2l, "id1", 3, 3, null)); + outboundMessages.add(createMessage(3l, "id1", 3, 2, null)); + for (Message message : outboundMessages) { + addingAggregator.handle(message); + } + MessageChannel replyChannel = (MessageChannel) context.getBean("replyChannel"); + Message response = replyChannel.receive(); + Assert.assertEquals(6l, response.getPayload()); + } + + @Test(expected=BeanCreationException.class) + public void testMissingMethodOnAggregator() { + context = new ClassPathXmlApplicationContext("invalidMethodNameAggregator.xml", this.getClass()); + } - private static Message createMessage(String payload, Object correlationId, int sequenceSize, int sequenceNumber, + private static Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, MessageChannel replyChannel) { - StringMessage message = new StringMessage(payload); + GenericMessage message = new GenericMessage(payload); message.getHeader().setCorrelationId(correlationId); message.getHeader().setSequenceSize(sequenceSize); message.getHeader().setSequenceNumber(sequenceNumber); 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 b4daefb50a..2a92e67db7 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,6 +17,7 @@ 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; @@ -34,16 +35,19 @@ public class TestAggregator implements Aggregator { private final ConcurrentMap> aggregatedMessages = new ConcurrentHashMap>(); - - public Message aggregate(List> messages) { + public Message aggregate(Collection> 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(messages.get(0).getHeader().getCorrelationId(), returnedMessage); + aggregatedMessages.put(correlationId, returnedMessage); return returnedMessage; } 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 adef845aa4..76ecb3c554 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 @@ -18,12 +18,16 @@ send-partial-result-on-timeout="true" reaper-interval="135" tracked-correlation-id-capacity="99"/> + + + + 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 new file mode 100644 index 0000000000..ec5293c7f3 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/invalidMethodNameAggregator.xml @@ -0,0 +1,16 @@ + + + + + + + + + + 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 414071d283..93bccb27b6 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 @@ -21,6 +21,7 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -207,7 +208,7 @@ public class AggregatingMessageHandlerTests { private static class TestAggregator implements Aggregator { - public Message aggregate(List> messages) { + public Message aggregate(Collection> 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 new file mode 100644 index 0000000000..360218f59a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatorAdapterTests.java @@ -0,0 +1,161 @@ +/* + * 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.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 + */ +public class AggregatorAdapterTests { + + private SimpleAggregator simpleAggregator; + + @Before + public void setUp() { + simpleAggregator = new SimpleAggregator(); + } + + @Test + public void testAdapterWithMessageCollectionBasedMethod() { + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnCollectionOfMessages"); + Collection> messages = createCollectionOfMessages(); + Message returnedMessge = aggregator.aggregate(messages); + Assert.assertTrue(simpleAggregator.isAggregationPerformed()); + Assert.assertEquals("123456789", returnedMessge.getPayload()); + } + + @Test + public void testAdapterWithWildcardParametrizedMessageBasedMethod() { + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnCollectionOfMessagesParametrizedWithWildcard"); + Collection> messages = createCollectionOfMessages(); + Message returnedMessge = aggregator.aggregate(messages); + Assert.assertTrue(simpleAggregator.isAggregationPerformed()); + Assert.assertEquals("123456789", returnedMessge.getPayload()); + } + + @Test + public void testAdapterWithTypeParametrizedMessageBasedMethod() { + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnCollectionOfMessagesParametrizedWithString"); + Collection> messages = createCollectionOfMessages(); + Message returnedMessge = aggregator.aggregate(messages); + Assert.assertTrue(simpleAggregator.isAggregationPerformed()); + Assert.assertEquals("123456789", returnedMessge.getPayload()); + } + + @Test + public void testAdapterWithPojoBasedMethod() { + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnCollectionOfStrings"); + Collection> messages = createCollectionOfMessages(); + Message returnedMessge = aggregator.aggregate(messages); + Assert.assertTrue(simpleAggregator.isAggregationPerformed()); + Assert.assertEquals("123456789", returnedMessge.getPayload()); + } + + @Test + public void testAdapterWithPojoBasedMethodReturningObject() { + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnCollectionOfStringsReturningLong"); + Collection> messages = createCollectionOfMessages(); + Message returnedMessge = aggregator.aggregate(messages); + Assert.assertTrue(simpleAggregator.isAggregationPerformed()); + Assert.assertEquals(123456789l, returnedMessge.getPayload()); + } + + @Test(expected=IllegalArgumentException.class) + public void testAdapterWithWrongMethodName() { + Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "methodThatDoesNotExist"); + } + + private static Collection> createCollectionOfMessages() { + Collection> messages = new ArrayList>(); + messages.add(new GenericMessage("123")); + messages.add(new GenericMessage("456")); + messages.add(new GenericMessage("789")); + return messages; + } + + private class SimpleAggregator { + + private volatile boolean aggregationPerformed; + + + public SimpleAggregator() { + aggregationPerformed = false; + } + + public Message doAggregationOnCollectionOfMessages(Collection messages) { + aggregationPerformed = true; + StringBuffer buffer = new StringBuffer(); + for (Message message : messages) { + buffer.append(message.getPayload()); + } + return new GenericMessage(buffer.toString()); + } + + public Message doAggregationOnCollectionOfMessagesParametrizedWithWildcard(Collection> messages) { + 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; + StringBuffer buffer = new StringBuffer(); + for (Message message : messages) { + buffer.append(message.getPayload()); + } + return new GenericMessage(buffer.toString()); + } + + public Message doAggregationOnCollectionOfStrings(Collection messages) { + aggregationPerformed = true; + StringBuffer buffer = new StringBuffer(); + for (String payload : messages) { + buffer.append(payload); + } + return new GenericMessage(buffer.toString()); + } + + public Long doAggregationOnCollectionOfStringsReturningLong(Collection messages) { + aggregationPerformed = true; + StringBuffer buffer = new StringBuffer(); + for (String payload : messages) { + buffer.append(payload); + } + return Long.parseLong(buffer.toString()); + } + + public boolean isAggregationPerformed() { + return aggregationPerformed; + } + + } + +}