Added 'clear()' and 'purge()' methods to channel and added 'isExpired()' method to Message.

spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.channel;
 
+import java.util.List;
+
 import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.message.Message;
 
@@ -83,4 +85,14 @@ public interface MessageChannel {
 	 */
 	Message receive(long timeout);
 
+	/**
+	 * Remove all {@link Message Messages} from this channel.
+	 */
+	List<Message> clear();
+
+	/**
+	 * Remove any expired {@link Message Messages} from this channel.
+	 */
+	List<Message> purge();
+
 }

spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.channel;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -103,4 +105,24 @@ public class SimpleChannel extends AbstractMessageChannel {
 		}
 	}
 
+	public List<Message> clear() {
+		List<Message> clearedMessages = new ArrayList<Message>();
+		this.queue.drainTo(clearedMessages);
+		return clearedMessages;
+	}
+
+	public List<Message> purge() {
+		List<Message> purgedMessages = new ArrayList<Message>();
+		// take a snapshot
+		Object[] array = this.queue.toArray();
+		for (Object o : array) {
+			Message message = (Message) o;
+			if (message.isExpired() && this.queue.remove(message)) {
+				// message was still in the queue
+				purgedMessages.add(message);
+			}
+		}
+		return purgedMessages;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.message;
 
+import java.util.Date;
+
 import org.springframework.integration.util.RandomGuidUidGenerator;
 import org.springframework.integration.util.UidGenerator;
 import org.springframework.util.Assert;
@@ -82,4 +84,9 @@ public class GenericMessage<T> implements Message<T> {
 		this.payload = newPayload;
 	}
 
+	public boolean isExpired() {
+		Date expiration = this.header.getExpiration();
+		return (expiration != null) ? expiration.getTime() < System.currentTimeMillis() : false;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/message/Message.java

@@ -29,4 +29,6 @@ public interface Message<T> {
 
 	T getPayload();
 
+	boolean isExpired();
+
 }