♻️ zb: Use async versions of UnixStream when possible

When connecting to a UnixSocket a blocking/threaded API is
being used. This commit changes this to use tokio::net::UnixStream
and Async<std::os::unix::net::UnixStream> when possible preventing
the use of extra threads. The windows version continues using the
blocking/threaded API.

The connect_addr API is dropped since neither tokio nor async-io
supports it. Instead, we use the connect functions in both
environments, since they allow for abstract names when the path is
prepended with '\0'.

Move the connection code to unix.rs for better future
maintainability.

Author: germaniuss

zbus/src/address/transport/mod.rs

@@ -12,14 +12,10 @@ 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(feature = "tokio")]
 use tokio::net::TcpStream;
 #[cfg(feature = "tokio-vsock")]
 use tokio_vsock::VsockStream;
-#[cfg(windows)]
-use uds_windows::UnixStream;
 #[cfg(all(feature = "vsock", not(feature = "tokio")))]
 use vsock::VsockStream;
 #[cfg(unix)]
@@ -33,7 +29,7 @@ use std::{
 };
 
 mod unix;
-pub use unix::{Unix, UnixSocket};
+pub use unix::{Unix, UnixSocket, UnixStream};
 mod tcp;
 pub use tcp::{Tcp, TcpTransportFamily};
 #[cfg(windows)]
@@ -55,8 +51,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 +96,7 @@ 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(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)
-                    }
-                }
-            }
+            Transport::Unix(unix) => unix.connect().await.map(Stream::Unix),
             #[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,5 @@
+use crate::{Error, Result};
+
 #[cfg(target_os = "linux")]
 use std::ffi::OsString;
 use std::{
@@ -9,6 +11,16 @@ use std::{
 #[cfg(unix)]
 use super::encode_percents;
 
+#[cfg(not(feature = "tokio"))]
+use async_io::Async;
+
+#[cfg(all(unix, not(feature = "tokio")))]
+pub use std::os::unix::net::UnixStream;
+#[cfg(all(unix, feature = "tokio"))]
+pub use tokio::net::UnixStream;
+#[cfg(windows)]
+pub use uds_windows::UnixStream;
+
 /// A Unix domain socket transport in a D-Bus address.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Unix {
@@ -31,6 +43,25 @@ impl Unix {
         self.path
     }
 
+    fn take_addr(self) -> Result<PathBuf> {
+        match self.take_path() {
+            // This is a `path` because neither uds_windows, tokio, nor async_io provide
+            // SocketAddrExt api
+            UnixSocket::File(path) => Ok(path),
+            #[cfg(target_os = "linux")]
+            UnixSocket::Abstract(name) => {
+                use std::{ffi::OsString, 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)
+            }
+        }
+    }
+
     pub(super) fn from_options(opts: std::collections::HashMap<&str, &str>) -> crate::Result<Self> {
         let path = opts.get("path");
         let abs = opts.get("abstract");
@@ -55,6 +86,45 @@ 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(windows)]
+        let stream = {
+            let stream = crate::Task::spawn_blocking(
+                move || -> Result<_> {
+                    let stream = UnixStream::connect(addr)?;
+                    stream.set_nonblocking(true)?;
+
+                    Ok(stream)
+                },
+                "unix stream connection",
+            )
+            .await??;
+            Async::new(stream)
+        };
+
+        stream.map_err(|e| Error::InputOutput(e.into()))
+    }
+
+    #[cfg(feature = "tokio")]
+    pub(super) async fn connect(self) -> Result<UnixStream> {
+        #[cfg(unix)]
+        {
+            let addr = self.take_addr()?;
+            UnixStream::connect(addr)
+                .await
+                .map_err(|e| Error::InputOutput(e.into()))
+        }
+
+        #[cfg(windows)]
+        Err(Error::Unsupported)
+    }
 }
 
 impl Display for Unix {