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 c7d284b7c3..651b469ee0 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 @@ -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 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 ea0af86251..3f233fc47b 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 @@ -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 queue; + /** + * Create a channel with the specified queue capacity. + */ public PointToPointChannel(int capacity) { queue = new LinkedBlockingQueue(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 true if the message is sent successfully, - * false if the message cannot be sent within the - * allotted time or the sending thread is interrupted. + * 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 { @@ -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 null 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 null 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); + } + }