1 files changed, 162 insertions, 0 deletions

diff --git a/src/channel.rs b/src/channel.rs
new file mode 100644
index 0000000..f80ca87
--- /dev/null
+++ b/src/channel.rs
@@ -0,0 +1,162 @@
+//! MPSC channel backed channel and listener implementations.
+//!
+//! This module provides an mpsc channel based listener/channel implementation that can be useful
+//! to for tests.
+//!
+//! The current implementation uses [`tokio`]'s unbounded mpsc channels.
+//!
+//! ```
+//! #[urpc::service]
+//! trait Hello {
+//!     type Error = ();
+//!
+//!     async fn hello(name: String) -> String;
+//! }
+//!
+//! struct HelloServer;
+//!
+//! impl Hello for HelloServer {
+//!     async fn hello(&self, _ctx: urpc::Context, name: String) -> Result<String, ()> {
+//!         Ok(format!("Hello, {name}!"))
+//!     }
+//! }
+//!
+//! #[tokio::main]
+//! async fn main() -> Result<(), Box<dyn std::error::Error>>{
+//!     let (dialer, listener) = urpc::channel::new();
+//!
+//!     // spawn the server
+//!     tokio::spawn(async move {
+//!         urpc::Server::default()
+//!             .with_listener(listener)
+//!             .with_service(HelloServer.into_service())
+//!             .serve()
+//!             .await
+//!     });
+//!
+//!     // create a client
+//!     let channel = urpc::ClientChannel::new(dialer.connect()?);
+//!     let client = HelloClient::new(channel);
+//!     let greeting = client.hello("World".into()).await.unwrap();
+//!     assert_eq!(greeting, "Hello, World!");
+//!     Ok(())
+//! }
+//! ```
+
+use futures::{Sink, Stream};
+use tokio::sync::mpsc;
+
+use crate::protocol::RpcMessage;
+
+pub struct ChannelListener {
+    receiver: mpsc::UnboundedReceiver<Channel>,
+}
+
+impl Stream for ChannelListener {
+    type Item = std::io::Result<Channel>;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        match self.get_mut().receiver.poll_recv(cx) {
+            std::task::Poll::Ready(Some(c)) => std::task::Poll::Ready(Some(Ok(c))),
+            std::task::Poll::Ready(None) => std::task::Poll::Ready(None),
+            std::task::Poll::Pending => std::task::Poll::Pending,
+        }
+    }
+}
+
+impl crate::Listener<Channel> for ChannelListener {}
+
+pub struct ChannelDialer {
+    sender: mpsc::UnboundedSender<Channel>,
+}
+
+impl ChannelDialer {
+    fn new() -> (Self, ChannelListener) {
+        let (sender, receiver) = mpsc::unbounded_channel();
+        (Self { sender }, ChannelListener { receiver })
+    }
+
+    pub fn connect(&self) -> std::io::Result<Channel> {
+        let (ch1, ch2) = Channel::new();
+        self.sender.send(ch1).expect("TODO: remove this");
+        Ok(ch2)
+    }
+}
+
+pub struct Channel {
+    sender: mpsc::UnboundedSender<RpcMessage>,
+    receiver: mpsc::UnboundedReceiver<RpcMessage>,
+}
+
+impl Channel {
+    fn new() -> (Self, Self) {
+        let (sender0, receiver0) = mpsc::unbounded_channel();
+        let (sender1, receiver1) = mpsc::unbounded_channel();
+        (
+            Self {
+                sender: sender0,
+                receiver: receiver1,
+            },
+            Self {
+                sender: sender1,
+                receiver: receiver0,
+            },
+        )
+    }
+}
+
+impl Stream for Channel {
+    type Item = std::io::Result<RpcMessage>;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        match self.get_mut().receiver.poll_recv(cx) {
+            std::task::Poll::Ready(Some(msg)) => std::task::Poll::Ready(Some(Ok(msg))),
+            std::task::Poll::Ready(None) => std::task::Poll::Ready(None),
+            std::task::Poll::Pending => std::task::Poll::Pending,
+        }
+    }
+}
+
+impl Sink<RpcMessage> for Channel {
+    type Error = std::io::Error;
+
+    fn poll_ready(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        return std::task::Poll::Ready(Ok(()));
+    }
+
+    fn start_send(self: std::pin::Pin<&mut Self>, item: RpcMessage) -> Result<(), Self::Error> {
+        match self.sender.send(item) {
+            Ok(()) => Ok(()),
+            Err(err) => Err(std::io::Error::other(err)),
+        }
+    }
+
+    fn poll_flush(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+
+    fn poll_close(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+}
+
+impl crate::Channel for Channel {}
+
+pub fn new() -> (ChannelDialer, ChannelListener) {
+    ChannelDialer::new()
+}