INT-2152

parent issue for INT-2134, INT-2135, INT-2142, INT-2155, INT-2158

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java

@@ -41,6 +41,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
 
 	public final Object processMessageGroup(MessageGroup group) {
 		Assert.notNull(group, "MessageGroup must not be null");
+		
 		Map<String, Object> headers = this.aggregateHeaders(group);
 		Object payload = this.aggregatePayloads(group, headers);
 		MessageBuilder<?> builder;
@@ -50,6 +51,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
 		else {
 			builder = MessageBuilder.withPayload(payload).copyHeadersIfAbsent(headers);
 		}
+		
 		return builder.popSequenceDetails().build();
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  * 
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
@@ -13,7 +13,10 @@
 
 package org.springframework.integration.aggregator;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -34,34 +37,36 @@ import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 
 /**
- * Message handler that holds a buffer of correlated messages in a
+ * Abstract Message handler that holds a buffer of correlated messages in a
  * {@link 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.
+ * that can be completed in batches. It is useful for custom implementation of MessageHandlers that require correlation 
+ * and is used as a base class for Aggregator - {@link AggregatingMessageHandler} and
+ * Resequencer - {@link ResequencingMessageHandler},
+ * or custom implementations requiring correlation. 
  * <p/>
  * To customize this handler inject {@link CorrelationStrategy},
  * {@link ReleaseStrategy}, and {@link MessageGroupProcessor} implementations as
  * you require.
  * <p/>
- * By default the CorrelationStrategy will be a
- * HeaderAttributeCorrelationStrategy and the ReleaseStrategy will be a
- * SequenceSizeReleaseStrategy.
+ * By default the {@link CorrelationStrategy} will be a
+ * {@link HeaderAttributeCorrelationStrategy} and the {@link ReleaseStrategy} will be a
+ * {@link SequenceSizeReleaseStrategy}.
  *
  * @author Iwein Fuld
  * @author Dave Syer
  * @author Oleg Zhurakousky
  * @since 2.0
  */
-public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer {
+public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer {
 
-	private static final Log logger = LogFactory.getLog(CorrelatingMessageHandler.class);
+	private static final Log logger = LogFactory.getLog(AbstractCorrelatingMessageHandler.class);
 
 	public static final long DEFAULT_SEND_TIMEOUT = 1000L;
 
-
-	private MessageGroupStore messageStore;
+	protected volatile MessageGroupStore messageStore;
 
 	private final MessageGroupProcessor outputProcessor;
 
@@ -80,9 +85,15 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 	private final Object correlationLocksMonitor = new Object();
 
 	private final ConcurrentMap<Object, Object> locks = new ConcurrentHashMap<Object, Object>();
+	
+	protected volatile boolean keepReleasedMessages = true;
 
 
-	public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
+	public void setKeepReleasedMessages(boolean keepReleasedMessages) {
+		this.keepReleasedMessages = keepReleasedMessages;
+	}
+
+	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
 									 CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
 		Assert.notNull(processor);
 		Assert.notNull(store);
@@ -94,11 +105,11 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		this.messagingTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
 	}
 
-	public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store) {
+	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store) {
 		this(processor, store, null, null);
 	}
 
-	public CorrelatingMessageHandler(MessageGroupProcessor processor) {
+	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor) {
 		this(processor, new SimpleMessageStore(0), null, null);
 	}
 
@@ -159,72 +170,59 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 	public String getComponentType() {
 		return "aggregator";
 	}
+	
+	protected MessageGroupStore getMessageStore() {
+		return messageStore;
+	}
 
-	@SuppressWarnings("rawtypes")
 	@Override
 	protected void handleMessageInternal(Message<?> message) throws Exception {
 		Object correlationKey = correlationStrategy.getCorrelationKey(message);
 		Assert.state(correlationKey!=null, "Null correlation not allowed.  Maybe the CorrelationStrategy is failing?");
 
 		if (logger.isDebugEnabled()) {
-			logger.debug("Handling message with correlationKey ["
-					+ correlationKey + "]: " + message);
+			logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message);
 		}
 
 		// TODO: INT-1117 - make the lock global?
 		Object lock = getLock(correlationKey);
 
 		synchronized (lock) {
-			MessageGroup group = messageStore.getMessageGroup(correlationKey);
-			if (group.canAdd(message)) {
+			MessageGroup messageGroup = messageStore.getMessageGroup(correlationKey);
+			if (!messageGroup.isComplete() && messageGroup.canAdd(message)) {
 				if (logger.isTraceEnabled()) {
-					logger.trace("Adding message to group [ " + group + "]");
+					logger.trace("Adding message to group [ " + messageGroup + "]");
 				}
-				group = store(correlationKey, message);
-				if (releaseStrategy.canRelease(group)) {
-					Collection<Message> completedMessages = null;
+				messageGroup = store(correlationKey, message);
+				
+				if (releaseStrategy.canRelease(messageGroup)) {
+					Collection<Message<?>> completedMessages = null;
 					try {
-						completedMessages = completeGroup(message, correlationKey, group);
+						completedMessages = completeGroup(message, correlationKey, messageGroup);
 					}
 					finally {
 						// Always clean up even if there was an exception
-						// processing messages
-						cleanUpForReleasedGroup(group, completedMessages);
-					}
-				} else if (group.isComplete()) {
-					try {
-						// If not releasing any messages the group might still
-						// be complete
-						for (Message<?> discard : group.getUnmarked()) {
-							discardChannel.send(discard);
+						// processing messages						
+						this.afterRelease(messageGroup, completedMessages);
+		
+						synchronized(correlationLocksMonitor){
+							locks.remove(messageGroup.getGroupId());
 						}
 					}
-					finally {
-						remove(group);
-					}
-				}
-			} else {
+				} 				
+			} 
+			else {
 				discardChannel.send(message);
 			}
 		}
 	}
 
-	@SuppressWarnings("rawtypes")
-	private void cleanUpForReleasedGroup(MessageGroup group, Collection<Message> completedMessages) {
-		if (group.isComplete() || group.getSequenceSize() == 0) {
-			// The group is complete or else there is no
-			// sequence so there is no more state to track
-			remove(group);
-		} else {
-			// Mark these messages as processed, but do not
-			// remove the group from store
-			if (completedMessages == null) {
-				mark(group);
-			} else {
-				mark(group, completedMessages);
-			}
-		}
-	}
+	/**
+	 * Allows you to provide additional logic that needs to be performed after the MessageGroup was released. 
+	 * @param group
+	 * @param completedMessages
+	 */
+	protected abstract void afterRelease(MessageGroup group, Collection<Message<?>> completedMessages);
 
 	private final boolean forceComplete(MessageGroup group) {
 
@@ -235,13 +233,14 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 			if (group.size() > 0) {
 				try {
 					if (releaseStrategy.canRelease(group)) {
-						completeGroup(correlationKey, group);
-					} else {
-						expireGroup(group, correlationKey);
+						this.completeGroup(correlationKey, group);
+					} 
+					else {
+						this.expireGroup(correlationKey, group);
 					}
 				}
 				finally {
-					remove(group);
+					this.remove(group);
 				}
 				return true;
 			}
@@ -256,31 +255,25 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
-	private void mark(MessageGroup group) {
-		messageStore.markMessageGroup(group);
-	}
-
-	@SuppressWarnings("rawtypes")
-	private void mark(MessageGroup group, Collection<Message> partialSequence) {
-		Object id = group.getGroupId();
-		for (Message message : partialSequence) {
-			messageStore.markMessageFromGroup(id, message);
-		}
-	}
-
-	private void remove(MessageGroup group) {
+	void remove(MessageGroup group) {
 		Object correlationKey = group.getGroupId();
 		messageStore.removeMessageGroup(correlationKey);
-		synchronized(correlationLocksMonitor){
-			locks.remove(correlationKey);
-		}
+	}
+
+	protected int findLastReleasedSequenceNumber(Object groupId, Collection<Message<?>> partialSequence){
+		List<Message<?>> sorted = new ArrayList<Message<?>>((Collection<? extends Message<?>>)partialSequence);
+		Collections.sort(sorted, new SequenceNumberComparator());
+		
+		Message<?> lastReleasedMessage = sorted.get(partialSequence.size()-1);
+		
+		return lastReleasedMessage.getHeaders().getSequenceNumber();
 	}
 
 	private MessageGroup store(Object correlationKey, Message<?> message) {
 		return messageStore.addMessageToGroup(correlationKey, message);
 	}
 
-	private void expireGroup(MessageGroup group, Object correlationKey) {
+	private void expireGroup(Object correlationKey, MessageGroup group) {
 		if (logger.isInfoEnabled()) {
 			logger.info("Expiring MessageGroup with correlationKey[" + correlationKey + "]");
 		}
@@ -309,21 +302,25 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		completeGroup(first, correlationKey, group);
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	private Collection<Message> completeGroup(Message<?> message, Object correlationKey, MessageGroup group) {
+	@SuppressWarnings("unchecked")
+	private Collection<Message<?>> completeGroup(Message<?> message, Object correlationKey, MessageGroup group) {
 		if (logger.isDebugEnabled()) {
-			logger.debug("Completing group with correlationKey ["
-					+ correlationKey + "]");
+			logger.debug("Completing group with correlationKey [" + correlationKey + "]");
 		}
 		Object result = outputProcessor.processMessageGroup(group);
-		Collection<Message> partialSequence = null;
+		Collection<Message<?>> partialSequence = null;
 		if (result instanceof Collection<?>) {
-			//Taking a risk here because of Type Erasure. This is covered in the processor contract
-			partialSequence = (Collection<Message>) result;
+			this.verifyResultCollectionConsistsOfMessages((Collection<?>) result);
+			partialSequence = (Collection<Message<?>>) result;
 		}
 		this.sendReplies(result, message);
 		return partialSequence;
 	}
+	
+	private void verifyResultCollectionConsistsOfMessages(Collection<?> elements){
+		Class<?> commonElementType = CollectionUtils.findCommonElementType(elements);
+		Assert.isAssignable(Message.class, commonElementType, "The expected collection of Messages contains non-Message element: " + commonElementType);
+	}
 
 	@SuppressWarnings("rawtypes")
 	private void sendReplies(Object processorResult, Message message) {

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2002-2011 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package org.springframework.integration.aggregator;
+
+import java.util.Collection;
+import java.util.Iterator;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.MessageGroupStore;
+
+/**
+ * Aggregator specific implementation of {@link AbstractCorrelatingMessageHandler}. 
+ * Will remove {@link MessageGroup}s only if 'expireGroupsUponCompletion' flag is set to 'true'.
+ *
+ * @author Oleg Zhurakousky
+ * @since 2.1
+ */
+public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler {
+
+	private volatile boolean expireGroupsUponCompletion = false;
+
+
+	public AggregatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
+			CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
+		super(processor, store, correlationStrategy, releaseStrategy);
+	}
+
+	public AggregatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store) {
+		super(processor, store);
+	}
+	
+	public AggregatingMessageHandler(MessageGroupProcessor processor) {
+		super(processor);
+	}
+
+	/**
+	 * Will set the 'expireGroupsUponCompletion' flag and if it is 
+	 * set to 'true' it will also remove all 'complete' {@link MessageGroup}s
+	 * @param expireGroupsUponCompletion
+	 */
+	public void setExpireGroupsUponCompletion(boolean expireGroupsUponCompletion) {
+		this.expireGroupsUponCompletion = expireGroupsUponCompletion;
+		if (expireGroupsUponCompletion) {
+			Iterator<MessageGroup> messageGroups = this.messageStore.iterator();
+			while (messageGroups.hasNext()) {
+				MessageGroup messageGroup = messageGroups.next();
+				if (messageGroup.isComplete()) {
+					remove(messageGroup);
+				}
+			}
+		}
+	}
+
+	@Override
+	protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages) {
+		this.messageStore.completeGroup(messageGroup.getGroupId());
+		
+		if (this.expireGroupsUponCompletion) {
+			remove(messageGroup);
+		}
+		else {
+			if (this.keepReleasedMessages){
+				messageStore.markMessageGroup(messageGroup);
+			}
+			else {
+				for (Message<?> message : messageGroup.getMarked()) {
+					this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), message);
+				}
+				for (Message<?> message : messageGroup.getUnmarked()) {
+					this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), message);
+				}
+			}
+		}	
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java

@@ -23,6 +23,7 @@ import java.util.*;
  *
  * @author Iwein Fuld
  * @author Dave Syer
+ * @author Oleg Zhurakousky
  * @since 2.0
  */
 public class ResequencingMessageGroupProcessor implements MessageGroupProcessor {
@@ -37,15 +38,14 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
 	public void setComparator(Comparator<Message<?>> comparator) {
 		this.comparator = comparator;
 	}
-
-	@SuppressWarnings("rawtypes")
+	
 	public Object processMessageGroup(MessageGroup group) {
 		Collection<Message<?>> messages = group.getUnmarked();
 
 		if (messages.size() > 0) {
 			List<Message<?>> sorted = new ArrayList<Message<?>>(messages);
 			Collections.sort(sorted, this.comparator);
-			ArrayList<Message> partialSequence = new ArrayList<Message>();
+			ArrayList<Message<?>> partialSequence = new ArrayList<Message<?>>();
 			int previousSequence = extractSequenceNumber(sorted.get(0));
 			int currentSequence = previousSequence;
 			for (Message<?> message : sorted) {
@@ -57,6 +57,7 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
 				}
 				partialSequence.add(message);
 			}
+			
 			return partialSequence;
 		}
 		return null;

spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2002-2011 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package org.springframework.integration.aggregator;
+
+import java.util.Collection;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.MessageGroupStore;
+
+/**
+ * Resequencer specific implementation of {@link AbstractCorrelatingMessageHandler}. 
+ * Will remove {@link MessageGroup}s only if 'sequenceSize' is provided and reached. 
+ *
+ * @author Oleg Zhurakousky
+ * @since 2.1
+ */
+public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandler {
+
+	public ResequencingMessageHandler(MessageGroupProcessor processor,
+			MessageGroupStore store, CorrelationStrategy correlationStrategy,
+			ReleaseStrategy releaseStrategy) {
+		super(processor, store, correlationStrategy, releaseStrategy);
+	}
+
+	
+	public ResequencingMessageHandler(MessageGroupProcessor processor,
+			MessageGroupStore store) {
+		super(processor, store);
+	}
+
+	
+	public ResequencingMessageHandler(MessageGroupProcessor processor) {
+		super(processor);
+	}
+	
+	@Override
+	protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages) {
+			
+		int size = messageGroup.getUnmarked().size() + messageGroup.getMarked().size();
+		int sequenceSize = 0;
+		Message<?> message = messageGroup.getOne();
+		if (message != null){
+			sequenceSize = message.getHeaders().getSequenceSize();
+		}
+		// If there is no sequence then it must be incomplete or unbounded
+		if (sequenceSize > 0 && sequenceSize == size){
+			remove(messageGroup);
+		}
+		else {
+			if (completedMessages != null){ 
+				int lastReleasedSequenceNumber = this.findLastReleasedSequenceNumber(messageGroup.getGroupId(), completedMessages);
+				messageStore.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), lastReleasedSequenceNumber);
+				
+				if (this.keepReleasedMessages){
+					Object id = messageGroup.getGroupId();
+					for (Message<?> msg : completedMessages) {
+						messageStore.markMessageFromGroup(id, msg);
+					}
+				}
+				else {
+					for (Message<?> msg : completedMessages) {
+						this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), msg);
+					}
+				}
+			}
+		}
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,17 +16,17 @@
 
 package org.springframework.integration.aggregator;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.integration.Message;
-import org.springframework.integration.store.MessageGroup;
-
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.Message;
+import org.springframework.integration.store.MessageGroup;
+
 /**
  * An implementation of {@link ReleaseStrategy} that simply compares the current size of the message list to the
  * expected 'sequenceSize'.
@@ -35,6 +35,7 @@ import java.util.List;
  * @author Marius Bogoevici
  * @author Dave Syer
  * @author Iwein Fuld
+ * @author Oleg Zhurakousky
  */
 public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
 
@@ -62,24 +63,42 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
 		this.releasePartialSequences = releasePartialSequences;
 	}
 
-	public boolean canRelease(MessageGroup messages) {
-		if (releasePartialSequences) {
-			Collection<Message<?>> unmarked = messages.getUnmarked();
-			if (!unmarked.isEmpty()) {
-				if (logger.isTraceEnabled()) {
-					logger.trace("Considering partial release of group [" + messages + "]");
-				}
-				List<Message<?>> sorted = new ArrayList<Message<?>>(unmarked);
-				Collections.sort(sorted, comparator);
-				int tail = sorted.get(0).getHeaders().getSequenceNumber() - 1;
-				boolean release = tail == messages.getMarked().size();
-				if (logger.isTraceEnabled() && release) {
-					logger.trace("Release imminent because tail [" + tail + "] is next in line.");
-				}
-				return release;
+	public boolean canRelease(MessageGroup messageGroup) {
+
+		boolean canRelease = false;
+		
+		Collection<Message<?>> unmarked = messageGroup.getUnmarked();
+		
+		if (releasePartialSequences && !unmarked.isEmpty()) {
+			
+			if (logger.isTraceEnabled()) {
+				logger.trace("Considering partial release of group [" + messageGroup + "]");
 			}
+			List<Message<?>> sorted = new ArrayList<Message<?>>(unmarked);
+			Collections.sort(sorted, comparator);
+			
+			int nextSequenceNumber = sorted.get(0).getHeaders().getSequenceNumber();
+			int lastReleasedMessageSequence = messageGroup.getLastReleasedMessageSequenceNumber();
+			
+			if (nextSequenceNumber - lastReleasedMessageSequence == 1){
+				canRelease = true;;
+			}	
 		}
-		return messages.isComplete();
+		else {
+			int size = messageGroup.getUnmarked().size();
+			
+			if (size == 0){
+				canRelease = true;
+			}
+			else {
+				int sequenceSize = messageGroup.getOne().getHeaders().getSequenceSize();
+				// If there is no sequence then it must be incomplete....
+				if (sequenceSize == size){
+					canRelease = true;
+				}
+			}
+		}	
+		return canRelease;
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,7 +22,7 @@ import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
-import org.springframework.integration.aggregator.CorrelatingMessageHandler;
+import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler;
 
 /**
  * Indicates that a method is capable of aggregating messages. 
@@ -32,6 +32,7 @@ import org.springframework.integration.aggregator.CorrelatingMessageHandler;
  * Message or a single Object to be used as a Message payload.
  * 
  * @author Marius Bogoevici
+ * @author Oleg Zhurakousky
  */
 @Target(ElementType.METHOD)
 @Retention(RetentionPolicy.RUNTIME)
@@ -56,7 +57,7 @@ public @interface Aggregator {
 	/**
 	 * timeout for sending results to the reply target (in milliseconds)
 	 */
-	long sendTimeout() default CorrelatingMessageHandler.DEFAULT_SEND_TIMEOUT;
+	long sendTimeout() default AbstractCorrelatingMessageHandler.DEFAULT_SEND_TIMEOUT;
 
 	/**
 	 * indicates whether to send an incomplete aggregate on expiry of the message group

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2009 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,7 +23,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.integration.MessageChannel;
-import org.springframework.integration.aggregator.CorrelatingMessageHandler;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
 import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
 import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
 import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
@@ -40,6 +40,7 @@ import org.springframework.util.StringUtils;
  * Post-processor for the {@link Aggregator @Aggregator} annotation.
  * 
  * @author Mark Fisher
+ * @author Oleg Zhurakousky
  */
 public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Aggregator> {
 
@@ -53,7 +54,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
 		MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(bean, method);
 		MethodInvokingReleaseStrategy releaseStrategy = getReleaseStrategy(bean);
 		MethodInvokingCorrelationStrategy correlationStrategy = getCorrelationStrategy(bean);
-		CorrelatingMessageHandler handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, releaseStrategy);
+		AggregatingMessageHandler handler = new AggregatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, releaseStrategy);
 		String discardChannelName = annotation.discardChannel();
 		if (StringUtils.hasText(discardChannelName)) {
 			MessageChannel discardChannel = this.channelResolver.resolveChannelName(discardChannelName);

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java

@@ -21,6 +21,8 @@ import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.parsing.BeanComponentDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
+import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
 import org.springframework.util.StringUtils;
 import org.w3c.dom.Element;
 
@@ -60,6 +62,10 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
 	private static final String RELEASE_STRATEGY_PROPERTY = "releaseStrategy";
 
 	private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy";
+	
+	private static final String EXPIRE_GROUPS_UPON_COMPLETION = "expire-groups-upon-completion";
+
+	private static final String KEEP_RELEASED_MESSAGES = "keep-released-messages";
 
 	@Override
 	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
@@ -68,14 +74,12 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
 		String ref = element.getAttribute(REF_ATTRIBUTE);
 		BeanDefinitionBuilder builder;
 
-		builder = BeanDefinitionBuilder.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
-				+ ".aggregator.CorrelatingMessageHandler");
+		builder = BeanDefinitionBuilder.genericBeanDefinition(AggregatingMessageHandler.class);
 		BeanDefinitionBuilder processorBuilder = null;
 		BeanMetadataElement processor = null;
 
 		if (innerHandlerDefinition != null || StringUtils.hasText(ref)) {
-			processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
-					+ ".aggregator.MethodInvokingMessageGroupProcessor");
+			processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingMessageGroupProcessor.class);
 			builder.addConstructorArgValue(processorBuilder.getBeanDefinition());
 			if (innerHandlerDefinition != null) {
 				processor = innerHandlerDefinition;
@@ -110,8 +114,10 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_COMPLETION);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, KEEP_RELEASED_MESSAGES);
 		this.injectPropertyWithAdapter(RELEASE_STRATEGY_REF_ATTRIBUTE, RELEASE_STRATEGY_METHOD_ATTRIBUTE,
 				RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE, RELEASE_STRATEGY_PROPERTY, "ReleaseStrategy", element, builder,
 				processor, parserContext);

spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2009 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  * 
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
@@ -18,6 +18,8 @@ import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
+import org.springframework.integration.aggregator.ResequencingMessageHandler;
 import org.springframework.util.StringUtils;
 import org.w3c.dom.Element;
 
@@ -27,6 +29,7 @@ import org.w3c.dom.Element;
  * @author Marius Bogoevici
  * @author Dave Syer
  * @author Iwein Fuld
+ * @author Oleg Zhurakousky
  */
 public class ResequencerParser extends AbstractConsumerEndpointParser {
 
@@ -53,15 +56,14 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
 	private static final String RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE = "release-strategy-expression";
 
 	private static final String RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE = "release-partial-sequences";
+	
+	private static final String KEEP_RELEASED_MESSAGES = "keep-released-messages";
 
 	@Override
 	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
 
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder
-				.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelatingMessageHandler");
-		BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder
-				.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
-						+ ".aggregator.ResequencingMessageGroupProcessor");
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ResequencingMessageHandler.class);
+		BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(ResequencingMessageGroupProcessor.class);
 
 		// Comparator
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(processorBuilder, element, COMPARATOR_REF_ATTRIBUTE);
@@ -86,6 +88,7 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, KEEP_RELEASED_MESSAGES);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
 		return builder;
 	}

spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java

@@ -0,0 +1,233 @@
+/*
+ * Copyright 2002-2011 the original author or authors
+ *
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ */
+
+package org.springframework.integration.store;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.UUID;
+
+import org.springframework.integration.Message;
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for implementations of Key/Value style {@link MessageGroupStore} and {@link MessageStore}
+ * 
+ * @author Oleg Zhurakousky
+ * @since 2.1
+ */
+public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupStore implements MessageStore{
+
+	protected static final String MESSAGE_KEY_PREFIX = "MESSAGE_";
+
+	protected static final String MESSAGE_GROUP_KEY_PREFIX = "MESSAGE_GROUP_";
+
+
+	// MessageStore methods
+
+	public Message<?> getMessage(UUID id) {
+		Assert.notNull(id, "'id' must not be null");
+		Object message = this.doRetrieve(MESSAGE_KEY_PREFIX + id);
+		if (message != null) {
+			Assert.isInstanceOf(Message.class, message);
+		}	
+		return (Message<?>) message;
+	}
+
+	@SuppressWarnings("unchecked")
+	public <T> Message<T> addMessage(Message<T> message) {
+		Assert.notNull(message, "'message' must not be null");
+		UUID messageId = message.getHeaders().getId(); 
+		this.doStore(MESSAGE_KEY_PREFIX + messageId, message);
+		return (Message<T>) this.getMessage(messageId);
+	}
+
+	public Message<?> removeMessage(UUID id) {
+		Assert.notNull(id, "'id' must not be null");
+		Object message = this.doRemove(MESSAGE_KEY_PREFIX + id);
+		if (message != null) {
+			Assert.isInstanceOf(Message.class, message);
+		}
+		return (Message<?>) message;
+	}
+
+	@ManagedAttribute
+	public long getMessageCount() {
+		Collection<?> messageIds = this.doListKeys(MESSAGE_KEY_PREFIX + "*");
+		return (messageIds != null) ? messageIds.size() : 0;
+	}
+
+
+	// MessageGroupStore methods
+
+	/**
+	 * Will create a new instance of SimpleMessageGroup if necessary.
+	 */
+	public MessageGroup getMessageGroup(Object groupId) {
+		Assert.notNull(groupId, "'groupId' must not be null");
+		Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
+		if (mgm != null) {
+			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
+			MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
+			ArrayList<Message<?>> markedMessages = new ArrayList<Message<?>>();
+			for (UUID uuid : messageGroupMetadata.getMarkedMessageIds()) {
+				markedMessages.add(this.getMessage(uuid));
+			}
+			ArrayList<Message<?>> unmarkedMessages = new ArrayList<Message<?>>();
+			for (UUID uuid : messageGroupMetadata.getUnmarkedMessageIds()) {
+				unmarkedMessages.add(this.getMessage(uuid));
+			}
+			SimpleMessageGroup messageGroup = new SimpleMessageGroup(unmarkedMessages, markedMessages, 
+						groupId, messageGroupMetadata.getTimestamp(), messageGroupMetadata.isComplete());
+			if (messageGroupMetadata.getLastReleasedMessageSequenceNumber() > 0) {
+				messageGroup.setLastReleasedMessageSequenceNumber(messageGroupMetadata.getLastReleasedMessageSequenceNumber());
+			}
+			return messageGroup;
+		}
+		else {
+			return new SimpleMessageGroup(groupId);
+		}
+	}
+
+	/**
+	 * Add a Message to the group with the provided group ID. 
+	 */
+	public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
+		Assert.notNull(groupId, "'groupId' must not be null");
+		Assert.notNull(message, "'message' must not be null");
+		SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
+		messageGroup.add(message);	
+		this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
+		this.addMessage(message);
+		return messageGroup;
+	}
+
+	/**
+	 * Mark all messages in the provided group. 
+	 */
+	public MessageGroup markMessageGroup(MessageGroup group) {
+		Assert.notNull(group, "'group' must not be null");
+		Object groupId = group.getGroupId();
+		SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(group);
+		messageGroup.markAll();
+		this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
+		return messageGroup;
+	}
+
+	/**
+	 * Remove a Message from the group with the provided group ID. 
+	 */
+	public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
+		Assert.notNull(groupId, "'groupId' must not be null");
+		Assert.notNull(messageToRemove, "'messageToRemove' must not be null");
+		SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
+		messageGroup.remove(messageToRemove);
+		this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
+		return messageGroup;
+	}
+
+	/**
+	 * Mark the given Message within the group corresponding to the provided group ID. 
+	 */
+	public MessageGroup markMessageFromGroup(Object groupId, Message<?> messageToMark) {
+		Assert.notNull(groupId, "'groupId' must not be null");
+		Assert.notNull(messageToMark, "'messageToMark' must not be null");
+		SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
+		messageGroup.mark(messageToMark);
+		this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
+		return messageGroup;
+	}
+
+	public void completeGroup(Object groupId) {
+		Assert.notNull(groupId, "'groupId' must not be null");
+		SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
+		messageGroup.complete();
+		this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
+	}
+
+	/**
+	 * Remove the MessageGroup with the provided group ID. 
+	 */
+	public void removeMessageGroup(Object groupId) {
+		Assert.notNull(groupId, "'groupId' must not be null");
+		Object mgm = this.doRemove(MESSAGE_GROUP_KEY_PREFIX + groupId);
+		if (mgm != null) {
+			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
+			MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
+			for (UUID messageId : messageGroupMetadata.getMarkedMessageIds()) {
+				this.removeMessage(messageId);
+			}
+			for (UUID messageId : messageGroupMetadata.getUnmarkedMessageIds()) {
+				this.removeMessage(messageId);
+			}
+		}
+	}
+
+	public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
+		Assert.notNull(groupId, "'groupId' must not be null");
+		SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
+		messageGroup.setLastReleasedMessageSequenceNumber(sequenceNumber);
+		this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
+	}
+
+	public Iterator<MessageGroup> iterator() {
+		final Iterator<?> idIterator = this.doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*").iterator();
+		return new MessageGroupIterator(idIterator);
+	}
+
+	private SimpleMessageGroup getSimpleMessageGroup(MessageGroup messageGroup){
+		if (messageGroup instanceof SimpleMessageGroup){
+			return (SimpleMessageGroup) messageGroup;
+		}
+		else {
+			return new SimpleMessageGroup(messageGroup);
+		}
+	}
+
+	protected abstract Object doRetrieve(Object id);
+	
+	protected abstract void doStore(Object id, Object objectToStore);
+	
+	protected abstract Object doRemove(Object id);  
+
+	protected abstract Collection<?> doListKeys(String keyPattern);
+
+
+	private class MessageGroupIterator implements Iterator<MessageGroup> {
+
+		private final Iterator<?> idIterator;
+
+		private MessageGroupIterator(Iterator<?> idIterator) {
+			this.idIterator = idIterator;
+		}
+
+		public boolean hasNext() {
+			return idIterator.hasNext();
+		}
+
+		public MessageGroup next() {
+			Object messageGroupId = idIterator.next();
+			return getMessageGroup(messageGroupId);
+		}
+
+		public void remove() {
+			throw new UnsupportedOperationException();
+		}
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java

@@ -14,7 +14,6 @@
 package org.springframework.integration.store;
 
 import java.util.Collection;
-import java.util.Iterator;
 import java.util.LinkedHashSet;
 
 import org.apache.commons.logging.Log;
@@ -67,8 +66,6 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It
 		}
 		return count;
 	}
-	
-	public abstract Iterator<MessageGroup> iterator();
 
 	@ManagedAttribute
 	public int getMessageCountForAllMessageGroups() {

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java

@@ -16,10 +16,10 @@
 
 package org.springframework.integration.store;
 
-import org.springframework.integration.Message;
-
 import java.util.Collection;
 
+import org.springframework.integration.Message;
+
 /**
  * A group of messages that are correlated with each other and should be processed in the same context. The group is
  * divided into marked and unmarked messages. The marked messages are typically already processed, the unmarked messages
@@ -49,11 +49,21 @@ public interface MessageGroup {
 	 * @return the key that links these messages together
 	 */
 	Object getGroupId();
+	
+	/**
+	 * Returns the sequenceNumber of the last released message. Used in Resequencer use cases only
+	 */
+	int getLastReleasedMessageSequenceNumber();
 
 	/**
 	 * @return true if the group is complete (i.e. no more messages are expected to be added)
 	 */
 	boolean isComplete();
+	
+	/**
+	 * 
+	 */
+	void complete();
 
 	/**
 	 * @return the size of the sequence expected 0 if unknown

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2002-2011 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.store;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import org.springframework.integration.Message;
+import org.springframework.util.Assert;
+
+/**
+ * Immutable Value Object holding metadata about a MessageGroup.
+ * 
+ * @author Oleg Zhurakousky
+ * @since 2.1
+ */
+public class MessageGroupMetadata implements Serializable{
+
+	private static final long serialVersionUID = 1L;
+
+
+	private final Object groupId;
+
+	private final List<UUID> markedMessageIds;
+
+	private final List<UUID> unmarkedMessageIds;
+
+	private final boolean complete;
+
+	private final long timestamp;
+
+	private final int lastReleasedMessageSequenceNumber;
+
+
+	public MessageGroupMetadata(MessageGroup messageGroup) {
+		Assert.notNull(messageGroup, "'messageGroup' must not be null");
+		this.groupId = messageGroup.getGroupId();
+		this.markedMessageIds = new ArrayList<UUID>();
+		for (Message<?> message : messageGroup.getMarked()) {
+			this.markedMessageIds.add(message.getHeaders().getId());
+		}
+		this.unmarkedMessageIds = new ArrayList<UUID>();
+		for (Message<?> message : messageGroup.getUnmarked()) {
+			this.unmarkedMessageIds.add(message.getHeaders().getId());
+		}
+		this.complete = messageGroup.isComplete();
+		this.timestamp = messageGroup.getTimestamp();
+		this.lastReleasedMessageSequenceNumber = messageGroup.getLastReleasedMessageSequenceNumber();
+	}
+
+
+	public Object getGroupId() {
+		return this.groupId;
+	}
+
+	public List<UUID> getMarkedMessageIds() {
+		return Collections.unmodifiableList(markedMessageIds);
+	}
+
+	public List<UUID> getUnmarkedMessageIds() {
+		return Collections.unmodifiableList(this.unmarkedMessageIds);
+	}
+
+	public boolean isComplete() {
+		return this.complete;
+	}
+
+	public long getTimestamp() {
+		return this.timestamp;
+	}
+
+	public int getLastReleasedMessageSequenceNumber() {
+		return this.lastReleasedMessageSequenceNumber;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  * 
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
@@ -12,6 +12,8 @@
  */
 package org.springframework.integration.store;
 
+import java.util.Iterator;
+
 import org.springframework.integration.Message;
 import org.springframework.jmx.export.annotation.ManagedAttribute;
 
@@ -19,6 +21,7 @@ import org.springframework.jmx.export.annotation.ManagedAttribute;
  * Interface for storage operations on groups of messages linked by a group id.
  * 
  * @author Dave Syer
+ * @author Oleg Zhurakousky
  * 
  * @since 2.0
  * 
@@ -121,4 +124,22 @@ public interface MessageGroupStore {
 	 * @see #registerMessageGroupExpiryCallback(MessageGroupCallback)
 	 */
 	int expireMessageGroups(long timeout);
+	
+	/**
+	 * Allows you to set the sequence number of the last released Message. Used for Resequencing use cases
+	 * @param sequenceNumber
+	 */
+	void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber);
+	
+	/**
+	 * Returns the iterator of currently accumulated {@link MessageGroup}s
+	 */
+	Iterator<MessageGroup> iterator();
+	
+	/**
+	 * Completes this MessageGroup. Completion of the MessageGroup generally means 
+	 * that this group should not be allowing any more mutating operation to be performed on it. 
+	 * For example any attempt to add/remove new Message form the group should not be allowed.
+	 */
+	void completeGroup(Object groupId);
 }

spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  * 
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
@@ -13,13 +13,13 @@
 
 package org.springframework.integration.store;
 
-import org.springframework.integration.Message;
-
 import java.util.Collection;
 import java.util.Collections;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 
+import org.springframework.integration.Message;
+
 /**
  * Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and group id. The
  * group will grow during its lifetime, when messages are <code>add</code>ed to it. This MessageGroup is thread safe.
@@ -41,22 +41,27 @@ public class SimpleMessageGroup implements MessageGroup {
 
 	// @GuardedBy(lock)
 	public final BlockingQueue<Message<?>> unmarked = new LinkedBlockingQueue<Message<?>>();
+	
+	private volatile int lastReleasedMessageSequence;
 
 	private final long timestamp;
+	
+	private volatile boolean complete;
 
 	public SimpleMessageGroup(Object groupId) {
 		this(Collections.<Message<?>> emptyList(), Collections.<Message<?>> emptyList(), groupId, System
-				.currentTimeMillis());
+				.currentTimeMillis(), false);
 	}
 
 	public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Object groupId) {
-		this(unmarked, Collections.<Message<?>> emptyList(), groupId, System.currentTimeMillis());
+		this(unmarked, Collections.<Message<?>> emptyList(), groupId, System.currentTimeMillis(), false);
 	}
 
 	public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Collection<? extends Message<?>> marked,
-			Object groupId, long timestamp) {
+			Object groupId, long timestamp, boolean complete) {
 		this.groupId = groupId;
 		this.timestamp = timestamp;
+		this.complete = complete;
 		synchronized (lock) {
 			for (Message<?> message : unmarked) {
 				addUnmarked(message);
@@ -69,6 +74,7 @@ public class SimpleMessageGroup implements MessageGroup {
 
 	public SimpleMessageGroup(MessageGroup template) {
 		this.groupId = template.getGroupId();
+		this.complete = template.isComplete();
 		synchronized (lock) {
 			// Explicit iteration to work around bug in JDK (before 1.6.0_20
 			for (Message<?> message : template.getMarked()) {
@@ -84,6 +90,8 @@ public class SimpleMessageGroup implements MessageGroup {
 		}
 		this.timestamp = template.getTimestamp();
 	}
+	
+	
 
 	public long getTimestamp() {
 		return timestamp;
@@ -103,6 +111,10 @@ public class SimpleMessageGroup implements MessageGroup {
 			unmarked.remove(message);
 		}
 	}
+	
+	public int getLastReleasedMessageSequenceNumber() {
+		return lastReleasedMessageSequence;
+	}
 
 	private boolean addUnmarked(Message<?> message) {
 		if (isMember(message)) {
@@ -127,6 +139,10 @@ public class SimpleMessageGroup implements MessageGroup {
 			return Collections.unmodifiableCollection(unmarked);
 		}
 	}
+	
+	public void setLastReleasedMessageSequenceNumber(int sequenceNumber){
+		this.lastReleasedMessageSequence = sequenceNumber;
+	}
 
 	public Collection<Message<?>> getMarked() {
 		synchronized (lock) {
@@ -139,14 +155,13 @@ public class SimpleMessageGroup implements MessageGroup {
 	}
 
 	public boolean isComplete() {
-		if (size() == 0) {
-			return true;
-		}
-		int sequenceSize = getSequenceSize();
-		// If there is no sequence then it must be incomplete....
-		return sequenceSize > 0 && sequenceSize == size();
+		return this.complete;
 	}
-
+	
+	public void complete() {
+		this.complete = true;
+	}
+	
 	public int getSequenceSize() {
 		if (size() == 0) {
 			return 0;
@@ -183,6 +198,11 @@ public class SimpleMessageGroup implements MessageGroup {
 		}
 		return one;
 	}
+	
+	public void clear(){
+		this.marked.clear();
+		this.unmarked.clear();
+	}
 
 	/**
 	 * This method determines whether messages have been added to this group that supersede the given message based on

spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java

@@ -148,10 +148,14 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
 		return group;
 	}
 
-	@Override
 	public Iterator<MessageGroup> iterator() {
 		return new HashSet<MessageGroup>(groupIdToMessageGroup.values()).iterator();
 	}
+	
+	public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
+		SimpleMessageGroup group = getMessageGroupInternal(groupId);
+		group.setLastReleasedMessageSequenceNumber(sequenceNumber);
+	}
 
 	private SimpleMessageGroup getMessageGroupInternal(Object groupId) {
 		if (!groupIdToMessageGroup.containsKey(groupId)) {
@@ -160,4 +164,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
 		return groupIdToMessageGroup.get(groupId);
 	}
 
+	public void completeGroup(Object groupId) {
+		SimpleMessageGroup group = getMessageGroupInternal(groupId);
+		group.complete();
+	}
+
 }

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd

@@ -2692,6 +2692,33 @@ endpoint itself is a Polling Consumer for a channel with a queue.
 						</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
+				<xsd:attribute name="expire-groups-upon-completion" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+							Boolean flag specifying if MessageGroup should be removed once completed. Useful for 
+							handling late arrival use cases where messages arriving with the correlationKey that 
+							is the same as the completed MessageGroup will be discarded. Default is 'false'
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="release-strategy-method" type="xsd:string">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation>
+								<tool:expected-method type-ref="@release-strategy" />
+							</tool:annotation>
+						</xsd:appinfo>
+						<xsd:documentation>
+							A method defined on the bean referenced by release-strategy, that implements the completion
+							decision algorithm.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>		
+				<xsd:attribute name="release-strategy-expression" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>A SpEL expression to apply to the message group (e.g, payload.size() > 6)</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>
@@ -2713,22 +2740,9 @@ endpoint itself is a Polling Consumer for a channel with a queue.
 						</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
-				<xsd:attribute name="release-strategy-method" type="xsd:string">
+				<xsd:attribute name="keep-released-messages" type="xsd:string">
 					<xsd:annotation>
-						<xsd:appinfo>
-							<tool:annotation>
-								<tool:expected-method type-ref="@release-strategy" />
-							</tool:annotation>
-						</xsd:appinfo>
-						<xsd:documentation>
-							A method defined on the bean referenced by release-strategy, that implements the completion
-							decision algorithm.
-						</xsd:documentation>
-					</xsd:annotation>
-				</xsd:attribute>
-				<xsd:attribute name="release-strategy-expression" type="xsd:string">
-					<xsd:annotation>
-						<xsd:documentation>A SpEL expression to apply to the message group (e.g, payload.size() > 6)</xsd:documentation>
+						<xsd:documentation>Will store messages after their release. Mainly used for monitoring purposes. Default is 'true'</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
 				<xsd:attribute name="correlation-strategy" type="xsd:string">

spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java

@@ -42,14 +42,14 @@ import org.springframework.integration.support.MessageBuilder;
  */
 public class AggregatorTests {
 
-	private CorrelatingMessageHandler aggregator;
+	private AggregatingMessageHandler aggregator;
 
 	private SimpleMessageStore store = new SimpleMessageStore(50);
 
 
 	@Before
 	public void configureAggregator() {
-		this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), store);
+		this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), store);
 	}
 
 
@@ -211,7 +211,7 @@ public class AggregatorTests {
 
 	@Test
 	public void testNullReturningAggregator() throws InterruptedException {
-		this.aggregator = new CorrelatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore(50));
+		this.aggregator = new AggregatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore(50));
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
 		Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);

spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java

@@ -16,19 +16,12 @@
 
 package org.springframework.integration.aggregator;
 
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertThat;
-
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
-
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.Message;
@@ -42,6 +35,13 @@ import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 
+import static org.hamcrest.CoreMatchers.is;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+
 /**
  * @author Mark Fisher
  * @author Marius Bogoevici
@@ -51,7 +51,7 @@ public class ConcurrentAggregatorTests {
 
 	private TaskExecutor taskExecutor;
 
-	private CorrelatingMessageHandler aggregator;
+	private AggregatingMessageHandler aggregator;
 
 	private MessageGroupStore store = new SimpleMessageStore();
 
@@ -59,7 +59,7 @@ public class ConcurrentAggregatorTests {
 	@Before
 	public void configureAggregator() {
 		this.taskExecutor = new SimpleAsyncTaskExecutor();
-		this.aggregator = new CorrelatingMessageHandler(new MultiplyingProcessor(), store);
+		this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), store);
 	}
 
 
@@ -274,7 +274,7 @@ public class ConcurrentAggregatorTests {
 
 	@Test
 	public void testNullReturningAggregator() throws InterruptedException {
-		this.aggregator = new CorrelatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore(
+		this.aggregator = new AggregatingMessageHandler(new NullReturningMessageProcessor(), new SimpleMessageStore(
 						50));
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);

spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java

@@ -16,21 +16,20 @@
 
 package org.springframework.integration.aggregator;
 
-import static org.mockito.Mockito.isA;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
 import org.junit.Before;
 import org.junit.Test;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
 public class CorrelatingMessageHandlerIntegrationTests {
 
 	private MessageGroupStore store = new SimpleMessageStore(100);
@@ -39,7 +38,7 @@ public class CorrelatingMessageHandlerIntegrationTests {
 
 	private MessageGroupProcessor processor = new PassThroughMessageGroupProcessor();
 
-	private CorrelatingMessageHandler defaultHandler = new CorrelatingMessageHandler(processor, store);
+	private AggregatingMessageHandler defaultHandler = new AggregatingMessageHandler(processor, store);
 
 	@Before
 	public void setupHandler() {

spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java

@@ -16,13 +16,6 @@
 
 package org.springframework.integration.aggregator;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-import static org.mockito.Matchers.isA;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -35,7 +28,6 @@ import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.internal.stubbing.answers.ThrowsException;
 import org.mockito.runners.MockitoJUnitRunner;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessageHandlingException;
@@ -45,6 +37,14 @@ import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
 /**
  * @author Iwein Fuld
  * @author Dave Syer
@@ -52,7 +52,7 @@ import org.springframework.test.util.ReflectionTestUtils;
 @RunWith(MockitoJUnitRunner.class)
 public class CorrelatingMessageHandlerTests {
 
-	private CorrelatingMessageHandler handler;
+	private AggregatingMessageHandler handler;
 
 	@Mock
 	private CorrelationStrategy correlationStrategy;
@@ -70,7 +70,7 @@ public class CorrelatingMessageHandlerTests {
 
 	@Before
 	public void initializeSubject() {
-		handler = new CorrelatingMessageHandler(processor, store, correlationStrategy, ReleaseStrategy);
+		handler = new AggregatingMessageHandler(processor, store, correlationStrategy, ReleaseStrategy);
 		handler.setOutputChannel(outputChannel);
 	}
 
@@ -94,7 +94,7 @@ public class CorrelatingMessageHandlerTests {
 		verify(processor).processMessageGroup(isA(SimpleMessageGroup.class));
 	}
 
-	private void verifyLocks(CorrelatingMessageHandler handler, int lockCount) {
+	private void verifyLocks(AggregatingMessageHandler handler, int lockCount) {
 		assertEquals(lockCount, ((Map<?, ?>) ReflectionTestUtils.getField(handler, "locks")).size());
 	}
 
@@ -110,6 +110,8 @@ public class CorrelatingMessageHandlerTests {
 
 		when(correlationStrategy.getCorrelationKey(isA(Message.class))).thenReturn(correlationKey);
 
+		handler.setExpireGroupsUponCompletion(true);
+		
 		handler.handleMessage(message1);
 
 		try {

spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java

@@ -479,7 +479,7 @@ public class MethodInvokingMessageGroupProcessorTests {
 		proxyFactory.setProxyTargetClass(false);
 		testBean = (GreetingService) proxyFactory.getProxy();
 		MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(testBean);
-		CorrelatingMessageHandler handler = new CorrelatingMessageHandler(aggregator);
+		AggregatingMessageHandler handler = new AggregatingMessageHandler(aggregator);
 		handler.setReleaseStrategy(new MessageCountReleaseStrategy());
 		handler.setOutputChannel(output);
 		EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler);
@@ -498,7 +498,7 @@ public class MethodInvokingMessageGroupProcessorTests {
 		proxyFactory.setProxyTargetClass(true);
 		testBean = (GreetingService) proxyFactory.getProxy();
 		MethodInvokingMessageGroupProcessor aggregator = new MethodInvokingMessageGroupProcessor(testBean);
-		CorrelatingMessageHandler handler = new CorrelatingMessageHandler(aggregator);
+		AggregatingMessageHandler handler = new AggregatingMessageHandler(aggregator);
 		handler.setReleaseStrategy(new MessageCountReleaseStrategy());
 		handler.setOutputChannel(output);
 		EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler);

spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2009 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,11 @@
 
 package org.springframework.integration.aggregator;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.integration.Message;
@@ -25,23 +30,23 @@ import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Comparator;
+import static org.hamcrest.Matchers.is;
 
-import static org.junit.Assert.*;
-import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 
 /**
  * @author Marius Bogoevici
  * @author Alex Peters
  * @author Dave Syer
  * @author Iwein Fuld
+ * @author Oleg Zhurakousky
  */
 public class ResequencerTests {
 
-	private CorrelatingMessageHandler resequencer;
+	private ResequencingMessageHandler resequencer;
 
 	private ResequencingMessageGroupProcessor processor = new ResequencingMessageGroupProcessor();
 
@@ -49,7 +54,7 @@ public class ResequencerTests {
 
 	@Before
 	public void configureResequencer() {
-		this.resequencer = new CorrelatingMessageHandler(processor, store, null, null);
+		this.resequencer = new ResequencingMessageHandler(processor, store, null, null);
 	}
 
 	@Test
@@ -71,6 +76,59 @@ public class ResequencerTests {
 		assertNotNull(reply3);
 		assertThat( reply3.getHeaders().getSequenceNumber(), is(3));
 	}
+	
+	@Test
+	public void testBasicResequencingA() throws InterruptedException {
+		SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();
+		releaseStrategy.setReleasePartialSequences(true);
+		this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy);
+		
+		QueueChannel replyChannel = new QueueChannel();
+		Message<?> message1 = createMessage("123", "ABC", 3, 1, replyChannel);
+		Message<?> message3 = createMessage("789", "ABC", 3, 3, replyChannel);
+		
+		this.resequencer.handleMessage(message3);
+		assertNull(replyChannel.receive(0));
+		this.resequencer.handleMessage(message1);
+		assertNotNull(replyChannel.receive(0));
+		assertNull(replyChannel.receive(0));
+	}
+	
+	@Test
+	public void testBasicUnboundedResequencing() throws InterruptedException {
+		SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();
+		releaseStrategy.setReleasePartialSequences(true);
+		this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy);
+		QueueChannel replyChannel = new QueueChannel();
+		this.resequencer.setCorrelationStrategy(new CorrelationStrategy() {	
+			public Object getCorrelationKey(Message<?> message) {
+				return "A";
+			}
+		});
+		//Message<?> message0 = MessageBuilder.withPayload("0").setSequenceNumber(0).build();
+		Message<?> message1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setReplyChannel(replyChannel).build();
+		Message<?> message2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setReplyChannel(replyChannel).build();
+		Message<?> message3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setReplyChannel(replyChannel).build();
+		Message<?> message4 = MessageBuilder.withPayload("4").setSequenceNumber(4).setReplyChannel(replyChannel).build();
+		Message<?> message5 = MessageBuilder.withPayload("5").setSequenceNumber(5).setReplyChannel(replyChannel).build();
+		
+		this.resequencer.handleMessage(message3);
+		assertNull(replyChannel.receive(0));
+		this.resequencer.handleMessage(message1);
+		assertNotNull(replyChannel.receive(0));
+		
+		this.resequencer.handleMessage(message2);
+
+		assertNotNull(replyChannel.receive(0));
+		assertNotNull(replyChannel.receive(0));
+		assertNull(replyChannel.receive(0));
+		
+		this.resequencer.handleMessage(message5);
+		assertNull(replyChannel.receive(0));
+		this.resequencer.handleMessage(message4);
+		assertNotNull(replyChannel.receive(0));
+	}
+
 
 	@Test
 	public void testBasicResequencingWithCustomComparator() throws InterruptedException {

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorExpressionIntegrationTests-context.xml

@@ -5,7 +5,7 @@
 	xmlns:task="http://www.springframework.org/schema/task"
 	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
-		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">
 
 	<channel id="input"/>
 

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml

@@ -5,7 +5,7 @@
 	xmlns:task="http://www.springframework.org/schema/task"
 	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
-		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">
 
 	<channel id="input">
 		<queue capacity="5" />
@@ -20,8 +20,19 @@
 	<channel id="output">
 		<queue capacity="5" />
 	</channel>
+	
+	<channel id="discard">
+		<queue capacity="5" />
+	</channel>
 
 	<beans:bean id="summer"
 		class="org.springframework.integration.aggregator.integration.AggregatorIntegrationTests$SummingAggregator" />
+		
+	<aggregator id="expiringAggregator" input-channel="expiringAggregatorInput" output-channel="output" 
+				expire-groups-upon-completion="true" discard-channel="discard"/>
+				
+	<aggregator id="nonExpiringAggregator" input-channel="nonExpiringAggregatorInput" output-channel="output" 
+				expire-groups-upon-completion="false" discard-channel="discard"/>
+		
 
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java

@@ -17,6 +17,8 @@
 package org.springframework.integration.aggregator.integration;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 
 import java.util.HashMap;
 import java.util.List;
@@ -45,10 +47,22 @@ public class AggregatorIntegrationTests {
 	@Autowired
 	@Qualifier("input")
 	private MessageChannel input;
+	
+	@Autowired
+	@Qualifier("expiringAggregatorInput")
+	private MessageChannel expiringAggregatorInput;
+	
+	@Autowired
+	@Qualifier("nonExpiringAggregatorInput")
+	private MessageChannel nonExpiringAggregatorInput;
 
 	@Autowired
 	@Qualifier("output")
 	private PollableChannel output;
+	
+	@Autowired
+	@Qualifier("discard")
+	private PollableChannel discard;
 
 	@Test//(timeout=5000)
 	public void testVanillaAggregation() throws Exception {
@@ -58,6 +72,49 @@ public class AggregatorIntegrationTests {
 		}
 		assertEquals(0 + 1 + 2 + 3 + 4, output.receive().getPayload());
 	}
+	
+	@Test
+	public void testNonExpiringAggregator() throws Exception {
+		for (int i = 0; i < 5; i++) {
+			Map<String, Object> headers = stubHeaders(i, 5, 1);
+			nonExpiringAggregatorInput.send(new GenericMessage<Integer>(i, headers));
+		}
+		assertNotNull(output.receive(0));
+		
+		assertNull(discard.receive(0));
+		
+		for (int i = 5; i < 10; i++) {
+			Map<String, Object> headers = stubHeaders(i, 5, 1);
+			nonExpiringAggregatorInput.send(new GenericMessage<Integer>(i, headers));
+		}
+		assertNull(output.receive(0));
+		
+		assertNotNull(discard.receive(0));
+		assertNotNull(discard.receive(0));
+		assertNotNull(discard.receive(0));
+		assertNotNull(discard.receive(0));
+		assertNotNull(discard.receive(0));
+	}
+	
+	@Test
+	public void testExpiringAggregator() throws Exception {
+		for (int i = 0; i < 5; i++) {
+			Map<String, Object> headers = stubHeaders(i, 5, 1);
+			expiringAggregatorInput.send(new GenericMessage<Integer>(i, headers));
+		}
+		assertNotNull(output.receive(0));
+		
+		assertNull(discard.receive(0));
+		
+		for (int i = 5; i < 10; i++) {
+			Map<String, Object> headers = stubHeaders(i, 5, 1);
+			expiringAggregatorInput.send(new GenericMessage<Integer>(i, headers));
+		}
+		assertNotNull(output.receive(0));
+		
+		assertNull(discard.receive(0));
+
+	}
 
 	// configured in context associated with this test
 	public static class SummingAggregator {

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorSupportedUseCasesTests.java

@@ -0,0 +1,168 @@
+/*
+ * Copyright 2002-2011 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package org.springframework.integration.aggregator.integration;
+
+import java.util.List;
+
+import org.junit.Test;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
+import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
+import org.springframework.integration.aggregator.ReleaseStrategy;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.store.SimpleMessageStore;
+import org.springframework.integration.support.MessageBuilder;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+/**
+ * @author Oleg Zhurakousky
+ *
+ */
+public class AggregatorSupportedUseCasesTests {
+	
+	private MessageGroupStore store = new SimpleMessageStore(100);
+	
+	private DefaultAggregatingMessageGroupProcessor processor = new DefaultAggregatingMessageGroupProcessor();
+	
+	private AggregatingMessageHandler defaultHandler = new AggregatingMessageHandler(processor, store);
+
+	@Test
+	public void waitForAllDefaultReleaseStrategyWithLateArrivals(){
+		QueueChannel outputChannel = new QueueChannel();
+		QueueChannel discardChannel = new QueueChannel();
+		defaultHandler.setOutputChannel(outputChannel);
+		defaultHandler.setDiscardChannel(discardChannel);
+		defaultHandler.setKeepReleasedMessages(false);
+		
+		for (int i = 0; i < 5; i++) {
+			defaultHandler.handleMessage(MessageBuilder.withPayload(i).setSequenceSize(5).setCorrelationId("A").setSequenceNumber(i).build());
+		}
+		assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
+		assertNull(discardChannel.receive(0));
+		assertEquals(0, store.getMessageGroup("A").getUnmarked().size());
+		assertEquals(0, store.getMessageGroup("A").getMarked().size());
+		
+		// send another message with the same correlation id and see it in the discard channel
+		defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setSequenceSize(5).setCorrelationId("A").setSequenceNumber(3).build());
+		assertNotNull(discardChannel.receive(0));
+		
+		// set 'expireGroupsUponCompletion' to 'true' and the messages should start accumulating again
+		defaultHandler.setExpireGroupsUponCompletion(true);
+		defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setSequenceSize(5).setCorrelationId("A").setSequenceNumber(3).build());
+		assertNull(discardChannel.receive(0));
+		assertEquals(1, store.getMessageGroup("A").getUnmarked().size());
+	}
+	
+	@Test
+	public void waitForAllCustomReleaseStrategyWithLateArrivals(){
+		QueueChannel outputChannel = new QueueChannel();
+		QueueChannel discardChannel = new QueueChannel();
+		defaultHandler.setOutputChannel(outputChannel);
+		defaultHandler.setDiscardChannel(discardChannel);
+		defaultHandler.setReleaseStrategy(new SampleSizeReleaseStrategy());
+		defaultHandler.setKeepReleasedMessages(false);
+		
+		for (int i = 0; i < 5; i++) {
+			defaultHandler.handleMessage(MessageBuilder.withPayload(i).setCorrelationId("A").build());
+		}
+		assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
+		assertNull(discardChannel.receive(0));
+		assertEquals(0, store.getMessageGroup("A").getUnmarked().size());
+		assertEquals(0, store.getMessageGroup("A").getMarked().size());
+		
+		// send another message with the same correlation id and see it in the discard channel
+		defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("A").build());
+		assertNotNull(discardChannel.receive(0));
+		
+		// set 'expireGroupsUponCompletion' to 'true' and the messages should start accumulating again
+		defaultHandler.setExpireGroupsUponCompletion(true);
+		defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("A").build());
+		assertNull(discardChannel.receive(0));
+		assertEquals(1, store.getMessageGroup("A").getUnmarked().size());
+	}
+	
+	@Test
+	public void firstBest(){
+		QueueChannel outputChannel = new QueueChannel();
+		QueueChannel discardChannel = new QueueChannel();
+		defaultHandler.setOutputChannel(outputChannel);
+		defaultHandler.setDiscardChannel(discardChannel);
+		defaultHandler.setReleaseStrategy(new FirstBestReleaseStrategy());
+		
+		for (int i = 0; i < 5; i++) {
+			defaultHandler.handleMessage(MessageBuilder.withPayload(i).setCorrelationId("A").build());
+		}
+		assertEquals(1, ((List<?>)outputChannel.receive(0).getPayload()).size());
+		assertNotNull(discardChannel.receive(0));
+		assertNotNull(discardChannel.receive(0));
+		assertNotNull(discardChannel.receive(0));
+		assertNotNull(discardChannel.receive(0));
+	}
+	
+	@Test
+	public void batchingWithoutLeftovers(){
+		QueueChannel outputChannel = new QueueChannel();
+		QueueChannel discardChannel = new QueueChannel();
+		defaultHandler.setOutputChannel(outputChannel);
+		defaultHandler.setDiscardChannel(discardChannel);
+		defaultHandler.setReleaseStrategy(new SampleSizeReleaseStrategy());
+		defaultHandler.setExpireGroupsUponCompletion(true);
+		
+		for (int i = 0; i < 10; i++) {
+			defaultHandler.handleMessage(MessageBuilder.withPayload(i).setCorrelationId("A").build());
+		}
+		assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
+		assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
+		assertNull(discardChannel.receive(0));
+	}
+	
+	@Test
+	public void batchingWithLeftovers(){
+		QueueChannel outputChannel = new QueueChannel();
+		QueueChannel discardChannel = new QueueChannel();
+		defaultHandler.setOutputChannel(outputChannel);
+		defaultHandler.setDiscardChannel(discardChannel);
+		defaultHandler.setReleaseStrategy(new SampleSizeReleaseStrategy());
+		defaultHandler.setExpireGroupsUponCompletion(true);
+		
+		for (int i = 0; i < 12; i++) {
+			defaultHandler.handleMessage(MessageBuilder.withPayload(i).setCorrelationId("A").build());
+		}
+		assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
+		assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
+		assertNull(discardChannel.receive(0));
+		assertEquals(2, store.getMessageGroup("A").getUnmarked().size());
+	}
+	
+	private class SampleSizeReleaseStrategy implements ReleaseStrategy {
+
+		public boolean canRelease(MessageGroup group) {
+			return group.getUnmarked().size() == 5;
+		}
+		
+	}
+	
+	private class FirstBestReleaseStrategy implements ReleaseStrategy {
+
+		public boolean canRelease(MessageGroup group) {
+			return true;
+		}
+		
+	}
+	
+}

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests-context.xml

@@ -5,7 +5,7 @@
 	xsi:schemaLocation="http://www.springframework.org/schema/beans
 			http://www.springframework.org/schema/beans/spring-beans.xsd
 			http://www.springframework.org/schema/integration
-			http://www.springframework.org/schema/integration/spring-integration.xsd">
+			http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">
 
 	<channel id="output">
 		<queue/>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests-context.xml

@@ -5,7 +5,7 @@
 	xmlns:task="http://www.springframework.org/schema/task"
 	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
-		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">
 
 	<annotation-config />
 

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/MethodInvokingAggregatorReturningMessageTests-context.xml

@@ -5,7 +5,7 @@
 	xsi:schemaLocation="http://www.springframework.org/schema/beans
 			http://www.springframework.org/schema/beans/spring-beans.xsd
 			http://www.springframework.org/schema/integration
-			http://www.springframework.org/schema/integration/spring-integration.xsd">
+			http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">
 
 	<channel id="pojoOutput">
 		<queue/>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerExpressionIntegrationTests-context.xml

@@ -1,16 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:beans="http://www.springframework.org/schema/beans" xmlns:task="http://www.springframework.org/schema/task"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
-		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
-
-	<channel id="input" />
-
-	<resequencer correlation-strategy-expression="headers['foo']" release-strategy-expression="size()>2" input-channel="input" output-channel="output" />
-
-	<channel id="output">
-		<queue capacity="5" />
-	</channel>
-
-</beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerExpressionIntegrationTests.java

@@ -1,78 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.integration.aggregator.integration;
-
-import static org.junit.Assert.assertEquals;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.integration.Message;
-import org.springframework.integration.MessageChannel;
-import org.springframework.integration.MessageHeaders;
-import org.springframework.integration.core.PollableChannel;
-import org.springframework.integration.message.GenericMessage;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
-
-/**
- * @author Iwein Fuld
- * @author Alex Peters
- * @author Oleg Zhurakousky
- */
-@RunWith(SpringJUnit4ClassRunner.class)
-@ContextConfiguration
-public class ResequencerExpressionIntegrationTests {
-
-	@Autowired
-	@Qualifier("input")
-	private MessageChannel input;
-
-	@Autowired
-	@Qualifier("output")
-	private PollableChannel output;
-
-	@Test//(timeout=5000)
-	public void testVanillaAggregation() throws Exception {
-		List<Message<?>> messages = new ArrayList<Message<?>>();
-		for (int i = 0; i < 5; i++) {
-			Map<String, Object> headers = stubHeaders(i, 5, 1);
-			messages.add(new GenericMessage<Integer>(i, headers));
-		}
-		input.send(messages.get(2));			
-		input.send(messages.get(1));			
-		input.send(messages.get(0));			
-		assertEquals(0, output.receive().getPayload());
-		assertEquals(1, output.receive().getPayload());
-		assertEquals(2, output.receive().getPayload());
-	}
-
-	private Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correllationId) {
-		Map<String, Object> headers = new HashMap<String, Object>();
-		headers.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber);
-		headers.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize);
-		headers.put("foo", correllationId);
-		return headers;
-	}
-
-}

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTest-context.xml

@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">
+
+
+	<int:resequencer id="resequencerLight" input-channel="resequencerLightInput" output-channel="outputChannel" release-partial-sequences="true"
+		keep-released-messages="false"/>
+	
+	<int:channel id="outputChannel">
+		<int:queue/>
+	</int:channel>
+	
+	<int:resequencer id="resequencerDeep" input-channel="resequencerDeepInput" output-channel="outputChannel" release-partial-sequences="true"/>
+
+</beans>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTest.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2002-2011 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.aggregator.integration;
+
+import org.junit.Test;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.aggregator.ResequencingMessageHandler;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+/**
+ * @author Oleg Zhurakousky
+ */
+public class ResequencerIntegrationTest {
+
+	@Test 
+	public void validateUnboundedResequencerLight(){
+		ApplicationContext context = new ClassPathXmlApplicationContext("ResequencerIntegrationTest-context.xml",  ResequencerIntegrationTest.class);
+		MessageChannel inputChannel = context .getBean("resequencerLightInput", MessageChannel.class);
+		QueueChannel outputChannel = context .getBean("outputChannel", QueueChannel.class);
+		EventDrivenConsumer edc = context.getBean("resequencerLight", EventDrivenConsumer.class);
+		ResequencingMessageHandler handler = TestUtils.getPropertyValue(edc, "handler", ResequencingMessageHandler.class);
+		MessageGroupStore store = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
+		
+		Message<?> message1 = MessageBuilder.withPayload("1").setCorrelationId("A").setSequenceNumber(1).build();
+		Message<?> message2 = MessageBuilder.withPayload("2").setCorrelationId("A").setSequenceNumber(2).build();
+		Message<?> message3 = MessageBuilder.withPayload("3").setCorrelationId("A").setSequenceNumber(3).build();
+		Message<?> message4 = MessageBuilder.withPayload("4").setCorrelationId("A").setSequenceNumber(4).build();
+		Message<?> message5 = MessageBuilder.withPayload("5").setCorrelationId("A").setSequenceNumber(5).build();
+		Message<?> message6 = MessageBuilder.withPayload("6").setCorrelationId("A").setSequenceNumber(6).build();
+		
+		inputChannel.send(message3);
+		assertNull(outputChannel.receive(0));
+		
+		inputChannel.send(message1);
+		message1 = outputChannel.receive(0);
+		assertNotNull(message1);
+		assertEquals((Integer)1, message1.getHeaders().getSequenceNumber());
+		
+		inputChannel.send(message2);
+		message2 = outputChannel.receive(0);
+		message3 = outputChannel.receive(0);
+		assertNotNull(message2);	
+		assertNotNull(message3);
+		assertEquals((Integer)2, message2.getHeaders().getSequenceNumber());
+		assertEquals((Integer)3, message3.getHeaders().getSequenceNumber());
+		
+		inputChannel.send(message5);
+		assertNull(outputChannel.receive(0));
+		
+		inputChannel.send(message6);
+		assertNull(outputChannel.receive(0));
+		
+		inputChannel.send(message4);
+		message4 = outputChannel.receive(0);
+		message5 = outputChannel.receive(0);
+		message6 = outputChannel.receive(0);
+		assertNotNull(message4);	
+		assertNotNull(message5);
+		assertNotNull(message6);
+		assertEquals((Integer)4, message4.getHeaders().getSequenceNumber());
+		assertEquals((Integer)5, message5.getHeaders().getSequenceNumber());
+		assertEquals((Integer)6, message6.getHeaders().getSequenceNumber());
+		
+		
+		assertEquals(0, store.getMessageGroup("A").getUnmarked().size());
+		assertEquals(0, store.getMessageGroup("A").getMarked().size());
+	}
+	
+	@Test 
+	public void validateUnboundedResequencerDeep(){
+		ApplicationContext context = new ClassPathXmlApplicationContext("ResequencerIntegrationTest-context.xml",  ResequencerIntegrationTest.class);
+		MessageChannel inputChannel = context .getBean("resequencerDeepInput", MessageChannel.class);
+		QueueChannel outputChannel = context .getBean("outputChannel", QueueChannel.class);
+		EventDrivenConsumer edc = context.getBean("resequencerDeep", EventDrivenConsumer.class);
+		ResequencingMessageHandler handler = TestUtils.getPropertyValue(edc, "handler", ResequencingMessageHandler.class);
+		MessageGroupStore store = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
+		
+		Message<?> message1 = MessageBuilder.withPayload("1").setCorrelationId("A").setSequenceNumber(1).build();
+		Message<?> message2 = MessageBuilder.withPayload("2").setCorrelationId("A").setSequenceNumber(2).build();
+		Message<?> message3 = MessageBuilder.withPayload("3").setCorrelationId("A").setSequenceNumber(3).build();
+		
+		inputChannel.send(message3);
+		assertNull(outputChannel.receive(0));
+		inputChannel.send(message1);
+		assertNotNull(outputChannel.receive(0));
+		inputChannel.send(message2);
+		assertNotNull(outputChannel.receive(0));
+		assertNotNull(outputChannel.receive(0));
+		assertEquals(0, store.getMessageGroup("A").getUnmarked().size());
+		assertEquals(3, store.getMessageGroup("A").getMarked().size());
+	}
+}

spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java

@@ -16,11 +16,6 @@
 
 package org.springframework.integration.config;
 
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -38,7 +33,7 @@ import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.MessageRejectedException;
-import org.springframework.integration.aggregator.CorrelatingMessageHandler;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
 import org.springframework.integration.aggregator.CorrelationStrategy;
 import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
 import org.springframework.integration.aggregator.ReleaseStrategy;
@@ -49,6 +44,12 @@ import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 
+import static org.hamcrest.CoreMatchers.is;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
 /**
  * @author Marius Bogoevici
  * @author Mark Fisher
@@ -111,7 +112,7 @@ public class AggregatorParserTests {
 		MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
 		MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel");
 		Object consumer = new DirectFieldAccessor(endpoint).getPropertyValue("handler");
-		assertThat(consumer, is(CorrelatingMessageHandler.class));
+		assertThat(consumer, is(AggregatingMessageHandler.class));
 		DirectFieldAccessor accessor = new DirectFieldAccessor(consumer);
 		Map<?, ?> map = (Map<?, ?>) new DirectFieldAccessor(new DirectFieldAccessor(new DirectFieldAccessor(accessor
 				.getPropertyValue("outputProcessor")).getPropertyValue("processor")).getPropertyValue("delegate"))

spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  * 
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
@@ -13,21 +13,27 @@
 
 package org.springframework.integration.config;
 
+import java.util.Comparator;
+
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
-import org.springframework.integration.aggregator.*;
+import org.springframework.integration.aggregator.CorrelationStrategy;
+import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
+import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
+import org.springframework.integration.aggregator.ResequencingMessageHandler;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 
-import java.util.Comparator;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 
-import static org.junit.Assert.*;
 import static org.springframework.integration.test.util.TestUtils.getPropertyValue;
 
 /**
@@ -47,8 +53,8 @@ public class ResequencerParserTests {
 	@Test
 	public void testDefaultResequencerProperties() {
 		EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("defaultResequencer");
-		CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
-				CorrelatingMessageHandler.class);
+		ResequencingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
+				ResequencingMessageHandler.class);
 		assertNull(getPropertyValue(resequencer, "outputChannel"));
 		assertTrue(getPropertyValue(resequencer, "discardChannel") instanceof NullChannel);
 		assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", 1000l, getPropertyValue(
@@ -65,8 +71,8 @@ public class ResequencerParserTests {
 		EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("completelyDefinedResequencer");
 		MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
 		MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel");
-		CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
-				CorrelatingMessageHandler.class);
+		ResequencingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
+				ResequencingMessageHandler.class);
 		assertEquals("The ResequencerEndpoint is not injected with the appropriate output channel", outputChannel,
 				getPropertyValue(resequencer, "outputChannel"));
 		assertEquals("The ResequencerEndpoint is not injected with the appropriate discard channel", discardChannel,
@@ -84,8 +90,8 @@ public class ResequencerParserTests {
 	public void testCorrelationStrategyRefOnly() throws Exception {
 		EventDrivenConsumer endpoint = (EventDrivenConsumer) context
 				.getBean("resequencerWithCorrelationStrategyRefOnly");
-		CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
-				CorrelatingMessageHandler.class);
+		ResequencingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
+				ResequencingMessageHandler.class);
 		assertEquals("The ResequencerEndpoint is not configured with the appropriate CorrelationStrategy", context
 				.getBean("testCorrelationStrategy"), getPropertyValue(resequencer, "correlationStrategy"));
 	}
@@ -93,8 +99,8 @@ public class ResequencerParserTests {
 	@Test
 	public void shouldSetReleasePartialSequencesFlag(){
 				EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("completelyDefinedResequencer");
-		CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
-				CorrelatingMessageHandler.class);
+				ResequencingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
+						ResequencingMessageHandler.class);
 		assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag",
 				true, getPropertyValue(getPropertyValue(resequencer, "releaseStrategy"), "releasePartialSequences"));
 	}
@@ -103,8 +109,8 @@ public class ResequencerParserTests {
 	public void testCorrelationStrategyRefAndMethod() throws Exception {
 		EventDrivenConsumer endpoint = (EventDrivenConsumer) context
 				.getBean("resequencerWithCorrelationStrategyRefAndMethod");
-		CorrelatingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
-				CorrelatingMessageHandler.class);
+		ResequencingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler",
+				ResequencingMessageHandler.class);
 		Object correlationStrategy = getPropertyValue(resequencer, "correlationStrategy");
 		assertEquals("The ResequencerEndpoint is not configured with a CorrelationStrategy adapter",
 				MethodInvokingCorrelationStrategy.class, correlationStrategy.getClass());
@@ -115,8 +121,8 @@ public class ResequencerParserTests {
 	@Test
 	public void testComparator() throws Exception {
 		EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithComparator");
-		CorrelatingMessageHandler handler = TestUtils.getPropertyValue(endpoint, "handler",
-				CorrelatingMessageHandler.class);
+		ResequencingMessageHandler handler = TestUtils.getPropertyValue(endpoint, "handler",
+				ResequencingMessageHandler.class);
 		ResequencingMessageGroupProcessor resequencer = TestUtils.getPropertyValue(handler, "outputProcessor",
 				ResequencingMessageGroupProcessor.class);
 		Object comparator = getPropertyValue(resequencer, "comparator");
@@ -124,16 +130,6 @@ public class ResequencerParserTests {
 				.getClass());
 	}
 
-	@Test
-	public void testReleaseStrategy() throws Exception {
-		EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("resequencerWithReleaseStrategy");
-		CorrelatingMessageHandler handler = TestUtils.getPropertyValue(endpoint, "handler",
-				CorrelatingMessageHandler.class);
-		Object releaseStrategy = getPropertyValue(handler, "releaseStrategy");
-		assertEquals("The Resequencer is not configured with an adapter", MethodInvokingReleaseStrategy.class, releaseStrategy
-				.getClass());
-	}
-
 	@SuppressWarnings("unused")
 	private static <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
 			MessageChannel outputChannel) {

spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java

@@ -16,12 +16,6 @@
 
 package org.springframework.integration.config.annotation;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
-import static org.springframework.integration.test.util.TestUtils.getPropertyValue;
-
 import java.lang.reflect.Method;
 import java.util.Map;
 
@@ -30,9 +24,9 @@ import org.junit.Test;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
-import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
-import org.springframework.integration.aggregator.CorrelatingMessageHandler;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
 import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
+import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
 import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.core.MessageHandler;
@@ -41,6 +35,13 @@ import org.springframework.integration.support.channel.BeanFactoryChannelResolve
 import org.springframework.integration.support.channel.ChannelResolver;
 import org.springframework.integration.test.util.TestUtils;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import static org.springframework.integration.test.util.TestUtils.getPropertyValue;
+
 /**
  * @author Marius Bogoevici
  * @author Mark Fisher
@@ -56,7 +57,7 @@ public class AggregatorAnnotationTests {
 		assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SequenceSizeReleaseStrategy);
 		assertNull(getPropertyValue(aggregator, "outputChannel"));
 		assertTrue(getPropertyValue(aggregator, "discardChannel") instanceof NullChannel);
-		assertEquals(CorrelatingMessageHandler.DEFAULT_SEND_TIMEOUT, getPropertyValue(aggregator,
+		assertEquals(AggregatingMessageHandler.DEFAULT_SEND_TIMEOUT, getPropertyValue(aggregator,
 				"messagingTemplate.sendTimeout"));
 		assertEquals(false, getPropertyValue(aggregator, "sendPartialResultOnExpiry"));
 	}

spring-integration-core/src/test/java/org/springframework/integration/config/resequencerParserTests.xml

@@ -5,7 +5,7 @@
 	xsi:schemaLocation="http://www.springframework.org/schema/beans
 			http://www.springframework.org/schema/beans/spring-beans.xsd
 			http://www.springframework.org/schema/integration
-			http://www.springframework.org/schema/integration/spring-integration.xsd">
+			http://www.springframework.org/schema/integration/spring-integration-2.1.xsd">
 
 	<channel id="inputChannel"/>
 
@@ -50,10 +50,10 @@
 		input-channel="inputChannel5"
 		comparator="testComparator"/>
 		
-	<resequencer id="resequencerWithReleaseStrategy"
-		input-channel="inputChannel6"
-		release-strategy="pojoReleaseStrategy"
-		release-strategy-method="checkCompletenessAsList"/>
+<!-- 	<resequencer id="resequencerWithReleaseStrategy" -->
+<!-- 		input-channel="inputChannel6" -->
+<!-- 		release-strategy="pojoReleaseStrategy" -->
+<!-- 		release-strategy-method="checkCompletenessAsList"/> -->
 
 	<beans:bean id="testComparator"
 		class="org.springframework.integration.config.ResequencerParserTests$TestComparator"/>
@@ -64,9 +64,9 @@
 	<beans:bean id="testCorrelationStrategyPojo"
 		class="org.springframework.integration.config.ResequencerParserTests$TestCorrelationStrategyPojo"/>
 
-	<beans:bean id="pojoReleaseStrategy"
-		class="org.springframework.integration.config.MaxValueReleaseStrategy">
-		<beans:constructor-arg value="10" />
-	</beans:bean>
+<!-- 	<beans:bean id="pojoReleaseStrategy" -->
+<!-- 		class="org.springframework.integration.config.MaxValueReleaseStrategy"> -->
+<!-- 		<beans:constructor-arg value="10" /> -->
+<!-- 	</beans:bean> -->
 
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/config/xml/PNamespaceTests.java

@@ -16,21 +16,20 @@
 
 package org.springframework.integration.config.xml;
 
-import static org.junit.Assert.assertEquals;
-
 import java.util.List;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.integration.aggregator.CorrelatingMessageHandler;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+import static org.junit.Assert.assertEquals;
+
 /**
  * Validates the "p:namespace" is working for inner "bean" definition within SI components.
  * 
@@ -92,7 +91,7 @@ public class PNamespaceTests {
 	@Test
 	public void testPNamespaceChain() {		
 		List<?> handlers = (List<?>) TestUtils.getPropertyValue(sampleChain, "handler.handlers");
-		CorrelatingMessageHandler handler = (CorrelatingMessageHandler) handlers.get(0);
+		AggregatingMessageHandler handler = (AggregatingMessageHandler) handlers.get(0);
 		SampleAggregator aggregator = 
 			(SampleAggregator) TestUtils.getPropertyValue(handler, "outputProcessor.processor.delegate.targetObject");
 		assertEquals("Bill", aggregator.getName());

spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java

@@ -87,7 +87,6 @@ public class MessageStoreTests {
 
 		private boolean removed = false;
 
-		@Override
 		public Iterator<MessageGroup> iterator() {
 			return Arrays.asList(testMessages).iterator();
 		}
@@ -118,6 +117,15 @@ public class MessageStoreTests {
 			}
 		}
 
+		public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
+			throw new UnsupportedOperationException();
+		}
+
+		public void completeGroup(Object groupId) {
+
+			throw new UnsupportedOperationException();
+		}
+
 	}
 
 }