OPEN - issue INT-1105: remove behaviour from MessageGroup. Still one test to fix....

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java

@@ -28,8 +28,8 @@ import org.springframework.util.Assert;
 import java.util.*;
 
 /**
- * Base class for MessageGroupProcessor implementations that aggregate the
- * group of Messages into a single Message.
+ * Base class for MessageGroupProcessor implementations that aggregate the group
+ * of Messages into a single Message.
  * 
  * @author Iwein Fuld
  * @author Alexander Peters
@@ -40,52 +40,48 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
-
-	public final void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
+	public final void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate,
+			MessageChannel outputChannel) {
 		Assert.notNull(group, "MessageGroup must not be null");
 		Assert.notNull(outputChannel, "'outputChannel' must not be null");
 		Object payload = this.aggregatePayloads(group);
 		Map<String, Object> headers = this.aggregateHeaders(group);
 		Message<?> message = MessageBuilder.withPayload(payload).copyHeadersIfAbsent(headers).build();
 		channelTemplate.send(message, outputChannel);
-		group.onCompleteProcessing();
 	}
 
 	/**
-	 * This default implementation simply returns all headers that have no conflicts
-	 * among the group. An absent header on one or more Messages within the group is
-	 * not considered a conflict. Subclasses may override this method with more
-	 * advanced conflict-resolution strategies if necessary.
+	 * This default implementation simply returns all headers that have no
+	 * conflicts among the group. An absent header on one or more Messages
+	 * within the group is not considered a conflict. Subclasses may override
+	 * this method with more advanced conflict-resolution strategies if
+	 * necessary.
 	 */
 	protected Map<String, Object> aggregateHeaders(MessageGroup group) {
 		Map<String, Object> aggregatedHeaders = new HashMap<String, Object>();
 		Set<String> conflictKeys = new HashSet<String>();
-		List<Message<?>> messages = group.getMessages();
-		if (messages != null) {
-			for (Message<?> message : messages) {
-				MessageHeaders currentHeaders = message.getHeaders();
-				for (String key : currentHeaders.keySet()) {
-					if (MessageHeaders.ID.equals(key) ||
-							MessageHeaders.TIMESTAMP.equals(key) ||
-							MessageHeaders.SEQUENCE_SIZE.equals(key)) {
-						continue;
-					}
-					Object value = currentHeaders.get(key);
-					if (!aggregatedHeaders.containsKey(key)) {
-						aggregatedHeaders.put(key, value);
-					}
-					else if (!value.equals(aggregatedHeaders.get(key))) {
-						conflictKeys.add(key);
-					}
+		for (Message<?> message : group.getMessages()) {
+			MessageHeaders currentHeaders = message.getHeaders();
+			for (String key : currentHeaders.keySet()) {
+				if (MessageHeaders.ID.equals(key) || MessageHeaders.TIMESTAMP.equals(key)
+						|| MessageHeaders.SEQUENCE_SIZE.equals(key)) {
+					continue;
+				}
+				Object value = currentHeaders.get(key);
+				if (!aggregatedHeaders.containsKey(key)) {
+					aggregatedHeaders.put(key, value);
+				}
+				else if (!value.equals(aggregatedHeaders.get(key))) {
+					conflictKeys.add(key);
 				}
 			}
-			for (String keyToRemove : conflictKeys) {
-				if (logger.isInfoEnabled()) {
-					logger.info("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) " +
-							"in MessageGroup with correlation key: " + group.getCorrelationKey());
-				}
-				aggregatedHeaders.remove(keyToRemove);
+		}
+		for (String keyToRemove : conflictKeys) {
+			if (logger.isInfoEnabled()) {
+				logger.info("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) "
+						+ "in MessageGroup with correlation key: " + group.getCorrelationKey());
 			}
+			aggregatedHeaders.remove(keyToRemove);
 		}
 		return aggregatedHeaders;
 	}

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategy.java

@@ -16,7 +16,7 @@
 
 package org.springframework.integration.aggregator;
 
-import java.util.List;
+import java.util.Collection;
 
 import org.springframework.integration.core.Message;
 
@@ -29,6 +29,6 @@ import org.springframework.integration.core.Message;
  */
 public interface CompletionStrategy {
 
-	boolean isComplete(List<? extends Message<?>> messages);
+	boolean isComplete(Collection<? extends Message<?>> messages);
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategyAdapter.java

@@ -17,7 +17,7 @@
 package org.springframework.integration.aggregator;
 
 import java.lang.reflect.Method;
-import java.util.List;
+import java.util.Collection;
 
 import org.springframework.integration.core.Message;
 import org.springframework.util.Assert;
@@ -43,7 +43,7 @@ public class CompletionStrategyAdapter extends MessageListMethodAdapter implemen
 	}
 
 
-	public boolean isComplete(List<? extends Message<?>> messages) {
+	public boolean isComplete(Collection<? extends Message<?>> messages) {
 		return ((Boolean) executeMethod(messages)).booleanValue();
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java

@@ -58,8 +58,9 @@ import org.springframework.util.Assert;
  * {@link org.springframework.integration.aggregator.MessageGroupProcessor}
  * implementations as you require.
  * <p/>
- * By default the CorrelationStrategy will be a HeaderAttributeCorrelationStrategy
- * and the CompletionStrategy will be a SequenceSizeCompletionStrategy.
+ * By default the CorrelationStrategy will be a
+ * HeaderAttributeCorrelationStrategy and the CompletionStrategy will be a
+ * SequenceSizeCompletionStrategy.
  * 
  * @author Iwein Fuld
  * @since 2.0
@@ -74,13 +75,12 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 
 	private static final long DEFAULT_TIMEOUT = 60000L;
 
-
 	private final MessageStore store;
 
 	private final MessageGroupProcessor outputProcessor;
 
-	private volatile CorrelationStrategy correlationStrategy =
-			new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID);
+	private volatile CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(
+			MessageHeaders.CORRELATION_ID);
 
 	private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
 
@@ -97,14 +97,13 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 	private volatile ScheduledFuture<?> reaperFutureTask;
 
 	private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL;
-	
+
 	private volatile long timeout = DEFAULT_TIMEOUT;
-	
+
 	private volatile boolean sendPartialResultOnTimeout;
 
 	private final Object lifecycleMonitor = new Object();
 
-
 	public CorrelatingMessageHandler(MessageStore store, CorrelationStrategy correlationStrategy,
 			CompletionStrategy completionStrategy, MessageGroupProcessor processor) {
 		Assert.notNull(store);
@@ -128,7 +127,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 				new SequenceSizeCompletionStrategy(), processor);
 	}
 
-
 	public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
 		Assert.notNull(correlationStrategy);
 		this.correlationStrategy = correlationStrategy;
@@ -185,18 +183,19 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		}
 		try {
 			if (tracker.waitForLockIfNotTracked(correlationKey)) {
-				MessageGroup group = new MessageGroup(store.list(correlationKey),
-						completionStrategy, correlationKey, deleteOrTrackCallback(correlationKey));
+				Collection<Message<?>> messages = store.list(correlationKey);
+				MessageGroup group = new MessageGroup(messages, correlationKey);
 
 				if (group.hasNoMessageSuperseding(message)) {
 					store(message, correlationKey);
 					group.add(message);
-					if (group.isComplete()) {
+					if (completionStrategy.isComplete(group.getMessages())) {
 						if (logger.isDebugEnabled()) {
 							logger.debug("Completing group with correlationKey [" + correlationKey + "]");
 						}
-						outputProcessor.processAndSend(group, channelTemplate,
-								this.resolveReplyChannel(message, this.outputChannel));
+						outputProcessor.processAndSend(group, channelTemplate, this.resolveReplyChannel(message,
+								this.outputChannel));
+						complete(group);
 					}
 				}
 				else {
@@ -212,20 +211,15 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
-	private MessageGroupListener deleteOrTrackCallback(final Object correlationKey) {
-		return new MessageGroupListener() {
-			
-			public void onProcessingOf(Message<?>... processedMessage) {
-				for (Message<?> message : processedMessage) {
-					store.delete(correlationKey, message.getHeaders().getId());
-				}
-			}
+	private void partialComplete(MessageGroup group) {
+		for (Message<?> message : group.getMessages()) {
+			store.delete(group.getCorrelationKey(), message.getHeaders().getId());
+		}
+	}
 
-			public void onCompletionOf(Object correlationKey) {
-				tracker.pushCorrelationId(correlationKey);
-				store.deleteAll(correlationKey);
-			}
-		};
+	private void complete(MessageGroup group) {
+		tracker.pushCorrelationId(group.getCorrelationKey());
+		store.deleteAll(group.getCorrelationKey());
 	}
 
 	@SuppressWarnings("unchecked")
@@ -252,8 +246,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 				return;
 			}
 			Assert.state(this.getTaskScheduler() != null, "'taskScheduler' must not be null");
-			this.reaperFutureTask = this.getTaskScheduler().scheduleWithFixedDelay(
-					new PrunerTask(), this.reaperInterval);
+			this.reaperFutureTask = this.getTaskScheduler().scheduleWithFixedDelay(new PrunerTask(),
+					this.reaperInterval);
 		}
 	}
 
@@ -265,7 +259,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
-
 	private class PrunerTask implements Runnable {
 		public void run() {
 			if (logger.isTraceEnabled()) {
@@ -289,38 +282,35 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
-
 	protected final boolean forceComplete(Object key) {
 		try {
 			if (tracker.tryLockFor(key)) {
 				Collection<Message<?>> all = store.list(key);
-				MessageGroup group = new MessageGroup(all, completionStrategy, key, deleteOrTrackCallback(key));
+				MessageGroup group = new MessageGroup(all, key);
 				if (all.size() > 0) {
 					// last chance for normal completion
 					MessageChannel outputChannel = resolveReplyChannel(all.iterator().next(), this.outputChannel);
-					boolean processed = false;
-					if (group.isComplete()) {
+					if (completionStrategy.isComplete(all)) {
 						outputProcessor.processAndSend(group, channelTemplate, outputChannel);
-						processed = true;
-					}
-					if (!processed) {
+						complete(group);
+					} else {
 						if (sendPartialResultOnTimeout) {
 							if (logger.isInfoEnabled()) {
-								logger.info("Processing partially complete messages for key [" +
-										key + "] to: " + outputChannel);
+								logger.info("Processing partially complete messages for key [" + key + "] to: "
+										+ outputChannel);
 							}
 							outputProcessor.processAndSend(group, channelTemplate, outputChannel);
 						}
 						else {
 							if (logger.isInfoEnabled()) {
-								logger.info("Discarding partially complete messages for key [" +
-										key + "] to: " + discardChannel);
+								logger.info("Discarding partially complete messages for key [" + key + "] to: "
+										+ discardChannel);
 							}
 							for (Message<?> message : all) {
 								discardChannel.send(message);
-								store.delete(key, message.getHeaders().getId());
 							}
 						}
+						partialComplete(group);
 					}
 				}
 				return true;
@@ -334,7 +324,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
-
 	private final class DelayedKey implements Delayed {
 
 		private final Object key;
@@ -362,7 +351,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
-
 	private final class IdTracker {
 
 		private final ConcurrentMap<Object, ReentrantLock> trackerLocks = new ConcurrentHashMap<Object, ReentrantLock>();
@@ -380,7 +368,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		/**
 		 * Call this method to check if an id is tracked and obtain a lock for
 		 * it. Don't forget to finally unlock afterwards.
-		 * @return false if the key was tracked, true after obtaining the lock otherwise
+		 * @return false if the key was tracked, true after obtaining the lock
+		 * otherwise
 		 */
 		private boolean waitForLockIfNotTracked(Object correlationKey) {
 			ReentrantLock lock = trackerLocks.get(correlationKey);

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.aggregator;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 
 import org.springframework.integration.core.Message;
@@ -35,7 +36,7 @@ public class DefaultAggregatingMessageGroupProcessor extends AbstractAggregating
 
 	@Override
 	protected final Object aggregatePayloads(MessageGroup group) {
-		List<Message<?>> messages = group.getMessages();
+		Collection<Message<?>> messages = group.getMessages();
 		Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups");
 		List<Object> payloads = new ArrayList<Object>(messages.size());
 		for (Message<?> message : messages) {

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.aggregator;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
@@ -30,7 +31,8 @@ import org.springframework.integration.core.MessageChannel;
 import org.springframework.integration.core.MessageHeaders;
 
 /**
- * This class implements all the strategy interfaces needed for a default resequencer.
+ * This class implements all the strategy interfaces needed for a default
+ * resequencer.
  * 
  * @author Iwein Fuld
  * @since 2.0
@@ -38,24 +40,25 @@ import org.springframework.integration.core.MessageHeaders;
 public class DefaultResequencerStrategies implements CorrelationStrategy, CompletionStrategy, MessageGroupProcessor {
 
 	private final ConcurrentMap<Object, AtomicInteger> nextMessagesToPass = new ConcurrentHashMap<Object, AtomicInteger>();
+	private final ConcurrentMap<Object, AtomicInteger> lastMessagesToPass = new ConcurrentHashMap<Object, AtomicInteger>();
 
 	private volatile SequenceNumberComparator sequenceNumberComparator = new SequenceNumberComparator();
 
 	private volatile boolean releasePartialSequences;
 
-
 	public Object getCorrelationKey(Message<?> message) {
 		Object key = message.getHeaders().getCorrelationId();
 		nextMessagesToPass.putIfAbsent(key, new AtomicInteger(1));
 		return key;
 	}
 
-	public boolean isComplete(List<? extends Message<?>> messages) {
-		return releasePartialSequences || messages.get(0).getHeaders().getSequenceSize() == messages.size();
+	public boolean isComplete(Collection<? extends Message<?>> messages) {
+		return releasePartialSequences
+				|| (!messages.isEmpty() && messages.iterator().next().getHeaders().getSequenceSize() == messages.size());
 	}
 
 	public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
-		List<Message<?>> all = group.getMessages();
+		Collection<Message<?>> all = group.getMessages();
 		Object correlationKey = group.getCorrelationKey();
 		if (all.size() > 0) {
 			List<Message<?>> sorted = new ArrayList<Message<?>>(all);
@@ -66,12 +69,11 @@ public class DefaultResequencerStrategies implements CorrelationStrategy, Comple
 				if (sequenceNumber <= nextSequence.get()) {
 					channelTemplate.send(message, outputChannel);
 					nextSequence.compareAndSet(sequenceNumber, sequenceNumber + 1);
-					group.onProcessingOf(message);
 				}
 			}
 			MessageHeaders headers = sorted.get(0).getHeaders();
 			if (all.size() == headers.getSequenceSize()) {
-				group.onCompletion();
+				// TODO: it's only complete if this is true...
 			}
 		}
 	}
@@ -80,7 +82,6 @@ public class DefaultResequencerStrategies implements CorrelationStrategy, Comple
 		this.releasePartialSequences = releasePartialSequences;
 	}
 
-
 	private static class SequenceNumberComparator implements Comparator<Message<?>> {
 		public int compare(Message<?> o1, Message<?> o2) {
 			return o1.getHeaders().getSequenceNumber().compareTo(o2.getHeaders().getSequenceNumber());

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java

@@ -17,18 +17,16 @@
 package org.springframework.integration.aggregator;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
 
 import org.springframework.integration.core.Message;
 
 /**
  * Represents a mutable group of correlated messages that is bound to a certain
  * {@link org.springframework.integration.store.MessageStore} and correlation
- * key. The group will grow during its lifetime, when messages are <code>add</code>ed to it.
- * <strong>This is not thread safe and should not be used for long running aggregations</strong>.
+ * key. The group will grow during its lifetime, when messages are
+ * <code>add</code>ed to it. <strong>This is not thread safe and should not be
+ * used for long running aggregations</strong>.
  * <p/>
  * According to its
  * {@link org.springframework.integration.aggregator.CompletionStrategy} it can
@@ -39,28 +37,20 @@ import org.springframework.integration.core.Message;
  * 
  * @author Iwein Fuld
  * @author Oleg Zhurakousky
+ * @author Dave Syer
  * @since 2.0
  */
 public class MessageGroup {
 
-	private final CompletionStrategy completionStrategy;
-
 	private final Object correlationKey;
 
-	private final ArrayList<Message<?>> messages = new ArrayList<Message<?>>();
+	private final Collection<Message<?>> messages = new ArrayList<Message<?>>();
 
-	private final List<MessageGroupListener> listeners;
-
-
-	public MessageGroup(Collection<? extends Message<?>> originalMessages, CompletionStrategy completionStrategy,
-			Object correlationKey, MessageGroupListener... listeners) {
-		this.completionStrategy = completionStrategy;
+	public MessageGroup(Collection<? extends Message<?>> originalMessages, Object correlationKey) {
 		this.correlationKey = correlationKey;
 		this.messages.addAll(originalMessages);
-		this.listeners = Collections.unmodifiableList(Arrays.asList(listeners));
 	}
 
-
 	/**
 	 * This method determines whether messages have been added to this group
 	 * that supersede the given message based on its sequence id. This can be
@@ -88,53 +78,20 @@ public class MessageGroup {
 		messages.add(message);
 	}
 
-	public boolean isComplete() {
-		return completionStrategy.isComplete(messages);
-	}
-
 	/**
 	 * @return internal message list, modification is allowed, but not
-	 *         recommended
+	 * recommended
 	 */
-	public List<Message<?>> getMessages() {
+	public Collection<Message<?>> getMessages() {
 		return messages;
 	}
 
 	/**
 	 * @return the correlation key that links these messages together according
-	 *         to a particular CorrelationStrategy
+	 * to a particular CorrelationStrategy
 	 */
 	public Object getCorrelationKey() {
 		return correlationKey;
 	}
 
-	/**
-	 * Call this method to sign off on processing of certain messages e.g. from
-	 * a MessageProcessor. Typically this will remove these messages from the
-	 * processing backlog.
-	 */
-	public void onProcessingOf(Message<?>... messages) {
-		for (MessageGroupListener listener : listeners) {
-			listener.onProcessingOf(messages);
-		}
-	}
-
-	/**
-	 * Call this method to signal the completion of the processing of an entire group.
-	 */
-	public void onCompletion() {
-		for (MessageGroupListener listener : listeners) {
-			listener.onCompletionOf(correlationKey);
-		}
-	}
-
-	/**
-	 * This method is a shorthand for signaling that all messages in the group have been
-	 * processed and that the group is completed.
-	 */
-	public void onCompleteProcessing() {
-		onProcessingOf(messages.toArray(new Message[messages.size()]));
-		onCompletion();
-    }
-
 }

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java

@@ -71,8 +71,6 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
 	protected final Object aggregatePayloads(MessageGroup group) {
 		final Collection<Message<?>> messagesUpForProcessing = group.getMessages();
 		Object result = this.adapter.executeMethod(messagesUpForProcessing);
-		group.onCompletion();
-		group.onProcessingOf(messagesUpForProcessing.toArray(new Message[messagesUpForProcessing.size()]));
 		return result;
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java

@@ -17,8 +17,7 @@ public class PassThroughMessageGroupProcessor implements MessageGroupProcessor {
     public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
         for (Message<?> message : group.getMessages()) {
             channelTemplate.send(message, outputChannel);
-            group.onProcessingOf(message);
         }
-        group.onCompletion();
     }
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeCompletionStrategy.java

@@ -16,7 +16,7 @@
 
 package org.springframework.integration.aggregator;
 
-import java.util.List;
+import java.util.Collection;
 
 import org.springframework.integration.core.Message;
 import org.springframework.util.CollectionUtils;
@@ -31,11 +31,11 @@ import org.springframework.util.CollectionUtils;
  */
 public class SequenceSizeCompletionStrategy implements CompletionStrategy {
 
-	public boolean isComplete(List<? extends Message<?>> messages) {
+	public boolean isComplete(Collection<? extends Message<?>> messages) {
 		if (CollectionUtils.isEmpty(messages)) {
 			return false;
 		}
-		return messages.size() != 0 && (messages.size() >= messages.get(0).getHeaders().getSequenceSize());
+		return messages.size() != 0 && (messages.size() >= messages.iterator().next().getHeaders().getSequenceSize());
 	}
 
 }