diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java similarity index 96% rename from org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java rename to org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 33cbb9e9f5..1378b1b4c2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -39,13 +39,13 @@ import java.util.concurrent.locks.ReentrantLock; * * @author Iwein Fuld */ -public class BufferingMessageHandler extends AbstractMessageHandler implements Lifecycle { +public class CorrelatingMessageHandler extends AbstractMessageHandler implements Lifecycle { private MessageStore store = new SimpleMessageStore(100); private final CorrelationStrategy correlationStrategy; private final IdTracker tracker = new IdTracker(); private final CompletionStrategy completionStrategy; - private MessagesProcessor outputProcessor; + private MessageGroupProcessor outputProcessor; private MessageChannel outputChannel; private volatile MessageChannel discardChannel = new NullChannel(); private TaskScheduler taskScheduler; @@ -57,10 +57,10 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L private volatile boolean sendPartialResultOnTimeout; private ChannelResolver channelResolver; - public BufferingMessageHandler(MessageStore store, + public CorrelatingMessageHandler(MessageStore store, CorrelationStrategy correlationStrategy, CompletionStrategy completionStrategy, - MessagesProcessor processor) { + MessageGroupProcessor processor) { Assert.notNull(store); Assert.notNull(correlationStrategy); Assert.notNull(completionStrategy); @@ -71,8 +71,8 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L this.outputProcessor = processor; } - public BufferingMessageHandler(MessageStore store, - MessagesProcessor processor) { + public CorrelatingMessageHandler(MessageStore store, + MessageGroupProcessor processor) { this(store, new HeaderAttributeCorrelationStrategy( MessageHeaders.CORRELATION_ID), new SequenceSizeCompletionStrategy(), processor); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java index 8aa80e5c9d..41f695ea23 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java @@ -27,7 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger; /** * @author Iwein Fuld */ -public class DefaultResequencerStrategies implements CorrelationStrategy, CompletionStrategy, MessagesProcessor { +public class DefaultResequencerStrategies implements CorrelationStrategy, CompletionStrategy, MessageGroupProcessor { private final ConcurrentMap nextMessagesToPass = new ConcurrentHashMap(); private volatile DefaultResequencerStrategies.SequenceNumberComparator sequenceSizeComparator = new SequenceNumberComparator(); private volatile boolean releasePartialSequences; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessagesProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java similarity index 93% rename from org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessagesProcessor.java rename to org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java index 90a49a9b98..89d2bfd784 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessagesProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java @@ -9,7 +9,7 @@ import java.util.Collection; /** * @author Iwein Fuld */ -public interface MessagesProcessor { +public interface MessageGroupProcessor { void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregatorFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregatorFactoryBean.java index 126c4f0d11..945b2675c2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregatorFactoryBean.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregatorFactoryBean.java @@ -3,12 +3,11 @@ package org.springframework.integration.aggregator; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.core.MessageHeaders; -import org.springframework.integration.handler.MethodInvokingMessageProcessor; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; public class MethodInvokingAggregatorFactoryBean implements - FactoryBean , InitializingBean{ + FactoryBean , InitializingBean{ private static final int DEFAULT_CAPACITY = Integer.MAX_VALUE; @@ -19,7 +18,7 @@ public class MethodInvokingAggregatorFactoryBean implements private CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); - private MessagesProcessor processor; + private MessageGroupProcessor processor; private Object target; @@ -33,13 +32,13 @@ public class MethodInvokingAggregatorFactoryBean implements // build processor } - public BufferingMessageHandler getObject() throws Exception { - return new BufferingMessageHandler(store, correlationStrategy, + public CorrelatingMessageHandler getObject() throws Exception { + return new CorrelatingMessageHandler(store, correlationStrategy, completionStrategy, processor); } - public Class getObjectType() { - return BufferingMessageHandler.class; + public Class getObjectType() { + return CorrelatingMessageHandler.class; } public boolean isSingleton() { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java similarity index 97% rename from org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessor.java rename to org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java index 299681cd96..d9fdcb4454 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java @@ -11,14 +11,14 @@ import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.*; -public class MethodInvokingMessagesProcessor implements MessagesProcessor { +public class MethodInvokingMessageGroupProcessor implements MessageGroupProcessor { private final Object target; private final Method method; private final MessageListMethodAdapter adapter; - public MethodInvokingMessagesProcessor(Object target) { + public MethodInvokingMessageGroupProcessor(Object target) { this.target = target; this.method = selectMethodFrom(target); this.adapter = new MessageListMethodAdapter(target, method); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessagesProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java similarity index 85% rename from org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessagesProcessor.java rename to org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java index 244764e941..20caebecca 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessagesProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java @@ -5,7 +5,7 @@ import org.springframework.integration.core.MessageChannel; import java.util.Collection; -public class PassThroughMessagesProcessor implements MessagesProcessor { +public class PassThroughMessageGroupProcessor implements MessageGroupProcessor { public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { for (Message message : messagesUpForProcessing) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java index 932cd998c5..029a902c7f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java @@ -15,11 +15,11 @@ public class BufferingMessageHandlerIntegrationTest { private CorrelationStrategy correlationStrategy; private MessageStore store = new SimpleMessageStore(100); private MessageChannel outputChannel = mock(MessageChannel.class); - private MessagesProcessor processor = new PassThroughMessagesProcessor(); -// private BufferingMessageHandler customizedHandler = new BufferingMessageHandler( + private MessageGroupProcessor processor = new PassThroughMessageGroupProcessor(); +// private CorrelatingMessageHandler customizedHandler = new CorrelatingMessageHandler( // store, correlationStrategy, completionStrategy, processor, // outputChannel); - private BufferingMessageHandler defaultHandler = new BufferingMessageHandler( + private CorrelatingMessageHandler defaultHandler = new CorrelatingMessageHandler( store, processor); @Before public void setupHandler(){ diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java index 272eaf969b..13726d995a 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java @@ -37,7 +37,7 @@ import static org.mockito.Mockito.*; @RunWith(MockitoJUnitRunner.class) public class BufferingMessageHandlerTest { - private BufferingMessageHandler buffer; + private CorrelatingMessageHandler buffer; @Mock private MessageStore store; @Mock @@ -45,13 +45,13 @@ public class BufferingMessageHandlerTest { @Mock private CompletionStrategy completionStrategy; @Mock - private MessagesProcessor processor; + private MessageGroupProcessor processor; @Mock private MessageChannel outputChannel; @Before public void initializeSubject() { - buffer = new BufferingMessageHandler(store, correlationStrategy, + buffer = new CorrelatingMessageHandler(store, correlationStrategy, completionStrategy, processor); buffer.setOutputChannel(outputChannel); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java similarity index 93% rename from org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessorTests.java rename to org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index c0ee627cdd..5d9a8521dc 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -24,7 +24,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @RunWith(MockitoJUnitRunner.class) -public class MethodInvokingMessagesProcessorTests { +public class MethodInvokingMessageGroupProcessorTests { @Mock private BufferedMessagesCallback processedCallback; @@ -61,7 +61,7 @@ public class MethodInvokingMessagesProcessorTests { @Test public void shouldFindAnnotatedAggregatorMethod() throws Exception { - MessagesProcessor processor = new MethodInvokingMessagesProcessor( + MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor( new AnnotatedAggregatorMethod()); ArgumentCaptor messageCaptor = ArgumentCaptor .forClass(Message.class); @@ -86,7 +86,7 @@ public class MethodInvokingMessagesProcessorTests { @Test public void shouldFindSimpleAggregatorMethod() throws Exception { - MessagesProcessor processor = new MethodInvokingMessagesProcessor( + MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor( new SimpleAggregator()); ArgumentCaptor messageCaptor = ArgumentCaptor .forClass(Message.class); @@ -119,7 +119,7 @@ public class MethodInvokingMessagesProcessorTests { @Test public void shouldFindFittingMethodAmongMultipleUnanotated() { - MessagesProcessor processor = new MethodInvokingMessagesProcessor( + MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor( new UnnanotatedAggregator() ); @@ -150,7 +150,7 @@ public class MethodInvokingMessagesProcessorTests { @Test public void shouldFindFittingMethodAmongMultipleWithAnnotatedParameters() { - MessagesProcessor processor = new MethodInvokingMessagesProcessor( + MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor( new AnnotatedParametersAggregator() ); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java index e366f51114..c8e2e5ae39 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java @@ -16,36 +16,25 @@ package org.springframework.integration.aggregator; -import org.junit.After; import static org.junit.Assert.*; import static org.hamcrest.CoreMatchers.*; import org.junit.Before; import org.junit.Test; import org.junit.Ignore; -import org.springframework.core.task.SimpleAsyncTaskExecutor; -import org.springframework.core.task.TaskExecutor; import org.springframework.integration.channel.QueueChannel; 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.integration.message.MessageHandler; import org.springframework.integration.message.MessageHandlingException; -import org.springframework.integration.aggregator.BufferingMessageHandler; -import org.springframework.integration.aggregator.MessagesProcessor; +import org.springframework.integration.aggregator.CorrelatingMessageHandler; +import org.springframework.integration.aggregator.MessageGroupProcessor; import org.springframework.integration.aggregator.BufferedMessagesCallback; import org.springframework.integration.store.SimpleMessageStore; -import org.springframework.integration.config.StubTaskScheduler; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.Trigger; -import org.springframework.test.annotation.Repeat; import java.util.Collection; -import java.util.Date; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import java.util.concurrent.ScheduledFuture; /** * @author Mark Fisher @@ -54,11 +43,11 @@ import java.util.concurrent.ScheduledFuture; */ public class NewAggregatorEndpointTests { - private BufferingMessageHandler aggregator; + private CorrelatingMessageHandler aggregator; @Before public void configureAggregator() { - this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor()); + this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor()); } @Test @@ -236,7 +225,7 @@ public class NewAggregatorEndpointTests { @Test public void testNullReturningAggregator() throws InterruptedException { - this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor()); + this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor()); QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); @@ -262,7 +251,7 @@ public class NewAggregatorEndpointTests { return builder.build(); } - private class MultiplyingProcessor implements MessagesProcessor { + private class MultiplyingProcessor implements MessageGroupProcessor { public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback ) { @@ -279,7 +268,7 @@ public class NewAggregatorEndpointTests { } } - private class NullReturningMessageProcessor implements MessagesProcessor { + private class NullReturningMessageProcessor implements MessageGroupProcessor { public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { //noop } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java index c0f871e753..89dc7af2df 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java @@ -31,12 +31,11 @@ import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageHandlingException; -import org.springframework.integration.aggregator.BufferingMessageHandler; -import org.springframework.integration.aggregator.MessagesProcessor; +import org.springframework.integration.aggregator.CorrelatingMessageHandler; +import org.springframework.integration.aggregator.MessageGroupProcessor; import org.springframework.integration.aggregator.BufferedMessagesCallback; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; -import org.springframework.test.annotation.Repeat; import java.util.Collection; import java.util.concurrent.CountDownLatch; @@ -53,7 +52,7 @@ public class NewConcurrentAggregatorEndpointTests { private ThreadPoolTaskScheduler taskScheduler; - private BufferingMessageHandler aggregator; + private CorrelatingMessageHandler aggregator; @Before public void configureAggregator() { @@ -61,7 +60,7 @@ public class NewConcurrentAggregatorEndpointTests { this.taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.afterPropertiesSet(); this.taskScheduler.afterPropertiesSet(); - this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor()); + this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor()); this.aggregator.setTaskScheduler(this.taskScheduler); } @@ -243,7 +242,7 @@ public class NewConcurrentAggregatorEndpointTests { @Test public void testNullReturningAggregator() throws InterruptedException { this.aggregator.start(); - this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor()); + this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor()); this.aggregator.setTaskScheduler(this.taskScheduler); QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); @@ -320,7 +319,7 @@ public class NewConcurrentAggregatorEndpointTests { if (this.aggregator != null) this.aggregator.stop(); } - private class MultiplyingProcessor implements MessagesProcessor { + private class MultiplyingProcessor implements MessageGroupProcessor { public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback ) { @@ -337,7 +336,7 @@ public class NewConcurrentAggregatorEndpointTests { } } - private class NullReturningMessageProcessor implements MessagesProcessor { + private class NullReturningMessageProcessor implements MessageGroupProcessor { public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { //noop } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java index 8c612d21ea..99c29b7666 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java @@ -22,7 +22,7 @@ import static org.hamcrest.CoreMatchers.*; import org.junit.Before; import org.junit.Test; import org.junit.Ignore; -import org.junit.matchers.JUnitMatchers; + import static org.junit.matchers.JUnitMatchers.*; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; @@ -33,7 +33,6 @@ import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.test.util.TestUtils; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; -import java.util.Arrays; import static java.util.Arrays.*; /** @@ -43,7 +42,7 @@ import static java.util.Arrays.*; */ public class NewResequencerTests { - private BufferingMessageHandler resequencer; + private CorrelatingMessageHandler resequencer; private ThreadPoolTaskScheduler taskScheduler; @@ -53,7 +52,7 @@ public class NewResequencerTests { public void configureResequencer() { this.resequencerStrategies = new DefaultResequencerStrategies(); MessageStore store = new SimpleMessageStore(30); - this.resequencer = new BufferingMessageHandler(store, resequencerStrategies, resequencerStrategies, resequencerStrategies); + this.resequencer = new CorrelatingMessageHandler(store, resequencerStrategies, resequencerStrategies, resequencerStrategies); this.taskScheduler = TestUtils.createTaskScheduler(10); this.resequencer.setTaskScheduler(taskScheduler); this.taskScheduler.afterPropertiesSet();