From 058e555de9bb7d3731085608ac007ce702d94f85 Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Tue, 29 Dec 2009 18:19:35 +0000 Subject: [PATCH] INT-330: Added MessageGroup and refactored CorrelatingMessageHandler and MessageGroupProcessors. This is an intermediary commit. Still there are some options for simplification around the MessageGroupListener interface, as onMessageProcessed could likely be moved into the message group. --- .../aggregator/CompletionStrategy.java | 2 +- .../aggregator/CompletionStrategyAdapter.java | 2 +- .../aggregator/CorrelatingMessageHandler.java | 92 +++++++-------- .../DefaultResequencerStrategies.java | 10 +- .../integration/aggregator/MessageGroup.java | 106 ++++++++++++++++++ ...allback.java => MessageGroupListener.java} | 5 +- .../aggregator/MessageGroupProcessor.java | 11 +- .../aggregator/MessageListMethodAdapter.java | 4 +- .../MethodInvokingMessageGroupProcessor.java | 12 +- .../PassThroughMessageGroupProcessor.java | 8 +- .../SequenceSizeCompletionStrategy.java | 2 +- ...elatingMessageHandlerIntegrationTest.java} | 12 +- .../CorrelatingMessageHandlerTests.java | 33 +++--- .../aggregator/MessageGroupTests.java | 33 ++++++ ...hodInvokingMessageGroupProcessorTests.java | 39 ++++--- .../NewAggregatorEndpointTests.java | 23 ++-- .../NewConcurrentAggregatorEndpointTests.java | 25 ++--- ...ggregatorWithCorrelationStrategyTests.java | 2 +- .../config/TestCompletionStrategy.java | 2 +- 19 files changed, 277 insertions(+), 146 deletions(-) create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java rename org.springframework.integration/src/main/java/org/springframework/integration/aggregator/{BufferedMessagesCallback.java => MessageGroupListener.java} (79%) rename org.springframework.integration/src/test/java/org/springframework/integration/aggregator/{BufferingMessageHandlerIntegrationTest.java => CorrelatingMessageHandlerIntegrationTest.java} (89%) create mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategy.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategy.java index e15db809d6..fcdccdebf4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategy.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategy.java @@ -29,6 +29,6 @@ import org.springframework.integration.core.Message; */ public interface CompletionStrategy { - boolean isComplete(List> messages); + boolean isComplete(List> messages); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategyAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategyAdapter.java index 3b47f8f7e0..043927337b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategyAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategyAdapter.java @@ -43,7 +43,7 @@ public class CompletionStrategyAdapter extends MessageListMethodAdapter implemen } - public boolean isComplete(List> messages) { + public boolean isComplete(List> messages) { return ((Boolean) executeMethod(messages)).booleanValue(); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 558955a732..0926f40b32 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -16,18 +16,6 @@ package org.springframework.integration.aggregator; -import java.util.List; -import java.util.Queue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.DelayQueue; -import java.util.concurrent.Delayed; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.ReentrantLock; - import org.springframework.context.Lifecycle; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.NullChannel; @@ -40,10 +28,15 @@ import org.springframework.integration.store.SimpleMessageStore; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.*; +import java.util.concurrent.locks.ReentrantLock; + /** - * MessageHandler that holds a buffer of messages in a MessageStore. This class takes care of + * MessageHandler that holds a buffer of messages in a MessageStore. This class takes care of * correlated groups of messages that can be completed in batches. It is useful for aggregating, - * resequencing, or custom implementations requiring correlation. + * resequencing, or custom implementations requiring correlation. * * @author Iwein Fuld */ @@ -51,24 +44,28 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements private MessageStore store = new SimpleMessageStore(100); private final CorrelationStrategy correlationStrategy; - private final IdTracker tracker = new IdTracker(); private final CompletionStrategy completionStrategy; private MessageGroupProcessor outputProcessor; + private MessageChannel outputChannel; private volatile MessageChannel discardChannel = new NullChannel(); - private TaskScheduler taskScheduler; - private Object lifecycleMonitor = new Object(); - private ScheduledFuture reaperFutureTask; - private volatile long reaperInterval = 1000l; - private final BlockingQueue keysInBuffer = new DelayQueue(); - private volatile long timeout = 60000l; - private volatile boolean sendPartialResultOnTimeout; private ChannelResolver channelResolver; + private final IdTracker tracker = new IdTracker(); + private final BlockingQueue keysInBuffer = new DelayQueue(); + + private volatile TaskScheduler taskScheduler; + private volatile ScheduledFuture reaperFutureTask; + private volatile long reaperInterval = 1000l; + private volatile long timeout = 60000l; + private volatile boolean sendPartialResultOnTimeout; + + private Object lifecycleMonitor = new Object(); + public CorrelatingMessageHandler(MessageStore store, - CorrelationStrategy correlationStrategy, - CompletionStrategy completionStrategy, - MessageGroupProcessor processor) { + CorrelationStrategy correlationStrategy, + CompletionStrategy completionStrategy, + MessageGroupProcessor processor) { Assert.notNull(store); Assert.notNull(correlationStrategy); Assert.notNull(completionStrategy); @@ -80,7 +77,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } public CorrelatingMessageHandler(MessageStore store, - MessageGroupProcessor processor) { + MessageGroupProcessor processor) { this(store, new HeaderAttributeCorrelationStrategy( MessageHeaders.CORRELATION_ID), new SequenceSizeCompletionStrategy(), processor); @@ -119,11 +116,16 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements Object correlationKey = correlationStrategy.getCorrelationKey(message); try { if (tracker.aquireLockFor(correlationKey)) { - List> group = store.list(correlationKey); - if (noSupersedingMessage(message, group)) { + MessageGroup group = + new MessageGroup(store, completionStrategy, correlationKey, deleteOrTrackCallback()); + if (group.hasNoMessageSuperseding(message)) { store(message, correlationKey); group.add(message); - complete(correlationKey, group, this.resolveReplyChannel(message, this.outputChannel, this.channelResolver)); + + if (group.isComplete()) { + outputProcessor.processAndSend(group, + this.resolveReplyChannel(message, this.outputChannel, this.channelResolver)); + } } else { discardChannel.send(message); } @@ -135,26 +137,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } } - private boolean noSupersedingMessage(Message message, List> group) { - for (Message member : group) { - if (member.getHeaders().getSequenceNumber() == message.getHeaders().getSequenceNumber()) { - return false; - } - } - return true; - } - - private boolean complete(Object correlationKey, List> correlatedMessages, MessageChannel messageChannel) { - boolean processed = false; - if (completionStrategy.isComplete(correlatedMessages)) { - outputProcessor.processAndSend(correlationKey, correlatedMessages, messageChannel, deleteOrTrackCallback()); - processed = true; - } - return processed; - } - - private BufferedMessagesCallback deleteOrTrackCallback() { - return new BufferedMessagesCallback() { + private MessageGroupListener deleteOrTrackCallback() { + return new MessageGroupListener() { public void onProcessingOf(Message... processedMessage) { for (Message message : processedMessage) { store.delete(message.getHeaders().getId()); @@ -221,17 +205,23 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } protected final void forceComplete(Object key) { + MessageGroup group = new MessageGroup(store, completionStrategy, key, deleteOrTrackCallback()); List> all = store.list(key); if (all.size() > 0) { //last chance for normal completion MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel, this.channelResolver); - boolean fullyCompleted = complete(key, all, outputChannel); + boolean processed = false; + if (completionStrategy.isComplete(all)) { + outputProcessor.processAndSend(group, outputChannel); + processed = true; + } + boolean fullyCompleted = processed; if (!fullyCompleted) { if (sendPartialResultOnTimeout) { if (logger.isInfoEnabled()) { logger.info("Processing partially complete messages for key [" + key + "] to: " + outputChannel); } - outputProcessor.processAndSend(key, all, outputChannel, deleteOrTrackCallback()); + outputProcessor.processAndSend(group, outputChannel); } else { if (logger.isInfoEnabled()) { logger.info("Discarding partially complete messages for key [" + key + "] to: " + discardChannel); 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 41f695ea23..d43182e2f3 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 @@ -38,12 +38,14 @@ public class DefaultResequencerStrategies implements CorrelationStrategy, Comple return key; } - public boolean isComplete(List> messages) { + public boolean isComplete(List> messages) { return releasePartialSequences|| messages.get(0).getHeaders().getSequenceSize()==messages.size(); } - public void processAndSend(Object correlationKey, Collection> all, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { + public void processAndSend(MessageGroup group, MessageChannel outputChannel) { + List> all = group.getMessages(); + Object correlationKey = group.getCorrelationKey(); if (all.size() > 0) { List sorted = new ArrayList(all); Collections.sort(sorted, sequenceSizeComparator); @@ -53,12 +55,12 @@ public class DefaultResequencerStrategies implements CorrelationStrategy, Comple if (sequenceNumber <= nextSequence.get()) { outputChannel.send(message); nextSequence.compareAndSet(sequenceNumber, sequenceNumber + 1); - processedCallback.onProcessingOf(message); + group.onProcessingOf(message); } } MessageHeaders headers = sorted.get(0).getHeaders(); if (all.size() == headers.getSequenceSize()){ - processedCallback.onCompletionOf(correlationKey); + group.onCompletion(); } } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java new file mode 100644 index 0000000000..a62e7c492a --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java @@ -0,0 +1,106 @@ +package org.springframework.integration.aggregator; + +import org.springframework.integration.core.Message; +import org.springframework.integration.store.MessageStore; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +/** + * Represents a group of correllated messages that is bound to a certain + * {@link org.springframework.integration.store.MessageStore}. The group can grow during its lifetime, if messages are + * added to it. According to its {@link org.springframework.integration.aggregator.CompletionStrategy} it + * can be complete depending on the messages in the group. + * + * @author Iwein Fuld + */ +public class MessageGroup { + private final MessageStore store; + private final CompletionStrategy completionStrategy; + private final Object correlationKey; + private final ArrayList> messages = new ArrayList>(); + private final List listeners; + + + public MessageGroup(MessageStore store, CompletionStrategy completionStrategy, Object correlationKey, MessageGroupListener... listeners) { + this.store = store; + this.completionStrategy = completionStrategy; + this.correlationKey = correlationKey; + this.messages.addAll(store.list(correlationKey)); + this.listeners = Collections.unmodifiableList(Arrays.asList(listeners)); + } + + public boolean hasNoMessageSuperseding(Message message) { + for (Message member : messages) { + if (member.getHeaders().getSequenceNumber() == message.getHeaders().getSequenceNumber()) { + return false; + } + } + return true; + } + + public void add(Message message) { + messages.add(message); + } + + public boolean isComplete() { + return completionStrategy.isComplete(messages); + } + + public static MessageGroupBuilder builder() { + return new MessageGroupBuilder(); + } + + public List> getMessages() { + return messages; + } + + public Object getCorrelationKey() { + return correlationKey; + } + + public void onProcessingOf(Message... messages) { + for (MessageGroupListener listener : listeners) { + listener.onProcessingOf(messages); + } + } + + public void onCompletion() { + for (MessageGroupListener listener : listeners) { + listener.onCompletionOf(correlationKey); + } + } + + final static class MessageGroupBuilder { + private MessageStore store; + private Object correlationKey; + private MessageGroupListener[] listeners; + private CompletionStrategy completionStrategy; + + public MessageGroupBuilder withCorrelationKey(Object key) { + this.correlationKey = key; + return this; + } + + public MessageGroupBuilder withStore(MessageStore store) { + this.store = store; + return this; + } + + public MessageGroupBuilder observedBy(MessageGroupListener... listeners) { + this.listeners = listeners; + return this; + } + + public MessageGroupBuilder completedBy(CompletionStrategy completionStrategy) { + this.completionStrategy = completionStrategy; + return this; + } + + public MessageGroup build() { + return new MessageGroup(store, completionStrategy, correlationKey, listeners); + } + } +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferedMessagesCallback.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupListener.java similarity index 79% rename from org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferedMessagesCallback.java rename to org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupListener.java index 84d83b7047..dd58adeec9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferedMessagesCallback.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupListener.java @@ -18,9 +18,12 @@ package org.springframework.integration.aggregator; import org.springframework.integration.core.Message; /** + * Listener that can be configured with a MessageGroup to receive notifications on processing of (parts of) the group + * and exactly one notification when the whole group completes. + * * @author Iwein Fuld */ -public interface BufferedMessagesCallback { +public interface MessageGroupListener { void onProcessingOf(Message... processedMessages); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java index 89d2bfd784..e1403b28d3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java @@ -1,18 +1,13 @@ package org.springframework.integration.aggregator; -import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.aggregator.BufferedMessagesCallback; - -import java.util.Collection; /** * @author Iwein Fuld */ public interface MessageGroupProcessor { - void processAndSend(Object correlationKey, - Collection> messagesUpForProcessing, - MessageChannel outputChannel, - BufferedMessagesCallback processedCallback); + void processAndSend(MessageGroup group, + MessageChannel outputChannel + ); } \ No newline at end of file diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java index 43118c75e8..a23e788818 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java @@ -72,7 +72,7 @@ public class MessageListMethodAdapter { && Message.class.isAssignableFrom((Class) ((ParameterizedType) getCollectionActualType(method)).getRawType()); } - protected final Object executeMethod(Collection> messages) { + protected final Object executeMethod(Collection> messages) { try { if (isMethodParameterParameterized(this.method) && isHavingActualTypeArguments(this.method) && (isActualTypeRawMessage(this.method) || isActualTypeParameterizedMessage(this.method))) { @@ -89,7 +89,7 @@ public class MessageListMethodAdapter { } } - private List extractPayloadsFromMessages(Collection> messages) { + private List extractPayloadsFromMessages(Collection> messages) { List payloadList = new ArrayList(); for (Message message : messages) { payloadList.add(message.getPayload()); 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 d9fdcb4454..2da9bd3f49 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 @@ -115,14 +115,14 @@ public class MethodInvokingMessageGroupProcessor implements MessageGroupProcesso }); } - public void processAndSend(Object correlationKey, - Collection> messagesUpForProcessing, - MessageChannel outputChannel, - BufferedMessagesCallback processedCallback) { + public void processAndSend(MessageGroup group, + MessageChannel outputChannel + ) { + final Collection> messagesUpForProcessing = group.getMessages(); Message reply = MessageBuilder.withPayload( this.adapter.executeMethod(messagesUpForProcessing)).build(); - processedCallback.onCompletionOf(correlationKey); - processedCallback.onProcessingOf(messagesUpForProcessing + group.onCompletion(); + group.onProcessingOf(messagesUpForProcessing .toArray(new Message[]{})); outputChannel.send(reply); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java index 20caebecca..564ace3168 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java @@ -3,13 +3,13 @@ package org.springframework.integration.aggregator; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; -import java.util.Collection; - public class PassThroughMessageGroupProcessor implements MessageGroupProcessor { - public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { - for (Message message : messagesUpForProcessing) { + public void processAndSend(MessageGroup group, MessageChannel outputChannel) { + for (Message message : group.getMessages()) { outputChannel.send(message); + group.onProcessingOf(message); } + group.onCompletion(); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeCompletionStrategy.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeCompletionStrategy.java index 2f71034a2e..cbeed4ddc2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeCompletionStrategy.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeCompletionStrategy.java @@ -31,7 +31,7 @@ import org.springframework.util.CollectionUtils; */ public class SequenceSizeCompletionStrategy implements CompletionStrategy { - public boolean isComplete(List> messages) { + public boolean isComplete(List> messages) { if (CollectionUtils.isEmpty(messages)) { return false; } 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/CorrelatingMessageHandlerIntegrationTest.java similarity index 89% rename from org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java rename to org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java index 029a902c7f..cbd3ea9cd2 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/CorrelatingMessageHandlerIntegrationTest.java @@ -1,24 +1,22 @@ package org.springframework.integration.aggregator; -import org.junit.Test; import org.junit.Before; -import static org.mockito.Mockito.*; +import org.junit.Test; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; -public class BufferingMessageHandlerIntegrationTest { +import static org.mockito.Mockito.*; + +public class CorrelatingMessageHandlerIntegrationTest { private CompletionStrategy completionStrategy; private CorrelationStrategy correlationStrategy; private MessageStore store = new SimpleMessageStore(100); private MessageChannel outputChannel = mock(MessageChannel.class); private MessageGroupProcessor processor = new PassThroughMessageGroupProcessor(); -// private CorrelatingMessageHandler customizedHandler = new CorrelatingMessageHandler( -// store, correlationStrategy, completionStrategy, processor, -// outputChannel); private CorrelatingMessageHandler defaultHandler = new CorrelatingMessageHandler( store, processor); @@ -53,7 +51,7 @@ public class BufferingMessageHandlerIntegrationTest { } @Test - public void completesWithoutReleasingIncompleteCorrellations() throws Exception { + public void completesWithoutReleasingIncompleteCorrelations() throws Exception { Message message1 = correlatedMessage(1, 2, 1); Message message2 = correlatedMessage(2, 2, 1); Message message1a = correlatedMessage(1, 2, 2); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index ebc5ba67f7..80f97cb493 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -16,27 +16,25 @@ package org.springframework.integration.aggregator; -import static org.mockito.Matchers.eq; -import static org.mockito.Matchers.isA; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -import java.util.ArrayList; -import java.util.List; - import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; - 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.store.MessageStore; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.mockito.Matchers.eq; +import static org.mockito.Matchers.isA; +import static org.mockito.Mockito.*; + /** * @author Iwein Fuld */ @@ -79,12 +77,12 @@ public class CorrelatingMessageHandlerTests { when(correlationStrategy.getCorrelationKey(isA(Message.class))) .thenReturn(correlationKey); - when(completionStrategy.isComplete(storedMessages)).thenReturn(false); + when(completionStrategy.isComplete(Arrays.asList(message1))).thenReturn(false); handler.handleMessage(message1); storedMessages.add(message1); - when(completionStrategy.isComplete(storedMessages)).thenReturn(true); + when(completionStrategy.isComplete(Arrays.asList(message1, message2))).thenReturn(true); handler.handleMessage(message2); storedMessages.add(message2); @@ -93,15 +91,16 @@ public class CorrelatingMessageHandlerTests { verify(store, times(2)).list(correlationKey); verify(correlationStrategy).getCorrelationKey(message1); verify(correlationStrategy).getCorrelationKey(message2); - verify(completionStrategy, times(2)).isComplete(storedMessages); - verify(processor).processAndSend(eq(correlationKey), - eq(storedMessages), eq(outputChannel), - isA(BufferedMessagesCallback.class)); + verify(completionStrategy).isComplete(Arrays.asList(message1)); + verify(completionStrategy).isComplete(Arrays.asList(message1, message2)); + verify(processor).processAndSend(isA(MessageGroup.class), + eq(outputChannel) + ); } private Message testMessage(int id, int sequenceNumber) { - return MessageBuilder.withPayload("test") + return MessageBuilder.withPayload("test"+id) .setHeader(MessageHeaders.ID, id) .setSequenceNumber(sequenceNumber).build(); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java new file mode 100644 index 0000000000..8b1f10f259 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java @@ -0,0 +1,33 @@ +package org.springframework.integration.aggregator; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import org.springframework.integration.store.MessageStore; + +/** + * + */ +@RunWith(MockitoJUnitRunner.class) +public class MessageGroupTests { + + @Mock + private MessageStore store; + + private Object key = new Object(); + + @Mock + private MessageGroupListener listener; + + @Mock + private CompletionStrategy completionStrategy; + + @Test + public void shouldBuildMessageGroup() { + MessageGroup group = MessageGroup.builder(). + withStore(store).withCorrelationKey(key). + completedBy(completionStrategy).observedBy(listener). + build(); + } +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index 5d9a8521dc..48f0065eb4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -13,7 +13,6 @@ import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; import java.util.ArrayList; -import java.util.Collection; import java.util.List; import static org.hamcrest.CoreMatchers.is; @@ -27,13 +26,15 @@ import static org.mockito.Mockito.when; public class MethodInvokingMessageGroupProcessorTests { @Mock - private BufferedMessagesCallback processedCallback; + private MessageGroupListener processedCallback; @Mock private MessageChannel outputChannel; - private Collection> messagesUpForProcessing = new ArrayList>( + private List> messagesUpForProcessing = new ArrayList>( 3); + @Mock + private MessageGroup messageGroupMock; @Before public void initializeMessagesUpForProcessing() { @@ -66,8 +67,9 @@ public class MethodInvokingMessageGroupProcessorTests { ArgumentCaptor messageCaptor = ArgumentCaptor .forClass(Message.class); when(outputChannel.send(isA(Message.class))).thenReturn(true); - processor.processAndSend(3, messagesUpForProcessing, outputChannel, - processedCallback); + when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); + processor.processAndSend(messageGroupMock, outputChannel + ); // verify verify(outputChannel).send(messageCaptor.capture()); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); @@ -91,8 +93,9 @@ public class MethodInvokingMessageGroupProcessorTests { ArgumentCaptor messageCaptor = ArgumentCaptor .forClass(Message.class); when(outputChannel.send(isA(Message.class))).thenReturn(true); - processor.processAndSend(3, messagesUpForProcessing, outputChannel, - processedCallback); + when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); + processor.processAndSend(messageGroupMock, outputChannel + ); // verify verify(outputChannel).send(messageCaptor.capture()); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); @@ -108,10 +111,11 @@ public class MethodInvokingMessageGroupProcessorTests { return result; } - public void voidMethodShouldBeIgnored(List flags){ + public void voidMethodShouldBeIgnored(List flags) { fail("this method should not be invoked"); } - public String methodAcceptingNoCollectionShouldBeIgnored(@Header String irrelevant){ + + public String methodAcceptingNoCollectionShouldBeIgnored(@Header String irrelevant) { fail("this method should not be invoked"); return null; } @@ -124,15 +128,17 @@ public class MethodInvokingMessageGroupProcessorTests { ); ArgumentCaptor messageCaptor = ArgumentCaptor - .forClass(Message.class) ; + .forClass(Message.class); when(outputChannel.send(isA(Message.class))).thenReturn(true); - processor.processAndSend(3, messagesUpForProcessing, outputChannel, - processedCallback); + when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); + processor.processAndSend(messageGroupMock, outputChannel + ); // verify verify(outputChannel).send(messageCaptor.capture()); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); } + private class AnnotatedParametersAggregator { public Integer and(List flags) { int result = 0; @@ -142,7 +148,7 @@ public class MethodInvokingMessageGroupProcessorTests { return result; } - public String listHeaderShouldBeIgnored(@Header List flags){ + public String listHeaderShouldBeIgnored(@Header List flags) { fail("this method should not be invoked"); return ""; } @@ -155,11 +161,12 @@ public class MethodInvokingMessageGroupProcessorTests { ); ArgumentCaptor messageCaptor = ArgumentCaptor - .forClass(Message.class) ; + .forClass(Message.class); when(outputChannel.send(isA(Message.class))).thenReturn(true); - processor.processAndSend(3, messagesUpForProcessing, outputChannel, - processedCallback); + when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); + processor.processAndSend(messageGroupMock, outputChannel + ); // verify verify(outputChannel).send(messageCaptor.capture()); assertThat((Integer) messageCaptor.getValue().getPayload(), is(7)); 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 c8e2e5ae39..5d2d84f320 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,26 +16,24 @@ package org.springframework.integration.aggregator; -import static org.junit.Assert.*; -import static org.hamcrest.CoreMatchers.*; import org.junit.Before; -import org.junit.Test; import org.junit.Ignore; +import org.junit.Test; 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.MessageHandlingException; -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 java.util.Collection; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.*; + /** * @author Mark Fisher * @author Marius Bogoevici @@ -252,24 +250,25 @@ public class NewAggregatorEndpointTests { } private class MultiplyingProcessor implements MessageGroupProcessor { - public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, - MessageChannel outputChannel, BufferedMessagesCallback processedCallback + public void processAndSend(MessageGroup group, + MessageChannel outputChannel ) { Integer product = 1; + List> messagesUpForProcessing = group.getMessages(); for (Message message : messagesUpForProcessing) { product *= (Integer) message.getPayload(); } outputChannel.send(MessageBuilder.withPayload(product).build()); - processedCallback.onProcessingOf( + group.onProcessingOf( messagesUpForProcessing.toArray(new Message[messagesUpForProcessing.size()]) ); - processedCallback.onCompletionOf(correlationKey); + group.onCompletion(); } } private class NullReturningMessageProcessor implements MessageGroupProcessor { - public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { + public void processAndSend(MessageGroup group, MessageChannel outputChannel) { //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 89dc7af2df..8051368f6e 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 @@ -17,11 +17,9 @@ 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.junit.Test; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.channel.QueueChannel; @@ -31,16 +29,16 @@ 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.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 java.util.Collection; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.*; + /** * @author Mark Fisher * @author Marius Bogoevici @@ -319,25 +317,26 @@ public class NewConcurrentAggregatorEndpointTests { if (this.aggregator != null) this.aggregator.stop(); } - private class MultiplyingProcessor implements MessageGroupProcessor { - public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, - MessageChannel outputChannel, BufferedMessagesCallback processedCallback + private class MultiplyingProcessor implements MessageGroupProcessor { + public void processAndSend(MessageGroup group, + MessageChannel outputChannel ) { Integer product = 1; + List> messagesUpForProcessing = group.getMessages(); for (Message message : messagesUpForProcessing) { product *= (Integer) message.getPayload(); } outputChannel.send(MessageBuilder.withPayload(product).build()); - processedCallback.onProcessingOf( + group.onProcessingOf( messagesUpForProcessing.toArray(new Message[messagesUpForProcessing.size()]) ); - processedCallback.onCompletionOf(correlationKey); + group.onCompletion(); } } private class NullReturningMessageProcessor implements MessageGroupProcessor { - public void processAndSend(Object correlationKey, Collection> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { + public void processAndSend(MessageGroup group, MessageChannel outputChannel) { //noop } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java index 7ae6a9c946..55902c8840 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java @@ -108,7 +108,7 @@ public class AggregatorWithCorrelationStrategyTests { this.expectedSize = expectedSize; } - public boolean isComplete(List> messages) { + public boolean isComplete(List> messages) { return messages.size() == expectedSize; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestCompletionStrategy.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestCompletionStrategy.java index 6384448e1c..e7b042d006 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestCompletionStrategy.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestCompletionStrategy.java @@ -26,7 +26,7 @@ import org.springframework.integration.core.Message; */ public class TestCompletionStrategy implements CompletionStrategy { - public boolean isComplete(List> messages) { + public boolean isComplete(List> messages) { throw new UnsupportedOperationException("This is not intended to be implemented, but to verify injection into an "); }