1 files changed, 160 insertions, 0 deletions

diff --git a/src/unix.rs b/src/unix.rs
new file mode 100644
index 0000000..e241b64
--- /dev/null
+++ b/src/unix.rs
@@ -0,0 +1,160 @@
+//! UNIX Domain Socket backed channel and listener implementations.
+//!
+//! ```no_run
+//! #[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 listener = urpc::unix::bind("./hello.service").await?;
+//!
+//!     // 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(urpc::unix::connect("./hello.service").await?);
+//!     let client = HelloClient::new(channel);
+//!     let greeting = client.hello("World".into()).await.unwrap();
+//!     assert_eq!(greeting, "Hello, World!");
+//!     Ok(())
+//! }
+//! ```
+use std::{path::Path, pin::Pin};
+
+use futures::{Sink, Stream};
+use tokio::{
+    net::{UnixListener, UnixStream},
+    sync::mpsc::Receiver,
+    task::AbortHandle,
+};
+use tokio_util::codec::Framed;
+
+use crate::{
+    Channel, Listener,
+    protocol::{RpcMessage, RpcMessageCodec},
+};
+
+pub struct UnixChannel(Framed<UnixStream, RpcMessageCodec>);
+
+impl UnixChannel {
+    fn new(stream: UnixStream) -> Self {
+        Self(Framed::new(stream, RpcMessageCodec::default()))
+    }
+
+    pub async fn connect(path: impl AsRef<Path>) -> std::io::Result<Self> {
+        let stream = UnixStream::connect(path).await?;
+        Ok(Self::new(stream))
+    }
+}
+
+impl Sink<RpcMessage> for UnixChannel {
+    type Error = std::io::Error;
+
+    fn poll_ready(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::result::Result<(), Self::Error>> {
+        Sink::poll_ready(Pin::new(&mut self.get_mut().0), cx)
+    }
+
+    fn start_send(self: Pin<&mut Self>, item: RpcMessage) -> std::result::Result<(), Self::Error> {
+        Sink::start_send(Pin::new(&mut self.get_mut().0), item)
+    }
+
+    fn poll_flush(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::result::Result<(), Self::Error>> {
+        Sink::poll_flush(Pin::new(&mut self.get_mut().0), cx)
+    }
+
+    fn poll_close(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::result::Result<(), Self::Error>> {
+        Sink::poll_close(Pin::new(&mut self.get_mut().0), cx)
+    }
+}
+
+impl Stream for UnixChannel {
+    type Item = std::io::Result<RpcMessage>;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        Stream::poll_next(Pin::new(&mut self.get_mut().0), cx)
+    }
+}
+
+impl Channel for UnixChannel {}
+
+pub struct UnixChannelListener {
+    receiver: Receiver<UnixChannel>,
+    abort: AbortHandle,
+}
+
+impl Drop for UnixChannelListener {
+    fn drop(&mut self) {
+        self.abort.abort();
+    }
+}
+
+impl UnixChannelListener {
+    pub async fn bind(path: impl AsRef<Path>) -> std::io::Result<Self> {
+        let path = path.as_ref();
+        if tokio::fs::try_exists(path).await? {
+            tokio::fs::remove_file(path).await?;
+        }
+        let listener = UnixListener::bind(path)?;
+        let (sender, receiver) = tokio::sync::mpsc::channel(8);
+        let abort = tokio::spawn(async move {
+            while let Ok((stream, _addr)) = listener.accept().await {
+                if sender.send(UnixChannel::new(stream)).await.is_err() {
+                    break;
+                }
+            }
+        })
+        .abort_handle();
+        Ok(Self { receiver, abort })
+    }
+}
+
+impl Stream for UnixChannelListener {
+    type Item = std::io::Result<UnixChannel>;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        self.get_mut().receiver.poll_recv(cx).map(|v| v.map(Ok))
+    }
+}
+
+impl Listener<UnixChannel> for UnixChannelListener {}
+
+pub async fn bind(path: impl AsRef<Path>) -> std::io::Result<UnixChannelListener> {
+    UnixChannelListener::bind(path).await
+}
+
+pub async fn connect(path: impl AsRef<Path>) -> std::io::Result<UnixChannel> {
+    UnixChannel::connect(path).await
+}