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 "); }