Added timeout-accepting versions of send and receive to the channel (INT-.

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

@@ -19,15 +19,19 @@ package org.springframework.integration.channel;
 import org.springframework.integration.message.Message;
 
 /**
- * The base channel interface defining the common behavior
+ * Base channel interface defining the common behavior
  * of sending and receiving messages.   
  * 
  * @author Mark Fisher
  */
 public interface MessageChannel {
 
-	void send(Message message);
+	boolean send(Message message);
+
+	boolean send(Message message, long timeout);
 
 	Message receive();
 
+	Message receive(long timeout);
+
 }

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

@@ -18,14 +18,14 @@ package org.springframework.integration.channel;
 
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.integration.message.Message;
 
 /**
- * Simple implementation of a point-to-point message channel.
- * Messages are stored in a queue whose capacity may be
- * provided upon construction. If no capacity is specified,
- * the {@link #DEFAULT_CAPACITY} will be used.
+ * 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.
  * 
  * @author Mark Fisher
  */
@@ -47,27 +47,58 @@ public class PointToPointChannel implements MessageChannel {
 
 
 	/**
-	 * Send a message on this channel. If the queue is
-	 * full, this method will block until either space
-	 * becomes available or the sending thread is
+	 * Send a message on this channel. If the queue is full, this method will
+	 * block until either space becomes available or the sending thread is
 	 * interrupted.
+	 * 
+	 * @param message the Message to send
+	 * 
+	 * @return <code>true</code> if the message is sent successfully or
+	 * <code>false</code> if the sending thread is interrupted.
 	 */
-	public void send(Message message) {
+	public boolean send(Message message) {
 		try {
 			queue.put(message);
+			return true;
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
+			return false;
 		}
 	}
 
 	/**
-	 * Receive the message at the head of the queue.
-	 * If the queue is empty, this method will block.
+	 * 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.
 	 * 
-	 * @return the Message at the head of the queue
-	 * or <code>null</code> if the receiving thread
-	 * is interrupted.
+	 * @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.
+	 */
+	public boolean send(Message message, long timeout) {
+		try {
+			if (timeout > 0) {
+				return this.queue.offer(message, timeout, TimeUnit.MILLISECONDS);
+			}
+			return this.queue.offer(message);
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return false;
+		}
+	}
+
+	/**
+	 * Receive the message at the head of the queue. If the queue is empty, this
+	 * method will block.
+	 * 
+	 * @return the Message at the head of the queue or <code>null</code> if
+	 * the receiving thread is interrupted.
 	 */
 	public Message receive() {
 		try {
@@ -79,4 +110,28 @@ 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 
+	 * timeout is less than 1, the method will return immediately.
+	 * 
+	 * @param timeout the timeout in milliseconds
+	 * 
+	 * @return the message at the head of the queue or <code>null</code> in
+	 * case no message is available within the allotted time or the receiving
+	 * thread is interrupted.
+	 */
+	public Message receive(long timeout) {
+		try {
+			if (timeout > 0) {
+				return queue.poll(timeout, TimeUnit.MILLISECONDS);
+			}
+			return queue.poll();
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return null;
+		}
+	}
+
 }

spring-eai-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java

@@ -20,13 +20,14 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Test;
-
-import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.message.DocumentMessage;
+import org.springframework.integration.message.Message;
 
 /**
  * @author Mark Fisher
@@ -34,15 +35,17 @@ import org.springframework.integration.message.DocumentMessage;
 public class PointToPointChannelTests {
 
 	@Test
-	public void testSimpleReceive() throws Exception {
+	public void testSimpleSendAndReceive() throws Exception {
 		final AtomicBoolean messageReceived = new AtomicBoolean(false);
 		final CountDownLatch latch = new CountDownLatch(1);
 		final PointToPointChannel channel = new PointToPointChannel();
 		new Thread(new Runnable() {
 			public void run() {
-				channel.receive();
-				messageReceived.set(true);
-				latch.countDown();
+				Message message = channel.receive();
+				if (message != null) {
+					messageReceived.set(true);
+					latch.countDown();
+				}
 			}
 		}).start();
 		assertFalse(messageReceived.get());
@@ -51,4 +54,97 @@ public class PointToPointChannelTests {
 		assertTrue(messageReceived.get());
 	}
 
+	@Test
+	public void testImmediateReceive() throws Exception {
+		final AtomicBoolean messageReceived = new AtomicBoolean(false);
+		final PointToPointChannel channel = new PointToPointChannel();
+		final CountDownLatch latch1 = new CountDownLatch(1);
+		final CountDownLatch latch2 = new CountDownLatch(1);
+		Executor singleThreadExecutor = Executors.newSingleThreadExecutor();
+		Runnable receiveTask1 = new Runnable() {
+			public void run() {
+				Message message = channel.receive(0);
+				if (message != null) {
+					messageReceived.set(true);
+				}
+				latch1.countDown();
+			}
+		};
+		Runnable sendTask = new Runnable() {
+			public void run() {
+				channel.send(new DocumentMessage(1, "testing"));
+			}
+		};
+		singleThreadExecutor.execute(receiveTask1);
+		latch1.await();
+		singleThreadExecutor.execute(sendTask);
+		assertFalse(messageReceived.get());
+		Runnable receiveTask2 = new Runnable() {
+			public void run() {
+				Message message = channel.receive(0);
+				if (message != null) {
+					messageReceived.set(true);
+				}
+				latch2.countDown();
+			}
+		};
+		singleThreadExecutor.execute(receiveTask2);
+		latch2.await();
+		assertTrue(messageReceived.get());
+	}
+
+	@Test
+	public void testBlockingReceive() throws Exception{
+		final PointToPointChannel channel = new PointToPointChannel();
+		final AtomicBoolean receiveInterrupted = new AtomicBoolean(false);
+		final CountDownLatch latch = new CountDownLatch(1);
+		Thread t = new Thread(new Runnable() {
+			public void run() {
+				Message message = channel.receive();
+				receiveInterrupted.set(true);
+				assertTrue(message == null);
+				latch.countDown();
+			}
+		});
+		t.start();
+		assertFalse(receiveInterrupted.get());
+		t.interrupt();
+		latch.await();
+		assertTrue(receiveInterrupted.get());
+	}
+
+	@Test
+	public void testImmediateSend() {
+		PointToPointChannel channel = new PointToPointChannel(3);
+		boolean result1 = channel.send(new DocumentMessage(1, "test-1"));
+		assertTrue(result1);
+		boolean result2 = channel.send(new DocumentMessage(2, "test-2"), 100);
+		assertTrue(result2);
+		boolean result3 = channel.send(new DocumentMessage(3, "test-3"), 0);
+		assertTrue(result3);
+		boolean result4 = channel.send(new DocumentMessage(4, "test-4"), 0);
+		assertFalse(result4);
+	}
+
+	@Test
+	public void testBlockingSend() throws Exception{
+		final PointToPointChannel channel = new PointToPointChannel(1);
+		boolean result1 = channel.send(new DocumentMessage(1, "test-1"));
+		assertTrue(result1);
+		final AtomicBoolean sendInterrupted = new AtomicBoolean(false);
+		final CountDownLatch latch = new CountDownLatch(1);
+		Thread t = new Thread(new Runnable() {
+			public void run() {
+				channel.send(new DocumentMessage(2, "test-2"));
+				sendInterrupted.set(true);
+				latch.countDown();
+			}
+		});
+		t.start();
+		assertFalse(sendInterrupted.get());
+		t.interrupt();
+		latch.await();
+		assertTrue(sendInterrupted.get());
+	}
+
 }