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

Renamed AggregatingMessageHandler to AggregatorEndpoint and AggregatorAdapter to MethodInvokingAggregator.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-04 21:00:26 +00:00
parent 3a5bfca9f9
commit 84073697e9
12 changed files with 104 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java
View File

@@ -32,7 +32,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.AbstractInOutEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
@@ -43,12 +43,12 @@ import org.springframework.util.ObjectUtils;
/**
* Base class for {@link MessageBarrier}-based MessageHandlers.
* A {@link MessageHandler} implementation that waits for a group of
* A {@link MessageEndpoint} implementation that waits for a group of
* {@link Message Messages} to arrive and processes them together.
* Uses a {@link MessageBarrier} to store messages and to decide how
* the messages should be released.
* <p>
* Each {@link Message} that is received by this handler will be associated with
* Each {@link Message} that is received by this endpoint will be associated with
* a group based upon the '<code>correlationId</code>' property of its
* header. If no such property is available, a {@link MessageHandlingException}
* will be thrown.
@@ -62,7 +62,7 @@ import org.springframework.util.ObjectUtils;
* @author Mark Fisher
* @author Marius Bogoevici
*/
public abstract class AbstractMessageBarrierHandler extends AbstractInOutEndpoint implements InitializingBean {
public abstract class AbstractMessageBarrierEndpoint extends AbstractInOutEndpoint implements InitializingBean {
public final static long DEFAULT_SEND_TIMEOUT = 1000;
@@ -96,7 +96,7 @@ public abstract class AbstractMessageBarrierHandler extends AbstractInOutEndpoin
private volatile boolean initialized;
public AbstractMessageBarrierHandler(ScheduledExecutorService executor) {
public AbstractMessageBarrierEndpoint(ScheduledExecutorService executor) {
this.executor = (executor != null) ? executor : Executors.newSingleThreadScheduledExecutor();
}

12
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorEndpoint.java
View File

@@ -25,7 +25,7 @@ import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* An {@link AbstractMessageBarrierHandler} that waits for a <em>complete</em>
* An {@link AbstractMessageBarrierEndpoint} 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>
@@ -34,13 +34,13 @@ import org.springframework.util.CollectionUtils;
* custom implementation of the {@link CompletionStrategy} may be provided.
* <p>
* All considerations regarding <code>timeout</code> and grouping by
* '<code>correlationId</code>' from {@link AbstractMessageBarrierHandler} apply
* '<code>correlationId</code>' from {@link AbstractMessageBarrierEndpoint} apply
* here as well.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class AggregatingMessageHandler extends AbstractMessageBarrierHandler {
public class AggregatorEndpoint extends AbstractMessageBarrierEndpoint {
private final Aggregator aggregator;
@@ -48,18 +48,18 @@ public class AggregatingMessageHandler extends AbstractMessageBarrierHandler {
/**
* Create a handler that delegates to the provided aggregator to combine a
* 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 AggregatingMessageHandler(Aggregator aggregator, ScheduledExecutorService executor) {
public AggregatorEndpoint(Aggregator aggregator, ScheduledExecutorService executor) {
super(executor);
Assert.notNull(aggregator, "'aggregator' must not be null");
this.aggregator = aggregator;
}
public AggregatingMessageHandler(Aggregator aggregator) {
public AggregatorEndpoint(Aggregator aggregator) {
this(aggregator, null);
}

6
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorAdapter.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregator.java
View File

@@ -31,13 +31,13 @@ import org.springframework.integration.message.Message;
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class AggregatorAdapter extends MessageListMethodAdapter implements Aggregator {
public class MethodInvokingAggregator extends MessageListMethodAdapter implements Aggregator {
public AggregatorAdapter(Object object, Method method) {
public MethodInvokingAggregator(Object object, Method method) {
super(object, method);
}
public AggregatorAdapter(Object object, String methodName) {
public MethodInvokingAggregator(Object object, String methodName) {
super(object, methodName);
}

6
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java
View File

@@ -23,7 +23,7 @@ import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHeaders;
/**
* An {@link AbstractMessageBarrierHandler} that waits for a group of
* An {@link AbstractMessageBarrierEndpoint} that waits for a group of
* {@link Message Messages} to arrive and re-sends them in order, sorted
* by their <code>sequenceNumber</code>.
* <p>
@@ -31,12 +31,12 @@ import org.springframework.integration.message.MessageHeaders;
* wait for the whole sequence to arrive before re-sending them.
* <p>
* All considerations regarding <code>timeout</code> and grouping by
* '<code>correlationId</code>' from {@link AbstractMessageBarrierHandler} apply
* '<code>correlationId</code>' from {@link AbstractMessageBarrierEndpoint} apply
* here as well.
*
* @author Marius Bogoevici
*/
public class ResequencingMessageHandler extends AbstractMessageBarrierHandler {
public class ResequencingMessageHandler extends AbstractMessageBarrierEndpoint {
private volatile boolean releasePartialSequences = true;

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.AggregatingMessageHandler;
import org.springframework.integration.aggregator.AggregatorEndpoint;
/**
* 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 AggregatingMessageHandler.DEFAULT_SEND_TIMEOUT;
long sendTimeout() default AggregatorEndpoint.DEFAULT_SEND_TIMEOUT;
/**
* maximum time to wait for completion (in milliseconds)
*/
long timeout() default AggregatingMessageHandler.DEFAULT_TIMEOUT;
long timeout() default AggregatorEndpoint.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 AggregatingMessageHandler.DEFAULT_REAPER_INTERVAL;
long reaperInterval() default AggregatorEndpoint.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 AggregatingMessageHandler.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
int trackedCorrelationIdCapacity() default AggregatorEndpoint.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
}

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

@@ -21,8 +21,8 @@ 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.AggregatingMessageHandler;
import org.springframework.integration.aggregator.AggregatorAdapter;
import org.springframework.integration.aggregator.AggregatorEndpoint;
import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.util.StringUtils;
@@ -59,12 +59,12 @@ public class AggregatorParser extends AbstractEndpointParser {
@Override
protected Class<? extends MessageEndpoint> getEndpointClass() {
return AggregatingMessageHandler.class;
return AggregatorEndpoint.class;
}
@Override
protected Class<?> getMethodInvokingAdapterClass() {
return AggregatorAdapter.class;
return MethodInvokingAggregator.class;
}
@Override

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

@@ -21,8 +21,8 @@ import java.lang.reflect.Method;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.AggregatorAdapter;
import org.springframework.integration.aggregator.AggregatorEndpoint;
import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CompletionStrategy;
@@ -47,13 +47,13 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
@Override
protected Object createMethodInvokingAdapter(Object bean, Method method, Aggregator annotation) {
return new AggregatorAdapter(bean, method);
return new MethodInvokingAggregator(bean, method);
}
@Override
protected AbstractEndpoint createEndpoint(Object originalBean, Object adapter) {
if (adapter instanceof org.springframework.integration.aggregator.Aggregator) {
AggregatingMessageHandler endpoint = new AggregatingMessageHandler((org.springframework.integration.aggregator.Aggregator) adapter);
AggregatorEndpoint endpoint = new AggregatorEndpoint((org.springframework.integration.aggregator.Aggregator) adapter);
this.configureCompletionStrategy(originalBean, endpoint);
return endpoint;
}
@@ -63,7 +63,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
@Override
protected void configureEndpoint(AbstractEndpoint endpoint, Aggregator annotation, Poller pollerAnnotation) {
super.configureEndpoint(endpoint, annotation, pollerAnnotation);
AggregatingMessageHandler aggregatorEndpoint = (AggregatingMessageHandler) endpoint;
AggregatorEndpoint aggregatorEndpoint = (AggregatorEndpoint) endpoint;
String discardChannelName = annotation.discardChannel();
if (StringUtils.hasText(discardChannelName)) {
MessageChannel discardChannel = this.getChannelRegistry().lookupChannel(discardChannelName);
@@ -80,7 +80,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
aggregatorEndpoint.afterPropertiesSet();
}
private void configureCompletionStrategy(final Object object, final AggregatingMessageHandler handler) {
private void configureCompletionStrategy(final Object object, final AggregatorEndpoint handler) {
ReflectionUtils.doWithMethods(object.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class);

26
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageHandlerTests.java → org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java
View File

@@ -39,12 +39,12 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
/**
* @author Mark Fisher
*/
public class AggregatingMessageHandlerTests {
public class AggregatorEndpointTests {
private final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
public AggregatingMessageHandlerTests() {
public AggregatorEndpointTests() {
this.executor.setMaxPoolSize(10);
this.executor.setQueueCapacity(0);
this.executor.afterPropertiesSet();
@@ -53,7 +53,7 @@ public class AggregatingMessageHandlerTests {
@Test
public void testCompleteGroupWithinTimeout() throws InterruptedException {
AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator());
AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 3, 2, replyChannel);
@@ -71,7 +71,7 @@ public class AggregatingMessageHandlerTests {
@Test
public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException {
QueueChannel discardChannel = new QueueChannel();
AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator());
AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
aggregator.setTimeout(50);
aggregator.setReaperInterval(10);
aggregator.setDiscardChannel(discardChannel);
@@ -91,7 +91,7 @@ public class AggregatingMessageHandlerTests {
@Test
public void testShouldSendPartialResultOnTimeoutTrue() throws InterruptedException {
AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator());
AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
aggregator.setTimeout(500);
aggregator.setReaperInterval(10);
aggregator.setSendPartialResultOnTimeout(true);
@@ -114,7 +114,7 @@ public class AggregatingMessageHandlerTests {
@Test
public void testMultipleGroupsSimultaneously() throws InterruptedException {
AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator());
AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
QueueChannel replyChannel1 = new QueueChannel();
QueueChannel replyChannel2 = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel1);
@@ -143,7 +143,7 @@ public class AggregatingMessageHandlerTests {
public void testDiscardChannelForTrackedCorrelationId() {
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator());
AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
aggregator.setDiscardChannel(discardChannel);
aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel));
assertEquals("test-1a", replyChannel.receive(100).getPayload());
@@ -155,7 +155,7 @@ public class AggregatingMessageHandlerTests {
public void testTrackedCorrelationIdsCapacityAtLimit() {
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator());
AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
aggregator.setTrackedCorrelationIdCapacity(3);
aggregator.setDiscardChannel(discardChannel);
aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel));
@@ -172,7 +172,7 @@ public class AggregatingMessageHandlerTests {
public void testTrackedCorrelationIdsCapacityPassesLimit() {
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator());
AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
aggregator.setTrackedCorrelationIdCapacity(3);
aggregator.setDiscardChannel(discardChannel);
aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel));
@@ -190,14 +190,14 @@ public class AggregatingMessageHandlerTests {
@Test(expected=MessageHandlingException.class)
public void testExceptionThrownIfNoCorrelationId() throws InterruptedException {
AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator());
AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
Message<?> message = createMessage("123", null, 2, 1, new QueueChannel());
aggregator.handle(message);
}
@Test
public void testAdditionalMessageAfterCompletion() throws InterruptedException {
AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator());
AggregatorEndpoint aggregator = new AggregatorEndpoint(new TestAggregator());
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 3, 2, replyChannel);
@@ -243,7 +243,7 @@ public class AggregatingMessageHandlerTests {
private static class AggregatorTestTask implements Runnable {
private AggregatingMessageHandler aggregator;
private AggregatorEndpoint aggregator;
private Message<?> message;
@@ -252,7 +252,7 @@ public class AggregatingMessageHandlerTests {
private CountDownLatch latch;
AggregatorTestTask(AggregatingMessageHandler aggregator, Message<?> message, CountDownLatch latch) {
AggregatorTestTask(AggregatorEndpoint aggregator, Message<?> message, CountDownLatch latch) {
this.aggregator = aggregator;
this.message = message;
this.latch = latch;

10
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CompletionStrategyAdapterTests.java
View File

@@ -25,7 +25,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.aggregator.AggregatorAdapter;
import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.integration.aggregator.CompletionStrategy;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.message.GenericMessage;
@@ -130,7 +130,7 @@ public class CompletionStrategyAdapterTests {
@Test(expected = ConfigurationException.class)
public void testInvalidParameterTypeUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleCompletionStrategy, simpleCompletionStrategy.getClass().getMethod(
new MethodInvokingAggregator(simpleCompletionStrategy, simpleCompletionStrategy.getClass().getMethod(
"invalidParameterType", String.class));
}
@@ -160,19 +160,19 @@ public class CompletionStrategyAdapterTests {
@Test(expected = IllegalArgumentException.class)
public void testNullObject() {
new AggregatorAdapter(null, "doesNotMatter");
new MethodInvokingAggregator(null, "doesNotMatter");
}
@Test(expected = IllegalArgumentException.class)
public void testNullMethodName() {
String methodName = null;
new AggregatorAdapter(simpleCompletionStrategy, methodName);
new MethodInvokingAggregator(simpleCompletionStrategy, methodName);
}
@Test(expected = IllegalArgumentException.class)
public void testNullMethodObject() {
Method method = null;
new AggregatorAdapter(simpleCompletionStrategy, method);
new MethodInvokingAggregator(simpleCompletionStrategy, method);
}
private static List<Message<?>> createListOfMessages() {

38
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorAdapterTests.java → org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingAggregatorTests.java
View File

@@ -26,7 +26,7 @@ import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.aggregator.Aggregator;
import org.springframework.integration.aggregator.AggregatorAdapter;
import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
@@ -34,7 +34,7 @@ import org.springframework.integration.message.Message;
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class AggregatorAdapterTests {
public class MethodInvokingAggregatorTests {
private SimpleAggregator simpleAggregator;
@@ -46,7 +46,7 @@ public class AggregatorAdapterTests {
@Test
public void testAdapterWithNonParameterizedMessageListBasedMethod() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnNonParameterizedListOfMessages");
Aggregator aggregator = new MethodInvokingAggregator(simpleAggregator, "doAggregationOnNonParameterizedListOfMessages");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
@@ -55,7 +55,7 @@ public class AggregatorAdapterTests {
@Test
public void testAdapterWithWildcardParametrizedMessageBasedMethod() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfMessagesParametrizedWithWildcard");
Aggregator aggregator = new MethodInvokingAggregator(simpleAggregator, "doAggregationOnListOfMessagesParametrizedWithWildcard");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
@@ -64,7 +64,7 @@ public class AggregatorAdapterTests {
@Test
public void testAdapterWithTypeParametrizedMessageBasedMethod() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfMessagesParametrizedWithString");
Aggregator aggregator = new MethodInvokingAggregator(simpleAggregator, "doAggregationOnListOfMessagesParametrizedWithString");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
@@ -73,7 +73,7 @@ public class AggregatorAdapterTests {
@Test
public void testAdapterWithPojoBasedMethod() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfStrings");
Aggregator aggregator = new MethodInvokingAggregator(simpleAggregator, "doAggregationOnListOfStrings");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
@@ -82,7 +82,7 @@ public class AggregatorAdapterTests {
@Test
public void testAdapterWithPojoBasedMethodReturningObject() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "doAggregationOnListOfStringsReturningLong");
Aggregator aggregator = new MethodInvokingAggregator(simpleAggregator, "doAggregationOnListOfStringsReturningLong");
List<Message<?>> messages = createListOfMessages();
Message<?> returnedMessge = aggregator.aggregate(messages);
Assert.assertTrue(simpleAggregator.isAggregationPerformed());
@@ -91,68 +91,68 @@ public class AggregatorAdapterTests {
@Test(expected=ConfigurationException.class)
public void testAdapterWithWrongMethodName() {
new AggregatorAdapter(simpleAggregator, "methodThatDoesNotExist");
new MethodInvokingAggregator(simpleAggregator, "methodThatDoesNotExist");
}
@Test(expected=ConfigurationException.class)
public void testInvalidParameterTypeUsingMethodName() {
new AggregatorAdapter(simpleAggregator, "invalidParameterType");
new MethodInvokingAggregator(simpleAggregator, "invalidParameterType");
}
@Test(expected=ConfigurationException.class)
public void testTooManyParametersUsingMethodName() {
new AggregatorAdapter(simpleAggregator, "tooManyParameters");
new MethodInvokingAggregator(simpleAggregator, "tooManyParameters");
}
@Test(expected=ConfigurationException.class)
public void testNotEnoughParametersUsingMethodName() {
new AggregatorAdapter(simpleAggregator, "notEnoughParameters");
new MethodInvokingAggregator(simpleAggregator, "notEnoughParameters");
}
@Test(expected=ConfigurationException.class)
public void testListSubclassParameterUsingMethodName() {
new AggregatorAdapter(simpleAggregator, "ListSubclassParameter");
new MethodInvokingAggregator(simpleAggregator, "ListSubclassParameter");
}
@Test(expected=ConfigurationException.class)
public void testInvalidParameterTypeUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod(
new MethodInvokingAggregator(simpleAggregator, simpleAggregator.getClass().getMethod(
"invalidParameterType", String.class));
}
@Test(expected=ConfigurationException.class)
public void testTooManyParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod(
new MethodInvokingAggregator(simpleAggregator, simpleAggregator.getClass().getMethod(
"tooManyParameters", List.class, List.class));
}
@Test(expected=ConfigurationException.class)
public void testNotEnoughParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod(
new MethodInvokingAggregator(simpleAggregator, simpleAggregator.getClass().getMethod(
"notEnoughParameters", new Class[] {} ));
}
@Test(expected= ConfigurationException.class)
public void testListSubclassParameterUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod(
new MethodInvokingAggregator(simpleAggregator, simpleAggregator.getClass().getMethod(
"listSubclassParameter", new Class[] {LinkedList.class} ));
}
@Test(expected=IllegalArgumentException.class)
public void testNullObject() {
new AggregatorAdapter(null, "doesNotMatter");
new MethodInvokingAggregator(null, "doesNotMatter");
}
@Test(expected=IllegalArgumentException.class)
public void testNullMethodName() {
String methodName = null;
new AggregatorAdapter(simpleAggregator, methodName);
new MethodInvokingAggregator(simpleAggregator, methodName);
}
@Test(expected=IllegalArgumentException.class)
public void testNullMethodObject() {
Method method = null;
new AggregatorAdapter(simpleAggregator, method);
new MethodInvokingAggregator(simpleAggregator, method);
}

58
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
View File

@@ -28,7 +28,7 @@ import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.AggregatorEndpoint;
import org.springframework.integration.aggregator.CompletionStrategy;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.channel.MessageChannel;
@@ -53,15 +53,15 @@ public class AggregatorParserTests {
@Test
public void testAggregation() {
AggregatingMessageHandler aggregatingHandler =
(AggregatingMessageHandler) context.getBean("aggregatorWithReference");
AggregatorEndpoint endpoint =
(AggregatorEndpoint) context.getBean("aggregatorWithReference");
TestAggregator aggregatorBean = (TestAggregator) context.getBean("aggregatorBean");
List<Message<?>> outboundMessages = new ArrayList<Message<?>>();
outboundMessages.add(createMessage("123", "id1", 3, 1, null));
outboundMessages.add(createMessage("789", "id1", 3, 3, null));
outboundMessages.add(createMessage("456", "id1", 3, 2, null));
for (Message<?> message : outboundMessages) {
aggregatingHandler.send(message);
endpoint.send(message);
}
Assert.assertEquals("One and only one message must have been aggregated", 1, aggregatorBean
.getAggregatedMessages().size());
@@ -72,41 +72,41 @@ public class AggregatorParserTests {
@Test
public void testPropertyAssignment() throws Exception {
AggregatingMessageHandler completeAggregatingMessageHandler =
(AggregatingMessageHandler) context.getBean("completelyDefinedAggregator");
AggregatorEndpoint endpoint =
(AggregatorEndpoint) context.getBean("completelyDefinedAggregator");
TestAggregator testAggregator = (TestAggregator) context.getBean("aggregatorBean");
CompletionStrategy completionStrategy = (CompletionStrategy) context.getBean("completionStrategy");
MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel");
DirectFieldAccessor messageHandlerFieldAccessor = new DirectFieldAccessor(completeAggregatingMessageHandler);
Assert.assertEquals("The AggregatingMessageHandler is not injected with the appropriate Aggregator instance",
testAggregator, messageHandlerFieldAccessor.getPropertyValue("aggregator"));
DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint);
Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate Aggregator instance",
testAggregator, accessor.getPropertyValue("aggregator"));
Assert.assertEquals(
"The AggregatingMessageHandler is not injected with the appropriate CompletionStrategy instance",
completionStrategy, messageHandlerFieldAccessor.getPropertyValue("completionStrategy"));
Assert.assertEquals("The AggregatingMessageHandler is not injected with the appropriate output channel",
outputChannel, messageHandlerFieldAccessor.getPropertyValue("target"));
Assert.assertEquals("The AggregatingMessageHandler is not injected with the appropriate discard channel",
discardChannel, messageHandlerFieldAccessor.getPropertyValue("discardChannel"));
Assert.assertEquals("The AggregatingMessageHandler is not set with the appropriate timeout value", 86420000l,
messageHandlerFieldAccessor.getPropertyValue("sendTimeout"));
"The AggregatorEndpoint is not injected with the appropriate CompletionStrategy instance",
completionStrategy, accessor.getPropertyValue("completionStrategy"));
Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate output channel",
outputChannel, accessor.getPropertyValue("target"));
Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate discard channel",
discardChannel, accessor.getPropertyValue("discardChannel"));
Assert.assertEquals("The AggregatorEndpoint is not set with the appropriate timeout value",
86420000l, accessor.getPropertyValue("sendTimeout"));
Assert.assertEquals(
"The AggregatingMessageHandler is not configured with the appropriate 'send partial results on timeout' flag",
true, messageHandlerFieldAccessor.getPropertyValue("sendPartialResultOnTimeout"));
Assert.assertEquals("The AggregatingMessageHandler is not configured with the appropriate reaper interval",
135l, messageHandlerFieldAccessor.getPropertyValue("reaperInterval"));
"The AggregatorEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
true, accessor.getPropertyValue("sendPartialResultOnTimeout"));
Assert.assertEquals("The AggregatorEndpoint is not configured with the appropriate reaper interval",
135l, accessor.getPropertyValue("reaperInterval"));
Assert.assertEquals(
"The AggregatingMessageHandler is not configured with the appropriate tracked correlationId capacity",
99, messageHandlerFieldAccessor.getPropertyValue("trackedCorrelationIdCapacity"));
Assert.assertEquals("The AggregatingMessageHandler is not configured with the appropriate timeout",
42l, messageHandlerFieldAccessor.getPropertyValue("timeout"));
"The AggregatorEndpoint is not configured with the appropriate tracked correlationId capacity",
99, accessor.getPropertyValue("trackedCorrelationIdCapacity"));
Assert.assertEquals("The AggregatorEndpoint is not configured with the appropriate timeout",
42l, accessor.getPropertyValue("timeout"));
}
@Test
public void testSimpleJavaBeanAggregator() {
List<Message<?>> outboundMessages = new ArrayList<Message<?>>();
AggregatingMessageHandler addingAggregator =
(AggregatingMessageHandler) context.getBean("aggregatorWithReferenceAndMethod");
AggregatorEndpoint addingAggregator =
(AggregatorEndpoint) context.getBean("aggregatorWithReferenceAndMethod");
outboundMessages.add(createMessage(1l, "id1", 3, 1, null));
outboundMessages.add(createMessage(2l, "id1", 3, 3, null));
outboundMessages.add(createMessage(3l, "id1", 3, 2, null));
@@ -131,8 +131,8 @@ public class AggregatorParserTests {
@Test
public void testAggregatorWithPojoCompletionStrategy(){
AggregatingMessageHandler aggregatorWithPojoCompletionStrategy =
(AggregatingMessageHandler) context.getBean("aggregatorWithPojoCompletionStrategy");
AggregatorEndpoint aggregatorWithPojoCompletionStrategy =
(AggregatorEndpoint) context.getBean("aggregatorWithPojoCompletionStrategy");
CompletionStrategy completionStrategy = (CompletionStrategy)
new DirectFieldAccessor(aggregatorWithPojoCompletionStrategy).getPropertyValue("completionStrategy");
Assert.assertTrue(completionStrategy instanceof CompletionStrategyAdapter);

12
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

@@ -24,7 +24,7 @@ import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.AggregatorEndpoint;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.aggregator.SequenceSizeCompletionStrategy;
import org.springframework.integration.bus.MessageBus;
@@ -46,14 +46,14 @@ public class AggregatorAnnotationTests {
Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof SequenceSizeCompletionStrategy);
Assert.assertNull(aggregatingMessageHandlerAccessor.getPropertyValue("target"));
Assert.assertNull(aggregatingMessageHandlerAccessor.getPropertyValue("discardChannel"));
Assert.assertEquals(AggregatingMessageHandler.DEFAULT_SEND_TIMEOUT, aggregatingMessageHandlerAccessor
Assert.assertEquals(AggregatorEndpoint.DEFAULT_SEND_TIMEOUT, aggregatingMessageHandlerAccessor
.getPropertyValue("sendTimeout"));
Assert.assertEquals(AggregatingMessageHandler.DEFAULT_TIMEOUT, aggregatingMessageHandlerAccessor
Assert.assertEquals(AggregatorEndpoint.DEFAULT_TIMEOUT, aggregatingMessageHandlerAccessor
.getPropertyValue("timeout"));
Assert.assertEquals(false, aggregatingMessageHandlerAccessor.getPropertyValue("sendPartialResultOnTimeout"));
Assert.assertEquals(AggregatingMessageHandler.DEFAULT_REAPER_INTERVAL, aggregatingMessageHandlerAccessor
Assert.assertEquals(AggregatorEndpoint.DEFAULT_REAPER_INTERVAL, aggregatingMessageHandlerAccessor
.getPropertyValue("reaperInterval"));
Assert.assertEquals(AggregatingMessageHandler.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY,
Assert.assertEquals(AggregatorEndpoint.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY,
aggregatingMessageHandlerAccessor.getPropertyValue("trackedCorrelationIdCapacity"));
}
@@ -101,7 +101,7 @@ public class AggregatorAnnotationTests {
@SuppressWarnings("unchecked")
private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context, final String endpointName) {
MessageBus messageBus = this.getMessageBus(context);
AggregatingMessageHandler endpoint = (AggregatingMessageHandler) messageBus.lookupEndpoint(endpointName + ".aggregator");
AggregatorEndpoint endpoint = (AggregatorEndpoint) messageBus.lookupEndpoint(endpointName + ".aggregator");
return new DirectFieldAccessor(endpoint);
}