async-std support

Cargo.lock

@@ -5576,7 +5576,6 @@ dependencies = [
  "nix 0.27.1",
  "num-traits",
  "once_cell",
- "owning_ref",
  "paranoid-android",
  "parking_lot 0.12.1",
  "paste",

veilid-cli/Cargo.toml

@@ -55,7 +55,7 @@ flexi_logger = { version = "^0", features = ["use_chrono_for_offset"] }
 thiserror = "^1"
 crossbeam-channel = "^0"
 hex = "^0"
-veilid-tools = { version = "0.2.5", path = "../veilid-tools" }
+veilid-tools = { version = "0.2.5", path = "../veilid-tools", default-features = false}
 
 json = "^0"
 stop-token = { version = "^0", default-features = false }

veilid-cli/src/client_api_connection.rs

@@ -9,8 +9,7 @@ use stop_token::{future::FutureExt as _, StopSource};
 
 cfg_if! {
     if #[cfg(feature="rt-async-std")] {
-        use async_std::io::prelude::BufReadExt;
+        use futures::{AsyncBufReadExt, AsyncWriteExt};
-        use async_std::io::WriteExt;
         use async_std::io::BufReader;
     } else if #[cfg(feature="rt-tokio")] {
         use tokio::io::AsyncBufReadExt;
@@ -243,7 +242,7 @@ impl ClientApiConnection {
         cfg_if! {
             if #[cfg(feature="rt-async-std")] {
                 use futures::AsyncReadExt;
-                let (reader, mut writer) = stream.split();
+                let (reader, writer) = stream.split();
                 let reader = BufReader::new(reader);
             } else {
                 let (reader, writer) = stream.into_split();

veilid-core/Cargo.toml

@@ -64,7 +64,6 @@ veilid-tools = { version = "0.2.5", path = "../veilid-tools", features = [
 ], default-features = false }
 paste = "1.0.14"
 once_cell = "1.19.0"
-owning_ref = "0.4.1"
 backtrace = "0.3.69"
 num-traits = "0.2.17"
 shell-words = "1.1.0"

veilid-server/Cargo.toml

@@ -15,6 +15,7 @@ path = "src/main.rs"
 
 [features]
 default = ["rt-tokio", "veilid-core/default"]
+default-async-std = ["rt-async-std", "veilid-core/default-async-std"]
 crypto-test = ["rt-tokio", "veilid-core/crypto-test"]
 crypto-test-none = ["rt-tokio", "veilid-core/crypto-test-none"]
 

veilid-server/src/client_api.rs

@@ -20,8 +20,7 @@ const MAX_NON_JSON_LOGGING: usize = 50;
 
 cfg_if! {
     if #[cfg(feature="rt-async-std")] {
-        use async_std::io::prelude::BufReadExt;
+        use futures_util::{AsyncBufReadExt, AsyncWriteExt};
-        use async_std::io::WriteExt;
     } else if #[cfg(feature="rt-tokio")] {
         use tokio::io::AsyncBufReadExt;
         use tokio::io::AsyncWriteExt;
@@ -116,11 +115,11 @@ impl ClientApi {
                 return Err(e);
             }
         }
-        let listener = IpcListener::bind(ipc_path.clone()).await?;
+        let mut listener = IpcListener::bind(ipc_path.clone()).await?;
         debug!("IPC Client API listening on: {:?}", ipc_path);
 
         // Process the incoming accept stream
-        let mut incoming_stream = listener.incoming();
+        let mut incoming_stream = listener.incoming()?;
 
         // Make wait group for all incoming connections
         let awg = AsyncWaitGroup::new();
@@ -444,7 +443,7 @@ impl ClientApi {
         cfg_if! {
             if #[cfg(feature="rt-async-std")] {
                 use futures_util::AsyncReadExt;
-                let (reader, mut writer) = stream.split();
+                let (reader, writer) = stream.split();
                 let reader = BufReader::new(reader);
             } else {
                 let (reader, writer) = stream.into_split();

veilid-tools/src/ipc/ipc_async_std/unix.rs

@@ -0,0 +1,141 @@
+use crate::*;
+use async_std::io::Read as AsyncRead;
+use async_std::io::Write as AsyncWrite;
+use async_std::os::unix::net::{Incoming, UnixListener, UnixStream};
+use futures_util::AsyncRead as FuturesAsyncRead;
+use futures_util::AsyncWrite as FuturesAsyncWrite;
+use futures_util::Stream;
+use std::path::PathBuf;
+use std::{io, path::Path};
+
+/////////////////////////////////////////////////////////////
+
+pub struct IpcStream {
+    internal: UnixStream,
+}
+
+impl IpcStream {
+    pub async fn connect<P: AsRef<Path>>(path: P) -> io::Result<IpcStream> {
+        Ok(IpcStream {
+            internal: UnixStream::connect(path.as_ref()).await?,
+        })
+    }
+}
+
+impl FuturesAsyncRead for IpcStream {
+    fn poll_read(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut [u8],
+    ) -> std::task::Poll<io::Result<usize>> {
+        <UnixStream as AsyncRead>::poll_read(std::pin::Pin::new(&mut self.internal), cx, buf)
+    }
+}
+
+impl FuturesAsyncWrite for IpcStream {
+    fn poll_write(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<io::Result<usize>> {
+        <UnixStream as AsyncWrite>::poll_write(std::pin::Pin::new(&mut self.internal), cx, buf)
+    }
+
+    fn poll_flush(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<io::Result<()>> {
+        <UnixStream as AsyncWrite>::poll_flush(std::pin::Pin::new(&mut self.internal), cx)
+    }
+
+    fn poll_close(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<io::Result<()>> {
+        <UnixStream as AsyncWrite>::poll_close(std::pin::Pin::new(&mut self.internal), cx)
+    }
+}
+
+/////////////////////////////////////////////////////////////
+
+pub struct IpcIncoming<'a> {
+    path: PathBuf,
+    internal: Incoming<'a>,
+}
+
+impl<'a> Drop for IpcIncoming<'a> {
+    fn drop(&mut self) {
+        // Clean up IPC path
+        if let Err(e) = std::fs::remove_file(&self.path) {
+            warn!("Unable to remove IPC socket: {}", e);
+        }
+    }
+}
+
+impl<'a> Stream for IpcIncoming<'a> {
+    type Item = io::Result<IpcStream>;
+
+    fn poll_next(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        match <Incoming as Stream>::poll_next(std::pin::Pin::new(&mut self.internal), cx) {
+            std::task::Poll::Ready(ro) => {
+                std::task::Poll::Ready(ro.map(|rr| rr.map(|s| IpcStream { internal: s })))
+            }
+            std::task::Poll::Pending => std::task::Poll::Pending,
+        }
+    }
+}
+
+/////////////////////////////////////////////////////////////
+
+pub struct IpcListener {
+    path: Option<PathBuf>,
+    internal: Option<Arc<UnixListener>>,
+}
+
+impl IpcListener {
+    /// Creates a new `IpcListener` bound to the specified path.
+    pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<Self> {
+        Ok(Self {
+            path: Some(path.as_ref().to_path_buf()),
+            internal: Some(Arc::new(UnixListener::bind(path.as_ref()).await?)),
+        })
+    }
+
+    /// Accepts a new incoming connection to this listener.
+    pub fn accept(&self) -> SendPinBoxFuture<io::Result<IpcStream>> {
+        let this = IpcListener {
+            path: self.path.clone(),
+            internal: self.internal.clone(),
+        };
+        Box::pin(async move {
+            Ok(IpcStream {
+                internal: this.internal.as_ref().unwrap().accept().await?.0,
+            })
+        })
+    }
+
+    /// Returns a stream of incoming connections.
+    pub fn incoming<'a>(&'a mut self) -> io::Result<IpcIncoming<'a>> {
+        if self.path.is_none() {
+            return Err(io::Error::from(io::ErrorKind::NotConnected));
+        }
+        Ok(IpcIncoming {
+            path: self.path.take().unwrap(),
+            internal: self.internal.as_ref().unwrap().incoming(),
+        })
+    }
+}
+
+impl Drop for IpcListener {
+    fn drop(&mut self) {
+        // Clean up IPC path
+        if let Some(path) = &self.path {
+            if let Err(e) = std::fs::remove_file(path) {
+                warn!("Unable to remove IPC socket: {}", e);
+            }
+        }
+    }
+}

veilid-tools/src/ipc/ipc_tokio/unix.rs

@@ -65,12 +65,13 @@ impl FuturesAsyncWrite for IpcStream {
 
 /////////////////////////////////////////////////////////////
 
-pub struct IpcIncoming {
+pub struct IpcIncoming<'a> {
     path: PathBuf,
     internal: UnixListenerStream,
+    phantom: std::marker::PhantomData<&'a ()>,
 }
 
-impl Stream for IpcIncoming {
+impl<'a> Stream for IpcIncoming<'a> {
     type Item = io::Result<IpcStream>;
 
     fn poll_next(
@@ -87,7 +88,7 @@ impl Stream for IpcIncoming {
     }
 }
 
-impl Drop for IpcIncoming {
+impl<'a> Drop for IpcIncoming<'a> {
     fn drop(&mut self) {
         // Clean up IPC path
         if let Err(e) = std::fs::remove_file(&self.path) {
@@ -126,13 +127,17 @@ impl IpcListener {
     }
 
     /// Returns a stream of incoming connections.
-    pub fn incoming(mut self) -> IpcIncoming {
+    pub fn incoming(&mut self) -> io::Result<IpcIncoming<'_>> {
-        IpcIncoming {
+        if self.path.is_none() {
+            return Err(io::Error::from(io::ErrorKind::NotConnected));
+        }
+        Ok(IpcIncoming {
             path: self.path.take().unwrap(),
             internal: UnixListenerStream::new(
                 Arc::into_inner(self.internal.take().unwrap()).unwrap(),
             ),
-        }
+            phantom: std::marker::PhantomData,
+        })
     }
 }