From 03f4c8e1ea00e00417b3cb1d485f326371c66dfc Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Mon, 3 Mar 2008 23:21:39 +0000 Subject: [PATCH] Adds AggregatorAdapter and support for the 'method' attribute in elements (INT-137). Target methods must conform with a signature pattern as follows: ReturnType methodName (Collection args). If CollectionType is Message or any of its descendants (including parametrized types), the aggregator will pass the collection of messages to be aggregated 'as is', leaving to the target method to extract the payloads. If CollectionType is non-Message, the payloads will be extracted and passed to the target collection. If the result of the method call is not a Message, it will be wrapped in a GenericMessage. --- .../integration/router/Aggregator.java | 4 +- .../integration/router/AggregatorAdapter.java | 121 +++++++++++++ .../integration/config/Adder.java | 34 ++++ .../config/AggregatorParserTests.java | 26 ++- .../integration/config/TestAggregator.java | 10 +- .../config/aggregatorParserTests.xml | 4 + .../config/invalidMethodNameAggregator.xml | 16 ++ .../AggregatingMessageHandlerTests.java | 3 +- .../router/AggregatorAdapterTests.java | 161 ++++++++++++++++++ 9 files changed, 370 insertions(+), 9 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/Adder.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/invalidMethodNameAggregator.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/router/AggregatorAdapterTests.java 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; + } + + } + +}