Merge branch 'work'

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

@@ -27,6 +27,7 @@ import org.springframework.integration.core.MessageHeaders;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.MessageGroupCallback;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.SimpleMessageStore;
@@ -81,6 +82,11 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 		Assert.notNull(store);
 		Assert.notNull(processor);
 		this.store = store;
+		store.registerExpiryCallback(new MessageGroupCallback() {
+			public void execute(MessageGroup group) {
+				forceComplete(group);
+			}
+		});
 		this.outputProcessor = processor;
 		this.correlationStrategy = correlationStrategy == null ? new HeaderAttributeCorrelationStrategy(
 				MessageHeaders.CORRELATION_ID) : correlationStrategy;
@@ -198,12 +204,12 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 	}
 
 	// TODO: INT-958 - arrange for this to be called if user desires, e.g. periodically
-	public final boolean forceComplete(Object correlationKey) {
+	private final boolean forceComplete(MessageGroup group) {
 
+		Object correlationKey = group.getCorrelationKey();
 		Object lock = getLock(correlationKey);
 		synchronized (lock) {
 
-			MessageGroup group = store.getMessageGroup(correlationKey);
 			if (group.size() > 0) {
 				// last chance for normal completion
 				if (releaseStrategy.canRelease(group)) {
@@ -234,7 +240,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
 				return true;
 			}
 			return false;
+
 		}
+
 	}
 
 	private Object getLock(Object correlationKey) {

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

@@ -38,6 +38,8 @@ public interface MessageGroup {
 
 	void mark();
 
-	public Message<?> getOne();
+	Message<?> getOne();
+	
+	long getTimestamp();
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupCallback.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2002-2010 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;
+
+/**
+ * @author Dave Syer
+ * 
+ * @since 2.0
+ *
+ */
+public interface MessageGroupCallback {
+	
+	void execute(MessageGroup group);
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java

@@ -1,17 +1,14 @@
 /*
  * 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.
+ * 
+ * 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;
 
@@ -61,4 +58,22 @@ public interface MessageGroupStore {
 	 */
 	void removeMessageGroup(Object correlationKey);
 
+	/**
+	 * Register a callback for when a message group is expired through {@link #expireMessageGroups(long)}.
+	 * 
+	 * @param callback a callback to execute when a message group is cleaned up
+	 */
+	void registerExpiryCallback(MessageGroupCallback callback);
+
+	/**
+	 * Extract all expired groups (those whose timestamp is less than the threshold provided) and call each of the
+	 * registered callbacks on them in turn.
+	 * 
+	 * @param timestamp the timestamp threshold to use
+	 * @return the number of message groups expired
+	 * 
+	 * @see #registerExpiryCallback(MessageGroupCallback)
+	 */
+	int expireMessageGroups(long timestamp);
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java

@@ -38,14 +38,11 @@ public class SimpleMessageGroup implements MessageGroup {
 
 	public final Collection<Message<?>> unmarked = new HashSet<Message<?>>();
 
+	private final long timestamp;
+
 	public SimpleMessageGroup(Object correlationKey) {
 		this.correlationKey = correlationKey;
-	}
-
-	public SimpleMessageGroup(MessageGroup template) {
-		this.correlationKey = template.getCorrelationKey();
-		this.marked.addAll(template.getMarked());
-		this.unmarked.addAll(template.getUnmarked());
+		this.timestamp = System.currentTimeMillis();
 	}
 
 	public SimpleMessageGroup(Collection<? extends Message<?>> originalMessages,
@@ -55,6 +52,17 @@ public class SimpleMessageGroup implements MessageGroup {
 			add(message);
 		}
 	}
+	
+	public SimpleMessageGroup(MessageGroup template) {
+		this.correlationKey = template.getCorrelationKey();
+		this.marked.addAll(template.getMarked());
+		this.unmarked.addAll(template.getUnmarked());
+		this.timestamp = template.getTimestamp();
+	}
+
+	public long getTimestamp() {
+		return timestamp;
+	}
 
 	public boolean add(Message<?> message) {
 		if (isMember(message)) {

org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java

@@ -1,25 +1,26 @@
 /*
  * Copyright 2002-2009 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.
+ * 
+ * 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.Collection;
+import java.util.LinkedHashSet;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessagingException;
 import org.springframework.integration.util.UpperBound;
@@ -37,12 +38,16 @@ import org.springframework.util.Assert;
  */
 public class SimpleMessageStore implements MessageStore, MessageGroupStore {
 
+	private static final Log logger = LogFactory.getLog(SimpleMessageStore.class);
+
 	private final ConcurrentMap<UUID, Message<?>> idToMessage;
 
 	private final ConcurrentMap<Object, SimpleMessageGroup> correlationToMessageGroup;
 
 	private final UpperBound upperBound;
 
+	private Collection<MessageGroupCallback> expiryCallbacks = new LinkedHashSet<MessageGroupCallback>();
+
 	/**
 	 * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given
 	 * capacity is less than 1.
@@ -77,36 +82,72 @@ public class SimpleMessageStore implements MessageStore, MessageGroupStore {
 		if (key != null) {
 			upperBound.release();
 			return this.idToMessage.remove(key);
-		}
-		else
+		} else
 			return null;
 	}
 
 	public MessageGroup getMessageGroup(Object correlationId) {
 		Assert.notNull(correlationId, "'correlationKey' must not be null");
-		MessageGroup collection = correlationToMessageGroup.get(correlationId);
-		if (collection == null) {
+		SimpleMessageGroup group = correlationToMessageGroup.get(correlationId);
+		if (group == null) {
 			return new SimpleMessageGroup(correlationId);
 		}
-		return new SimpleMessageGroup(collection);
+		return new SimpleMessageGroup(group);
 	}
 
 	public void addMessageToGroup(Object correlationId, Message<?> message) {
 		getMessageGroupInternal(correlationId).add(message);
 	}
-	
-	public void markMessageGroup(MessageGroup group) {		
+
+	public void markMessageGroup(MessageGroup group) {
 		Object correlationId = group.getCorrelationKey();
 		MessageGroup internal = getMessageGroupInternal(correlationId);
 		internal.mark();
-		group.mark();	
+		group.mark();
 	}
 
 	public void removeMessageGroup(Object correlationId) {
 		correlationToMessageGroup.remove(correlationId);
 	}
 
-	private MessageGroup getMessageGroupInternal(Object correlationId) {
+	public void registerExpiryCallback(MessageGroupCallback callback) {
+		expiryCallbacks.add(callback);
+	}
+
+	public int expireMessageGroups(long timestamp) {
+		int count = 0;
+		for (MessageGroup group : correlationToMessageGroup.values()) {
+			if (group.getTimestamp() < timestamp) {
+				count++;
+				expire(group);
+				removeMessageGroup(group.getCorrelationKey());
+			}
+		}
+		return count;
+	}
+
+	private void expire(MessageGroup group) {
+
+		RuntimeException exception = null;
+
+		for (MessageGroupCallback callback : expiryCallbacks) {
+			try {
+				callback.execute(group);
+			} catch (RuntimeException e) {
+				if (exception == null) {
+					exception = e;
+				}
+				logger.error("Exception in expiry callback", e);
+			}
+		}
+
+		if (exception != null) {
+			throw exception;
+		}
+
+	}
+
+	private SimpleMessageGroup getMessageGroupInternal(Object correlationId) {
 		if (!correlationToMessageGroup.containsKey(correlationId)) {
 			correlationToMessageGroup.putIfAbsent(correlationId, new SimpleMessageGroup(correlationId));
 		}