From f86eff78930b81992196428d408712f5ab485bf8 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 20 Nov 2007 00:17:01 +0000 Subject: [PATCH] Added MessageChannel interface and simple point-to-point implementation (EAI-2) --- .../eai/channel/MessageChannel.java | 33 ++++++++ .../eai/channel/PointToPointChannel.java | 82 +++++++++++++++++++ .../eai/channel/PointToPointChannelTests.java | 53 ++++++++++++ 3 files changed, 168 insertions(+) create mode 100644 spring-eai-core/src/main/java/org/springframework/eai/channel/MessageChannel.java create mode 100644 spring-eai-core/src/main/java/org/springframework/eai/channel/PointToPointChannel.java create mode 100644 spring-eai-core/src/test/java/org/springframework/eai/channel/PointToPointChannelTests.java diff --git a/spring-eai-core/src/main/java/org/springframework/eai/channel/MessageChannel.java b/spring-eai-core/src/main/java/org/springframework/eai/channel/MessageChannel.java new file mode 100644 index 0000000000..a618824bfd --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/eai/channel/MessageChannel.java @@ -0,0 +1,33 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.eai.channel; + +import org.springframework.eai.message.Message; + +/** + * The base channel interface defining the common behavior + * of sending and receiving messages. + * + * @author Mark Fisher + */ +public interface MessageChannel { + + void send(Message message); + + Message receive(); + +} diff --git a/spring-eai-core/src/main/java/org/springframework/eai/channel/PointToPointChannel.java b/spring-eai-core/src/main/java/org/springframework/eai/channel/PointToPointChannel.java new file mode 100644 index 0000000000..ada318ee24 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/eai/channel/PointToPointChannel.java @@ -0,0 +1,82 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.eai.channel; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +import org.springframework.eai.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. + * + * @author Mark Fisher + */ +public class PointToPointChannel implements MessageChannel { + + private static final int DEFAULT_CAPACITY = 25; + + + private BlockingQueue queue; + + + public PointToPointChannel(int capacity) { + queue = new LinkedBlockingQueue(capacity); + } + + public PointToPointChannel() { + this(DEFAULT_CAPACITY); + } + + + /** + * 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. + */ + public void send(Message message) { + try { + queue.put(message); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** + * 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 { + return queue.take(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } + } + +} diff --git a/spring-eai-core/src/test/java/org/springframework/eai/channel/PointToPointChannelTests.java b/spring-eai-core/src/test/java/org/springframework/eai/channel/PointToPointChannelTests.java new file mode 100644 index 0000000000..f07f56671b --- /dev/null +++ b/spring-eai-core/src/test/java/org/springframework/eai/channel/PointToPointChannelTests.java @@ -0,0 +1,53 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.eai.channel; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Test; + +import org.springframework.eai.message.DocumentMessage; + +/** + * @author Mark Fisher + */ +public class PointToPointChannelTests { + + @Test + public void testSimpleReceive() 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(); + } + }).start(); + assertFalse(messageReceived.get()); + channel.send(new DocumentMessage(1, "testing")); + latch.await(25, TimeUnit.MILLISECONDS); + assertTrue(messageReceived.get()); + } + +}