From 4dfc783d651a9e52f605be86d15f1d83788dc9f9 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 21 Apr 2010 01:00:49 +0000 Subject: [PATCH] INT-1059 Both DefaultAggregatingMessageGroupProcessor and MethodInvokingMessageGroupProcessor now extend AbstractAggregatingMessageGroupProcessor. The base class provides default header aggregation. For more detail, see its JavaDoc and the AggregatingMessageGroupProcessorHeaderTests. --- ...tractAggregatingMessageGroupProcessor.java | 50 ++- ...faultAggregatingMessageGroupProcessor.java | 10 +- .../MethodInvokingMessageGroupProcessor.java | 310 +++++++++--------- ...atingMessageGroupProcessorHeaderTests.java | 200 +++++++++++ 4 files changed, 405 insertions(+), 165 deletions(-) create mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index 45862fb1e4..7a1dd17d4d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -16,11 +16,19 @@ package org.springframework.integration.aggregator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; import java.util.Map; +import java.util.Set; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.MessageBuilder; import org.springframework.util.Assert; @@ -35,6 +43,9 @@ import org.springframework.util.Assert; */ public abstract class AbstractAggregatingMessageGroupProcessor implements MessageGroupProcessor { + private final Log logger = LogFactory.getLog(this.getClass()); + + public final void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { Assert.notNull(group, "MessageGroup must not be null"); Assert.notNull(outputChannel, "'outputChannel' must not be null"); @@ -44,7 +55,44 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag channelTemplate.send(message, outputChannel); } - protected abstract Map aggregateHeaders(MessageGroup group); + /** + * This default implementation simply returns all headers that have no conflicts + * among the group. An absent header on one or more Messages within the group is + * not considered a conflict. Subclasses may override this method with more + * advanced conflict-resolution strategies if necessary. + */ + protected Map aggregateHeaders(MessageGroup group) { + Map aggregatedHeaders = new HashMap(); + Set conflictKeys = new HashSet(); + List> messages = group.getMessages(); + if (messages != null) { + for (Message message : messages) { + MessageHeaders currentHeaders = message.getHeaders(); + for (String key : currentHeaders.keySet()) { + if (MessageHeaders.ID.equals(key) || + MessageHeaders.TIMESTAMP.equals(key) || + MessageHeaders.SEQUENCE_SIZE.equals(key)) { + continue; + } + Object value = currentHeaders.get(key); + if (!aggregatedHeaders.containsKey(key)) { + aggregatedHeaders.put(key, value); + } + else if (!value.equals(aggregatedHeaders.get(key))) { + conflictKeys.add(key); + } + } + } + for (String keyToRemove : conflictKeys) { + if (logger.isInfoEnabled()) { + logger.info("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) " + + "in MessageGroup with correlation key: " + group.getCorrelationKey()); + } + aggregatedHeaders.remove(keyToRemove); + } + } + return aggregatedHeaders; + } protected abstract Object aggregatePayloads(MessageGroup group); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java index 0da367891d..82e28fa78b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java @@ -17,9 +17,7 @@ package org.springframework.integration.aggregator; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; -import java.util.Map; import org.springframework.integration.core.Message; import org.springframework.util.Assert; @@ -36,13 +34,7 @@ import org.springframework.util.Assert; public class DefaultAggregatingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor { @Override - protected Map aggregateHeaders(MessageGroup group) { - // TODO: return all non-conflicting headers - return new HashMap(); - } - - @Override - protected Object aggregatePayloads(MessageGroup group) { + protected final Object aggregatePayloads(MessageGroup group) { List> messages = group.getMessages(); Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups"); List payloads = new ArrayList(messages.size()); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java index 0ed111cb59..75deecd4e4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java @@ -16,171 +16,171 @@ package org.springframework.integration.aggregator; -import org.springframework.integration.annotation.Aggregator; -import org.springframework.integration.annotation.Header; -import org.springframework.integration.channel.MessageChannelTemplate; -import org.springframework.integration.core.Message; -import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.message.MessageBuilder; -import org.springframework.util.Assert; - import java.lang.annotation.Annotation; import java.lang.reflect.Method; -import java.util.*; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Set; + +import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.annotation.Header; +import org.springframework.integration.core.Message; +import org.springframework.util.Assert; /** - * MessageGroupProcessor that serves as a wrapper around a POJO. - * + * MessageGroupProcessor that serves as an adapter for the invocation of a POJO method. + * * @author Iwein Fuld - * @since 2.0.0 + * @author Mark Fisher + * @since 2.0 */ -public class MethodInvokingMessageGroupProcessor implements MessageGroupProcessor { +public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor { + + private final MessageListMethodAdapter adapter; - private final MessageListMethodAdapter adapter; + /** + * Creates a wrapper around the target passed in. This constructor will + * choose the best fitting method and throw an exception when methods are + * ambiguous or no fitting methods can be found. + * + * @param target the object to wrap + * @throws IllegalStateException when no single method can be found unambiguously + */ + public MethodInvokingMessageGroupProcessor(Object target) { + this.adapter = new MessageListMethodAdapter(target, this.selectMethodFrom(target)); + } - /** - * Creates a wrapper around the target passed in. This constructor will choose the best fitting method and throw an - * exception when methods are ambiguous or no fitting methods can be found. - * - * @param target the object to wrap - * @throws IllegalStateException when no single method can be found unambiguously - */ - public MethodInvokingMessageGroupProcessor(Object target) { - this.adapter = new MessageListMethodAdapter(target, selectMethodFrom(target)); - } - - /** - * Creates a wrapper around the object passed in. This constructor will look for a named method specifically and - * fail when it cannot find a method with the given name. - * - * @param target the object to wrap - * @param method the name of the method to look for - */ - public MethodInvokingMessageGroupProcessor(Object target, String method) { - this.adapter = new MessageListMethodAdapter(target, method); - } - - private Method selectMethodFrom(Object target) { - Method[] methods = target.getClass().getMethods(); - Set candidates = new HashSet(Arrays.asList(methods)); - - removeObjectMethodsFrom(candidates); - removeVoidMethodsFrom(candidates); - removeListIncompatibleMethodsFrom(candidates); - Set notAnnotatedCandidates = new HashSet(); - if (candidates.size() > 1) { - notAnnotatedCandidates.addAll(removeNotAnnotatedFrom(candidates)); - } - //if no methods are annotated we need to look in more detail in the unannotated methods - if (candidates.size() < 1) { - candidates = notAnnotatedCandidates; - removeUnfittingFrom(candidates); - } - - Assert.state(candidates.size() == 1, - "Method selection failed, there should be exactly one candidate, found [" - + candidates + "]"); - return candidates.iterator().next(); - } - - private void removeListIncompatibleMethodsFrom(Set candidates) { - removeMethodsMatchingSelector(candidates, new MethodSelector() { - public boolean select(Method method) { - int found = 0; - for (Class parameterClass : method.getParameterTypes()) { - if (parameterClass.isAssignableFrom(List.class)) { - found++; - } - } - return found != 1; - } - }); - } - - private void removeVoidMethodsFrom(Set candidates) { - removeMethodsMatchingSelector(candidates, new MethodSelector() { - public boolean select(Method method) { - return method.getReturnType().getName().equals("void"); - } - }); - } - - private Set removeNotAnnotatedFrom(Set candidates) { - return removeMethodsMatchingSelector(candidates, new MethodSelector() { - public boolean select(Method method) { - Aggregator annotation = method.getAnnotation(Aggregator.class); - return (annotation == null); - } - }); - } - - private Set removeUnfittingFrom(Set candidates) { - return removeMethodsMatchingSelector(candidates, new MethodSelector() { - public boolean select(Method method) { - Annotation[][] parameterAnnotations = method.getParameterAnnotations(); - Class[] parameterTypes = method.getParameterTypes(); - return (!isFittinglyAnnotated(parameterTypes, parameterAnnotations)); - } - }); - } - - private boolean isFittinglyAnnotated(Class[] parameterTypes, Annotation[][] parameterAnnotations) { - int candidateParametersFound = 0; - for (int i = 0; i < parameterTypes.length; i++) { - Class parameterType = parameterTypes[i]; - if (parameterType.isAssignableFrom(List.class)) { - boolean headerAnnotationFound = false; - for (Annotation annotation : parameterAnnotations[i]) { - if (annotation instanceof Header) { - headerAnnotationFound = true; - } - } - if (!headerAnnotationFound) { - candidateParametersFound++; - } - } - } - return candidateParametersFound == 1; - } - - private void removeObjectMethodsFrom(Set candidates) { - removeMethodsMatchingSelector(candidates, new MethodSelector() { - public boolean select(Method method) { - return method.getDeclaringClass().equals(Object.class); - } - }); - } - - public void processAndSend(MessageGroup group, - MessageChannelTemplate channelTemplate, MessageChannel outputChannel) { - final Collection> messagesUpForProcessing = group.getMessages(); - Message reply = MessageBuilder.withPayload( - this.adapter.executeMethod(messagesUpForProcessing)).build(); - - group.onCompletion(); - group.onProcessingOf(messagesUpForProcessing - .toArray(new Message[messagesUpForProcessing.size()])); - - channelTemplate.send(reply, outputChannel); - } - - private Set removeMethodsMatchingSelector(Set candidates, MethodSelector selector) { - Set removed = new HashSet(); - Iterator iterator = candidates.iterator(); - while (iterator.hasNext()) { - Method method = iterator.next(); - if (selector.select(method)) { - iterator.remove(); - removed.add(method); - } - } - return removed; - } + /** + * Creates a wrapper around the object passed in. This constructor will look + * for a named method specifically and fail when it cannot find a method + * with the given name. + * + * @param target the object to wrap + * @param method the name of the method to look for + */ + public MethodInvokingMessageGroupProcessor(Object target, String method) { + this.adapter = new MessageListMethodAdapter(target, method); + } - private interface MethodSelector { - boolean select(Method method); - } + @Override + protected final Object aggregatePayloads(MessageGroup group) { + final Collection> messagesUpForProcessing = group.getMessages(); + Object result = this.adapter.executeMethod(messagesUpForProcessing); + group.onCompletion(); + group.onProcessingOf(messagesUpForProcessing.toArray(new Message[messagesUpForProcessing.size()])); + return result; + } + + private Method selectMethodFrom(Object target) { + Method[] methods = target.getClass().getMethods(); + Set candidates = new HashSet(Arrays.asList(methods)); + + removeObjectMethodsFrom(candidates); + removeVoidMethodsFrom(candidates); + removeListIncompatibleMethodsFrom(candidates); + Set notAnnotatedCandidates = new HashSet(); + if (candidates.size() > 1) { + notAnnotatedCandidates.addAll(removeNotAnnotatedFrom(candidates)); + } + + // if no methods are annotated we need to look in more detail in the unannotated methods + if (candidates.size() < 1) { + candidates = notAnnotatedCandidates; + removeUnfittingFrom(candidates); + } + Assert.state(candidates.size() == 1, + "Method selection failed, there should be exactly one candidate, found [" + candidates + "]"); + return candidates.iterator().next(); + } + + private void removeListIncompatibleMethodsFrom(Set candidates) { + removeMethodsMatchingSelector(candidates, new MethodSelector() { + public boolean select(Method method) { + int found = 0; + for (Class parameterClass : method.getParameterTypes()) { + if (parameterClass.isAssignableFrom(List.class)) { + found++; + } + } + return found != 1; + } + }); + } + + private void removeVoidMethodsFrom(Set candidates) { + removeMethodsMatchingSelector(candidates, new MethodSelector() { + public boolean select(Method method) { + return method.getReturnType().getName().equals("void"); + } + }); + } + + private Set removeNotAnnotatedFrom(Set candidates) { + return removeMethodsMatchingSelector(candidates, new MethodSelector() { + public boolean select(Method method) { + Aggregator annotation = method.getAnnotation(Aggregator.class); + return (annotation == null); + } + }); + } + + private Set removeUnfittingFrom(Set candidates) { + return removeMethodsMatchingSelector(candidates, new MethodSelector() { + public boolean select(Method method) { + Annotation[][] parameterAnnotations = method.getParameterAnnotations(); + Class[] parameterTypes = method.getParameterTypes(); + return (!isFittinglyAnnotated(parameterTypes, parameterAnnotations)); + } + }); + } + + private boolean isFittinglyAnnotated(Class[] parameterTypes, Annotation[][] parameterAnnotations) { + int candidateParametersFound = 0; + for (int i = 0; i < parameterTypes.length; i++) { + Class parameterType = parameterTypes[i]; + if (parameterType.isAssignableFrom(List.class)) { + boolean headerAnnotationFound = false; + for (Annotation annotation : parameterAnnotations[i]) { + if (annotation instanceof Header) { + headerAnnotationFound = true; + } + } + if (!headerAnnotationFound) { + candidateParametersFound++; + } + } + } + return candidateParametersFound == 1; + } + + private void removeObjectMethodsFrom(Set candidates) { + removeMethodsMatchingSelector(candidates, new MethodSelector() { + public boolean select(Method method) { + return method.getDeclaringClass().equals(Object.class); + } + }); + } + + private Set removeMethodsMatchingSelector(Set candidates, MethodSelector selector) { + Set removed = new HashSet(); + Iterator iterator = candidates.iterator(); + while (iterator.hasNext()) { + Method method = iterator.next(); + if (selector.select(method)) { + iterator.remove(); + removed.add(method); + } + } + return removed; + } + + private interface MethodSelector { + boolean select(Method method); + } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java new file mode 100644 index 0000000000..7e5ac617b5 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java @@ -0,0 +1,200 @@ +/* + * Copyright 2002-2010 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.aggregator; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.junit.Test; + +import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageBuilder; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class AggregatingMessageGroupProcessorHeaderTests { + + private final QueueChannel outputChannel = new QueueChannel(1); + + private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + + private final MessageGroupProcessor defaultProcessor = new DefaultAggregatingMessageGroupProcessor(); + + private final MessageGroupProcessor methodInvokingProcessor = + new MethodInvokingMessageGroupProcessor(new TestAggregatorBean(), "aggregate"); + + + @Test + public void singleMessageUsingDefaultProcessor() { + this.singleMessage(defaultProcessor); + } + + @Test + public void singleMessageUsingMethodInvokingProcessor() { + this.singleMessage(methodInvokingProcessor); + } + + @Test + public void twoMessagesWithoutConflictsUsingDefaultProcessor() { + this.twoMessagesWithoutConflicts(defaultProcessor); + } + + @Test + public void twoMessagesWithoutConflictsUsingMethodInvokingProcessor() { + this.twoMessagesWithoutConflicts(methodInvokingProcessor); + } + + @Test + public void twoMessagesWithConflictsUsingDefaultProcessor() { + this.twoMessagesWithConflicts(defaultProcessor); + } + + @Test + public void twoMessagesWithConflictsUsingMethodInvokingProcessor() { + this.twoMessagesWithConflicts(methodInvokingProcessor); + } + + @Test + public void missingValuesDoNotConflictUsingDefaultProcessor() { + this.missingValuesDoNotConflict(defaultProcessor); + } + + @Test + public void missingValuesDoNotConflictUsingMethodInvokingProcessor() { + this.missingValuesDoNotConflict(methodInvokingProcessor); + } + + + private void singleMessage(MessageGroupProcessor processor) { + Map headers = new HashMap(); + headers.put("k1", "value1"); + headers.put("k2", new Integer(2)); + Message message = correlatedMessage(1, 1, 1, headers); + List> messages = Collections.>singletonList(message); + MessageGroup group = new MessageGroup(messages, new SequenceSizeCompletionStrategy(), 1); + processor.processAndSend(group, channelTemplate, outputChannel); + Message result = outputChannel.receive(0); + assertNotNull(result); + assertEquals("value1", result.getHeaders().get("k1")); + assertEquals(2, result.getHeaders().get("k2")); + } + + public void twoMessagesWithoutConflicts(MessageGroupProcessor processor) { + Map headers = new HashMap(); + headers.put("k1", "value1"); + headers.put("k2", new Integer(2)); + Message message1 = correlatedMessage(1, 2, 1, headers); + Message message2 = correlatedMessage(1, 2, 2, headers); + List> messages = Arrays.>asList(message1, message2); + MessageGroup group = new MessageGroup(messages, new SequenceSizeCompletionStrategy(), 1); + processor.processAndSend(group, channelTemplate, outputChannel); + Message result = outputChannel.receive(0); + assertNotNull(result); + assertEquals("value1", result.getHeaders().get("k1")); + assertEquals(2, result.getHeaders().get("k2")); + } + + public void twoMessagesWithConflicts(MessageGroupProcessor processor) { + Map headers1 = new HashMap(); + headers1.put("k1", "foo"); + headers1.put("k2", new Integer(123)); + Message message1 = correlatedMessage(1, 2, 1, headers1); + Map headers2 = new HashMap(); + headers2.put("k1", "bar"); + headers2.put("k2", new Integer(123)); + Message message2 = correlatedMessage(1, 2, 2, headers2); + List> messages = Arrays.>asList(message1, message2); + MessageGroup group = new MessageGroup(messages, new SequenceSizeCompletionStrategy(), 1); + processor.processAndSend(group, channelTemplate, outputChannel); + Message result = outputChannel.receive(0); + assertNotNull(result); + assertNull(result.getHeaders().get("k1")); + assertEquals(123, result.getHeaders().get("k2")); + } + + public void missingValuesDoNotConflict(MessageGroupProcessor processor) { + Map headers1 = new HashMap(); + headers1.put("only1", "value1"); + headers1.put("commonTo1And2", "foo"); + headers1.put("commonToAll", new Integer(123)); + headers1.put("conflictBetween1And2", "valueFor1"); + Message message1 = correlatedMessage(1, 3, 1, headers1); + Map headers2 = new HashMap(); + headers2.put("only2", "value2"); + headers2.put("commonTo1And2", "foo"); + headers2.put("commonTo2And3", "bar"); + headers2.put("conflictBetween1And2", "valueFor2"); + headers2.put("conflictBetween2And3", "valueFor2"); + headers2.put("commonToAll", new Integer(123)); + Message message2 = correlatedMessage(1, 3, 2, headers2); + Map headers3 = new HashMap(); + headers3.put("only3", "value3"); + headers3.put("commonTo2And3", "bar"); + headers3.put("commonToAll", new Integer(123)); + headers3.put("conflictBetween2And3", "valueFor3"); + Message message3 = correlatedMessage(1, 3, 3, headers3); + List> messages = Arrays.>asList(message1, message2, message3); + MessageGroup group = new MessageGroup(messages, new SequenceSizeCompletionStrategy(), 1); + processor.processAndSend(group, channelTemplate, outputChannel); + Message result = outputChannel.receive(0); + assertNotNull(result); + assertEquals("value1", result.getHeaders().get("only1")); + assertEquals("value2", result.getHeaders().get("only2")); + assertEquals("value3", result.getHeaders().get("only3")); + assertEquals("foo", result.getHeaders().get("commonTo1And2")); + assertEquals("bar", result.getHeaders().get("commonTo2And3")); + assertEquals(123, result.getHeaders().get("commonToAll")); + assertNull(result.getHeaders().get("conflictBetween1And2")); + assertNull(result.getHeaders().get("conflictBetween2And3")); + } + + + private Message correlatedMessage(Object correlationId, Integer sequenceSize, + Integer sequenceNumber, Map headers) { + return MessageBuilder.withPayload("test") + .setCorrelationId(correlationId) + .setSequenceNumber(sequenceNumber) + .setSequenceSize(sequenceSize) + .copyHeadersIfAbsent(headers) + .build(); + } + + + private static class TestAggregatorBean { + + @SuppressWarnings("unused") + public Object aggregate(List payloads) { + StringBuilder sb = new StringBuilder(); + for (String s : payloads) { + sb.append(s); + } + return sb.toString(); + } + } + +}