diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index b77a4bfba9..b8b6af7f40 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/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); + } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/PointToPointChannel.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/PointToPointChannel.java index bcf0486ceb..ea0af86251 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/PointToPointChannel.java +++ b/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 true if the message is sent successfully or + * false 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 null if the receiving thread - * is interrupted. + * @param message the Message to send + * @param timeout the timeout in milliseconds + * + * @return true if the message is sent successfully, + * false 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 null 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 null 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; + } + } + } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java b/spring-eai-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java index d8c5e005e4..f561c5530b 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java +++ b/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()); + } + }