Added MessageChannel interface and simple point-to-point implementation (EAI-2)

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();
+
+}

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<Message> queue;
+
+
+	public PointToPointChannel(int capacity) {
+		queue = new LinkedBlockingQueue<Message>(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 <code>null</code> if the receiving thread
+	 * is interrupted.
+	 */
+	public Message receive() {
+		try {
+			return queue.take();
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return null;
+		}
+	}
+
+}

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());
+	}
+
+}