Added MessageSelector receive methods to PointToPointChannel

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

@@ -20,8 +20,8 @@ import org.springframework.integration.MessageSource;
 import org.springframework.integration.MessageTarget;
 
 /**
- * Base channel interface to combine the definitions of {@link MessageSource}
- * for message reception and and {@link MessageTarget} for message sending.
+ * Base channel interface that combines the definitions of {@link MessageSource}
+ * for message reception and {@link MessageTarget} for message sending.
  * 
  * @author Mark Fisher
  * @see MessageSource

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

@@ -21,11 +21,12 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageSelector;
 
 /**
- * Simple implementation of a point-to-point message channel. Messages are
- * stored in a queue whose capacity may be specified upon construction. If
- * no capacity is specified, the {@link #DEFAULT_CAPACITY} will be used.
+ * Simple implementation of a point-to-point message channel. Each Messages is
+ * placed in a queue whose capacity may be specified upon construction. If no
+ * capacity is specified, the {@link #DEFAULT_CAPACITY} will be used.
  * 
  * @author Mark Fisher
  */
@@ -37,10 +38,16 @@ public class PointToPointChannel implements MessageChannel {
 	private BlockingQueue<Message> queue;
 
 
+	/**
+	 * Create a channel with the specified queue capacity.
+	 */
 	public PointToPointChannel(int capacity) {
 		queue = new LinkedBlockingQueue<Message>(capacity);
 	}
 
+	/**
+	 * Create a channel with the default queue capacity.
+	 */
 	public PointToPointChannel() {
 		this(DEFAULT_CAPACITY);
 	}
@@ -70,15 +77,15 @@ public class PointToPointChannel implements MessageChannel {
 	/**
 	 * Send a message on this channel. If the queue is full, this method will
 	 * block until either the timeout occurs or the sending thread is
-	 * interrupted. If the specified timeout is less than 1, the method
-	 * will return immediately.
+	 * interrupted. If the specified timeout is less than 1, the method will
+	 * return immediately.
 	 * 
 	 * @param message the Message to send
 	 * @param timeout the timeout in milliseconds
 	 * 
 	 * @return <code>true</code> if the message is sent successfully,
-	 * <code>false</code> if the message cannot be sent within the
-	 * allotted time or the sending thread is interrupted.
+	 * <code>false</code> if the message cannot be sent within the allotted
+	 * time or the sending thread is interrupted.
 	 */
 	public boolean send(Message message, long timeout) {
 		try {
@@ -112,7 +119,7 @@ public class PointToPointChannel implements MessageChannel {
 
 	/**
 	 * Receive the message at the head of the queue. If the queue is empty, this
-	 * method will block until the allotted timeout elapses. If the specified 
+	 * method will block until the allotted timeout elapses. If the specified
 	 * timeout is less than 1, the method will return immediately.
 	 * 
 	 * @param timeout the timeout in milliseconds
@@ -134,4 +141,49 @@ public class PointToPointChannel implements MessageChannel {
 		}
 	}
 
+	/**
+	 * Receive the first message that is accepted by the specified selector
+	 * starting from the head of the queue. If the queue is empty, this method
+	 * will block until the allotted timeout elapses. If the specified timeout
+	 * is less than 1, the method will return immediately.
+	 * 
+	 * @param selector the selector to use
+	 * @param timeout the timeout in milliseconds
+	 * 
+	 * @return the first accepted message or <code>null</code> in case the
+	 * selector does not accept any message within the allotted time or the
+	 * receiving thread is interrupted.
+	 */
+	public Message receive(MessageSelector selector, long timeout) {
+		long start = System.currentTimeMillis();
+		while (timeout <= 0 || System.currentTimeMillis() - start < timeout) {
+			Object[] elements = this.queue.toArray();
+			for (int i = (elements.length - 1); i >= 0; i--) {
+				Message m = (Message) elements[i];
+				if (selector.accept(m) && this.queue.remove(m)) {
+					return m;
+				}
+			}
+			if (timeout == 0) {
+				return null;
+			}
+		}
+		return null;
+	}
+
+	/**
+	 * Receive the first message that is accepted by the specified selector
+	 * starting from the head of the queue. If the queue is empty, this method
+	 * will block.
+	 * 
+	 * @param selector the selector to use
+	 * 
+	 * @return the first accepted message or <code>null</code> in case the
+	 * selector does not accept any message or the receiving thread is
+	 * interrupted.
+	 */
+	public Message receive(MessageSelector selector) {
+		return this.receive(selector, -1);
+	}
+
 }