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.

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<Message<?>> messages);
+	boolean isComplete(List<? extends Message<?>> messages);
 
 }

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<Message<?>> messages) {
+	public boolean isComplete(List<? extends Message<?>> messages) {
 		return ((Boolean) executeMethod(messages)).booleanValue();
 	}
 

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<DelayedKey> keysInBuffer = new DelayQueue<DelayedKey>();
-    private volatile long timeout = 60000l;
-    private volatile boolean sendPartialResultOnTimeout;
     private ChannelResolver channelResolver;
 
+    private final IdTracker tracker = new IdTracker();
+    private final BlockingQueue<DelayedKey> keysInBuffer = new DelayQueue<DelayedKey>();
+
+    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<Message<?>> 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<Message<?>> group) {
-        for (Message<?> member : group) {
-            if (member.getHeaders().getSequenceNumber() == message.getHeaders().getSequenceNumber()) {
-                return false;
-            }
-        }
-        return true;
-    }
-
-    private boolean complete(Object correlationKey, List<Message<?>> 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<Message<?>> 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);

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<Message<?>> messages) {
+    public boolean isComplete(List<? extends Message<?>> messages) {
         return releasePartialSequences||
                 messages.get(0).getHeaders().getSequenceSize()==messages.size();
     }
 
-    public void processAndSend(Object correlationKey, Collection<Message<?>> all, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
+    public void processAndSend(MessageGroup group, MessageChannel outputChannel) {
+        List<Message<?>> all =  group.getMessages();
+        Object correlationKey = group.getCorrelationKey();
         if (all.size() > 0) {
             List<Message> 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();
             }
         }
     }

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
+ * <code>add</code>ed 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<Message<?>> messages = new ArrayList<Message<?>>();
+    private final List<MessageGroupListener> 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<Message<?>> 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);
+        }
+    }
+}

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

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<Message<?>> messagesUpForProcessing,
-                        MessageChannel outputChannel,
-                        BufferedMessagesCallback processedCallback);
+    void processAndSend(MessageGroup group,
+                        MessageChannel outputChannel
+    );
 }

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<Message<?>> messages) {
+    protected final Object executeMethod(Collection<? extends Message<?>> 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<Message<?>> messages) {
+    private List<?> extractPayloadsFromMessages(Collection<? extends Message<?>> messages) {
         List<Object> payloadList = new ArrayList<Object>();
         for (Message<?> message : messages) {
             payloadList.add(message.getPayload());

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<Message<?>> messagesUpForProcessing,
-                               MessageChannel outputChannel,
-                               BufferedMessagesCallback processedCallback) {
+    public void processAndSend(MessageGroup group,
+                               MessageChannel outputChannel
+    ) {
+        final Collection<Message<?>> 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);
     }

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<Message<?>> 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();
     }
 }

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<Message<?>> messages) {
+	public boolean isComplete(List<? extends Message<?>> messages) {
 		if (CollectionUtils.isEmpty(messages)) {
 			return false;
 		}

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

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

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();
+    }
+}

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<Message<?>> messagesUpForProcessing = new ArrayList<Message<?>>(
+    private List<Message<?>> messagesUpForProcessing = new ArrayList<Message<?>>(
             3);
+    @Mock
+    private MessageGroup messageGroupMock;
 
     @Before
     public void initializeMessagesUpForProcessing() {
@@ -66,8 +67,9 @@ public class MethodInvokingMessageGroupProcessorTests {
         ArgumentCaptor<Message> 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<Message> 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<Integer> flags){
+        public void voidMethodShouldBeIgnored(List<Integer> 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<Message> 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<Integer> flags) {
             int result = 0;
@@ -142,7 +148,7 @@ public class MethodInvokingMessageGroupProcessorTests {
             return result;
         }
 
-        public String listHeaderShouldBeIgnored(@Header List<Integer> flags){
+        public String listHeaderShouldBeIgnored(@Header List<Integer> flags) {
             fail("this method should not be invoked");
             return "";
         }
@@ -155,11 +161,12 @@ public class MethodInvokingMessageGroupProcessorTests {
         );
 
         ArgumentCaptor<Message> 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));

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<Message<?>> messagesUpForProcessing,
-                                   MessageChannel outputChannel, BufferedMessagesCallback processedCallback
+        public void processAndSend(MessageGroup group,
+                                   MessageChannel outputChannel
         ) {
             Integer product = 1;
+            List<Message<?>> 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<Message<?>> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
+        public void processAndSend(MessageGroup group, MessageChannel outputChannel) {
             //noop
         }
     }

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<Message<?>> messagesUpForProcessing,
-                                   MessageChannel outputChannel, BufferedMessagesCallback processedCallback
+     private class MultiplyingProcessor implements MessageGroupProcessor {
+        public void processAndSend(MessageGroup group,
+                                   MessageChannel outputChannel
         ) {
             Integer product = 1;
+            List<Message<?>> 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<Message<?>> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
+        public void processAndSend(MessageGroup group, MessageChannel outputChannel) {
             //noop
         }
     }

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<Message<?>> messages) {
+        public boolean isComplete(List<? extends Message<?>> messages) {
             return messages.size() == expectedSize;
         }
 

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<Message<?>> messages) {
+	public boolean isComplete(List<? extends Message<?>> messages) {
 		throw new UnsupportedOperationException("This is not intended to be implemented, but to verify injection into an <aggregator>");
 	}