Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.

Browse Source
This commit is contained in:
Mark Fisher
2010-04-21 01:00:49 +00:00
parent 9ac3361529
commit 4dfc783d65
4 changed files with 405 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
View File

@@ -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<String, Object> 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<String, Object> aggregateHeaders(MessageGroup group) {
Map<String, Object> aggregatedHeaders = new HashMap<String, Object>();
Set<String> conflictKeys = new HashSet<String>();
List<Message<?>> 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);

10
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java
View File

@@ -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<String, Object> aggregateHeaders(MessageGroup group) {
// TODO: return all non-conflicting headers
return new HashMap<String, Object>();
}
@Override
protected Object aggregatePayloads(MessageGroup group) {
protected final Object aggregatePayloads(MessageGroup group) {
List<Message<?>> messages = group.getMessages();
Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups");
List<Object> payloads = new ArrayList<Object>(messages.size());

310
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java
View File

@@ -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<Method> candidates = new HashSet<Method>(Arrays.asList(methods));
removeObjectMethodsFrom(candidates);
removeVoidMethodsFrom(candidates);
removeListIncompatibleMethodsFrom(candidates);
Set<Method> notAnnotatedCandidates = new HashSet<Method>();
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<Method> 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<Method> candidates) {
removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
return method.getReturnType().getName().equals("void");
}
});
}
private Set<Method> removeNotAnnotatedFrom(Set<Method> candidates) {
return removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
Aggregator annotation = method.getAnnotation(Aggregator.class);
return (annotation == null);
}
});
}
private Set<Method> removeUnfittingFrom(Set<Method> 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<Method> 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<Message<?>> 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<Method> removeMethodsMatchingSelector(Set<Method> candidates, MethodSelector selector) {
Set<Method> removed = new HashSet<Method>();
Iterator<Method> 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<Message<?>> 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<Method> candidates = new HashSet<Method>(Arrays.asList(methods));
removeObjectMethodsFrom(candidates);
removeVoidMethodsFrom(candidates);
removeListIncompatibleMethodsFrom(candidates);
Set<Method> notAnnotatedCandidates = new HashSet<Method>();
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<Method> 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<Method> candidates) {
removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
return method.getReturnType().getName().equals("void");
}
});
}
private Set<Method> removeNotAnnotatedFrom(Set<Method> candidates) {
return removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
Aggregator annotation = method.getAnnotation(Aggregator.class);
return (annotation == null);
}
});
}
private Set<Method> removeUnfittingFrom(Set<Method> 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<Method> candidates) {
removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
return method.getDeclaringClass().equals(Object.class);
}
});
}
private Set<Method> removeMethodsMatchingSelector(Set<Method> candidates, MethodSelector selector) {
Set<Method> removed = new HashSet<Method>();
Iterator<Method> 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);
}
}