🧵 zb: Use async versions of UnixStream when possible

zbus/src/address/transport/mod.rs

@@ -12,13 +12,13 @@ use async_io::Async;
 use std::collections::HashMap;
 #[cfg(not(feature = "tokio"))]
 use std::net::TcpStream;
-#[cfg(unix)]
-use std::os::unix::net::{SocketAddr, UnixStream};
+#[cfg(all(unix, not(feature = "tokio")))]
+use std::os::unix::net::UnixStream;
 #[cfg(feature = "tokio")]
 use tokio::net::TcpStream;
 #[cfg(feature = "tokio-vsock")]
 use tokio_vsock::VsockStream;
-#[cfg(windows)]
+#[cfg(all(windows, not(feature = "tokio")))]
 use uds_windows::UnixStream;
 #[cfg(all(feature = "vsock", not(feature = "tokio")))]
 use vsock::VsockStream;
@@ -55,8 +55,6 @@ pub use ibus::Ibus;
 #[path = "vsock.rs"]
 // Gotta rename to avoid name conflict with the `vsock` crate.
 mod vsock_transport;
-#[cfg(target_os = "linux")]
-use std::os::linux::net::SocketAddrExt;
 #[cfg(any(
     all(feature = "vsock", not(feature = "tokio")),
     feature = "tokio-vsock"
@@ -102,59 +100,11 @@ impl Transport {
     #[cfg_attr(any(unix, windows), async_recursion::async_recursion)]
     pub(super) async fn connect(self) -> Result<Stream> {
         match self {
-            Transport::Unix(unix) => {
-                // This is a `path` in case of Windows until uds_windows provides the needed API:
-                // https://github.com/haraldh/rust_uds_windows/issues/14
-                let addr = match unix.take_path() {
-                    #[cfg(unix)]
-                    UnixSocket::File(path) => SocketAddr::from_pathname(path)?,
-                    #[cfg(windows)]
-                    UnixSocket::File(path) => path,
-                    #[cfg(target_os = "linux")]
-                    UnixSocket::Abstract(name) => {
-                        SocketAddr::from_abstract_name(name.as_encoded_bytes())?
-                    }
-                    UnixSocket::Dir(_) | UnixSocket::TmpDir(_) => {
-                        // you can't connect to a unix:dir
-                        return Err(Error::Unsupported);
-                    }
-                };
-                let stream = crate::Task::spawn_blocking(
-                    move || -> Result<_> {
-                        #[cfg(unix)]
-                        let stream = UnixStream::connect_addr(&addr)?;
-                        #[cfg(windows)]
-                        let stream = UnixStream::connect(addr)?;
-                        stream.set_nonblocking(true)?;
-
-                        Ok(stream)
-                    },
-                    "unix stream connection",
-                )
-                .await??;
-                #[cfg(not(feature = "tokio"))]
-                {
-                    Async::new(stream)
-                        .map(Stream::Unix)
-                        .map_err(|e| Error::InputOutput(e.into()))
-                }
+            #[cfg(all(not(unix), feature = "tokio"))]
+            Transport::Unix(_) => Err(Error::Unsupported),
+            #[cfg(any(unix, not(feature = "tokio")))]
+            Transport::Unix(unix) => unix.connect().await.map(Stream::Unix),
 
-                #[cfg(feature = "tokio")]
-                {
-                    #[cfg(unix)]
-                    {
-                        tokio::net::UnixStream::from_std(stream)
-                            .map(Stream::Unix)
-                            .map_err(|e| Error::InputOutput(e.into()))
-                    }
-
-                    #[cfg(not(unix))]
-                    {
-                        let _ = stream;
-                        Err(Error::Unsupported)
-                    }
-                }
-            }
             #[cfg(unix)]
             Transport::Unixexec(unixexec) => unixexec.connect().await.map(Stream::Unixexec),
             #[cfg(all(feature = "vsock", not(feature = "tokio")))]

zbus/src/address/transport/unix.rs

@@ -1,3 +1,6 @@
+#[cfg(any(unix, not(feature = "tokio")))]
+use crate::{Error, Result};
+
 #[cfg(target_os = "linux")]
 use std::ffi::OsString;
 use std::{
@@ -9,6 +12,14 @@ use std::{
 #[cfg(unix)]
 use super::encode_percents;
 
+#[cfg(not(feature = "tokio"))]
+use async_io::Async;
+
+#[cfg(all(unix, not(feature = "tokio")))]
+use std::os::unix::net::UnixStream;
+#[cfg(all(windows, not(feature = "tokio")))]
+use uds_windows::UnixStream;
+
 /// A Unix domain socket transport in a D-Bus address.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Unix {
@@ -55,6 +66,63 @@ impl Unix {
 
         Ok(Self::new(path))
     }
+
+    #[cfg(not(feature = "tokio"))]
+    pub(super) async fn connect(self) -> Result<Async<UnixStream>> {
+        let addr = self.take_addr()?;
+
+        #[cfg(unix)]
+        let stream = Async::<UnixStream>::connect(addr).await;
+
+        #[cfg(not(unix))]
+        let stream = Async::new(self.get_stream(addr).await?);
+
+        stream.map_err(|e| Error::InputOutput(e.into()))
+    }
+
+    #[cfg(all(unix, feature = "tokio"))]
+    pub(super) async fn connect(self) -> Result<tokio::net::UnixStream> {
+        let addr = self.take_addr()?;
+        tokio::net::UnixStream::connect(addr)
+            .await
+            .map_err(|e| Error::InputOutput(e.into()))
+    }
+
+    #[cfg(not(unix))]
+    async fn get_stream(self, addr: PathBuf) -> Result<UnixStream> {
+        crate::Task::spawn_blocking(
+            move || -> Result<_> {
+                let stream = UnixStream::connect(addr)?;
+                stream.set_nonblocking(true)?;
+
+                Ok(stream)
+            },
+            "unix stream connection",
+        )
+        .await?
+    }
+
+    #[cfg(any(unix, not(feature = "tokio")))]
+    fn take_addr(self) -> Result<PathBuf> {
+        // This is a `path` because neither uds_windows, tokio, nor async_io provide
+        // the SocketAddrExt functions.
+        match self.take_path() {
+            UnixSocket::File(path) => Ok(path),
+            #[cfg(target_os = "linux")]
+            UnixSocket::Abstract(name) => {
+                use std::os::unix::ffi::OsStringExt;
+
+                let mut v = name.into_vec();
+                v.insert(0, 0);
+
+                Ok(PathBuf::from(OsString::from_vec(v)))
+            }
+            UnixSocket::Dir(_) | UnixSocket::TmpDir(_) => {
+                // You can't connect to a unix:dir.
+                Err(Error::Unsupported)
+            }
+        }
+    }
 }
 
 impl Display for Unix {