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

AggregatorEndpoint has been replaced by AbstractMessageAggregator, and the Aggregator interface has been removed. The MethodInvokingAggregator is now capable of detecting a single method with the @Aggregator annotation if no "method" attribute is defined on an <aggregator/> element, and it will fall back to detect a single public Method (else throw a IllegalArgumentException).

Browse Source
This commit is contained in:
Mark Fisher
2008-10-07 03:19:28 +00:00
parent 7991ef889c
commit 05d9528024
13 changed files with 117 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java
View File

@@ -20,46 +20,30 @@ import java.util.List;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
* An {@link AbstractMessageBarrierConsumer} that waits for a <em>complete</em>
* group of {@link Message Messages} to arrive and then delegates to an
* {@link Aggregator} to combine them into a single {@link Message}.
* <p>
* The default strategy for determining whether a group is complete is based on
* the '<code>sequenceSize</code>' property of the header. Alternatively, a
* A base class for aggregating a group of Messages into a single Message.
* Extends {@link AbstractMessageBarrierConsumer} and waits for a
* <em>complete</em> group of {@link Message Messages} to arrive. Subclasses
* must provide the implementation of the {@link #aggregateMessages(List)}
* method to combine the group of Messages into a single {@link Message}.
*
* <p>The default strategy for determining whether a group is complete is based
* on the '<code>sequenceSize</code>' property of the header. Alternatively, a
* custom implementation of the {@link CompletionStrategy} may be provided.
* <p>
* All considerations regarding <code>timeout</code> and grouping by '
* <code>correlationId</code>' from {@link AbstractMessageBarrierConsumer}
*
* <p>All considerations regarding <code>timeout</code> and grouping by
* <code>correlationId</code> from {@link AbstractMessageBarrierConsumer}
* apply here as well.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class AggregatorEndpoint extends AbstractMessageBarrierConsumer {
private final Aggregator aggregator;
public abstract class AbstractMessageAggregator extends AbstractMessageBarrierConsumer {
private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
/**
* Create an endpoint that delegates to the provided Aggregator to combine a
* group of messages into a single message. The executor will be used for
* scheduling a background maintenance thread. If <code>null</code>, a new
* single-threaded executor will be created.
*/
public AggregatorEndpoint(Aggregator aggregator, TaskScheduler executor) {
super();
Assert.notNull(aggregator, "'aggregator' must not be null");
this.aggregator = aggregator;
}
public AggregatorEndpoint(Aggregator aggregator) {
this(aggregator, null);
}
/**
* Strategy to determine whether the group of messages is complete.
@@ -78,7 +62,7 @@ public class AggregatorEndpoint extends AbstractMessageBarrierConsumer {
}
protected Message<?>[] processReleasedMessages(Object correlationId, List<Message<?>> messages) {
Message<?> result = aggregator.aggregate(messages);
Message<?> result = this.aggregateMessages(messages);
if (result == null) {
return new Message<?>[0];
}
@@ -88,4 +72,6 @@ public class AggregatorEndpoint extends AbstractMessageBarrierConsumer {
return new Message<?>[] { result };
}
protected abstract Message<?> aggregateMessages(List<Message<?>> messages);
}

33
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Aggregator.java
View File

@@ -1,33 +0,0 @@
/*
* Copyright 2002-2008 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 java.util.List;
import org.springframework.integration.message.Message;
/**
* Strategy interface for aggregating a list of {@link Message Messages} into a
* single {@link Message}.
*
* @author Mark Fisher
*/
public interface Aggregator {
Message<?> aggregate(List<Message<?>> messages);
}

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java
View File

@@ -35,7 +35,7 @@ import org.springframework.util.ReflectionUtils;
*
* @author Marius Bogoevici
*/
public abstract class MessageListMethodAdapter {
public class MessageListMethodAdapter {
private final DefaultMethodInvoker invoker;

33
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregator.java
View File

@@ -19,34 +19,51 @@ package org.springframework.integration.aggregator;
import java.lang.reflect.Method;
import java.util.List;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.util.DefaultMethodResolver;
import org.springframework.integration.util.MethodResolver;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* Aggregator adapter for methods annotated with {@link org.springframework.integration.annotation.Aggregator @Aggregator}
* and for '<code>aggregator</code>' elements that include a '<code>method</code>' attribute
* (e.g. &lt;aggregator ref="beanReference" method="methodName"/&gt;).
* {@link AbstractMessageAggregator} adapter for methods annotated with
* {@link Aggregator @Aggregator} annotation and for <code>aggregator</code>
* elements (e.g. &lt;aggregator ref="beanReference" method="methodName"/&gt;).
*
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class MethodInvokingAggregator extends MessageListMethodAdapter implements Aggregator {
public class MethodInvokingAggregator extends AbstractMessageAggregator {
private final MethodResolver methodResolver = new DefaultMethodResolver(Aggregator.class);
private final MessageListMethodAdapter methodInvoker;
public MethodInvokingAggregator(Object object, Method method) {
super(object, method);
this.methodInvoker = new MessageListMethodAdapter(object, method);
}
public MethodInvokingAggregator(Object object, String methodName) {
super(object, methodName);
this.methodInvoker = new MessageListMethodAdapter(object, methodName);
}
public MethodInvokingAggregator(Object object) {
Assert.notNull(object, "object must not be null");
Method method = this.methodResolver.findMethod(object.getClass());
Assert.notNull(method, "unable to resolve Aggregator method on target class ["
+ object.getClass() + "]");
this.methodInvoker = new MessageListMethodAdapter(object, method);
}
public Message<?> aggregate(List<Message<?>> messages) {
public Message<?> aggregateMessages(List<Message<?>> messages) {
if (CollectionUtils.isEmpty(messages)) {
return null;
}
Object returnedValue = this.executeMethod(messages);
Object returnedValue = this.methodInvoker.executeMethod(messages);
if (returnedValue == null) {
return null;
}

10
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java
View File

@@ -22,7 +22,7 @@ import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.integration.aggregator.AggregatorEndpoint;
import org.springframework.integration.aggregator.AbstractMessageAggregator;
/**
* Indicates that a method is capable of aggregating messages.
@@ -56,12 +56,12 @@ public @interface Aggregator {
/**
* timeout for sending results to the reply target (in milliseconds)
*/
long sendTimeout() default AggregatorEndpoint.DEFAULT_SEND_TIMEOUT;
long sendTimeout() default AbstractMessageAggregator.DEFAULT_SEND_TIMEOUT;
/**
* maximum time to wait for completion (in milliseconds)
*/
long timeout() default AggregatorEndpoint.DEFAULT_TIMEOUT;
long timeout() default AbstractMessageAggregator.DEFAULT_TIMEOUT;
/**
* indicates whether to send an incomplete aggregate on timeout
@@ -71,13 +71,13 @@ public @interface Aggregator {
/**
* interval for the task that checks for timed-out aggregates
*/
long reaperInterval() default AggregatorEndpoint.DEFAULT_REAPER_INTERVAL;
long reaperInterval() default AbstractMessageAggregator.DEFAULT_REAPER_INTERVAL;
/**
* maximum number of correlation IDs to maintain so that received messages
* may be recognized as belonging to an aggregate that has already completed
* or timed out
*/
int trackedCorrelationIdCapacity() default AggregatorEndpoint.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
int trackedCorrelationIdCapacity() default AbstractMessageAggregator.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
}

12
org.springframework.integration/src/main/java/org/springframework/integration/config/AggregatorParser.java
View File

@@ -21,9 +21,9 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.aggregator.AggregatorEndpoint;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
@@ -56,8 +56,14 @@ public class AggregatorParser extends AbstractEndpointParser {
@Override
protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(AggregatorEndpoint.class);
builder.addConstructorArgReference(this.parseAdapter(element, parserContext, MethodInvokingAggregator.class));
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingAggregator.class);
String ref = element.getAttribute(REF_ATTRIBUTE);
Assert.hasText(ref, "The '" + REF_ATTRIBUTE + "' attribute is required.");
builder.addConstructorArgReference(ref);
if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
String method = element.getAttribute(METHOD_ATTRIBUTE);
builder.addConstructorArgValue(method);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);

7
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -20,7 +20,7 @@ import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.aggregator.AggregatorEndpoint;
import org.springframework.integration.aggregator.AbstractMessageAggregator;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.integration.annotation.Aggregator;
@@ -46,8 +46,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
@Override
protected MessageConsumer createConsumer(Object bean, Method method, Aggregator annotation) {
MethodInvokingAggregator adapter = new MethodInvokingAggregator(bean, method);
AggregatorEndpoint aggregator = new AggregatorEndpoint(adapter);
MethodInvokingAggregator aggregator = new MethodInvokingAggregator(bean, method);
this.configureCompletionStrategy(bean, aggregator);
String discardChannelName = annotation.discardChannel();
if (StringUtils.hasText(discardChannelName)) {
@@ -64,7 +63,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
return aggregator;
}
private void configureCompletionStrategy(final Object bean, final AggregatorEndpoint aggregator) {
private void configureCompletionStrategy(final Object bean, final AbstractMessageAggregator aggregator) {
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class);