🧵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 `UnixStream::connect_addr` function from `SocketAddrExt` is
no longer used since neither tokio nor async-io supports it.
Instead, we use the plain `connect` functions in both environments,
since they allow for abstract names when the path is prepended
with '\0' (a null byte).

Author: germaniuss

zbus/src/address/transport/mod.rs

@@ -226,7 +226,7 @@ pub(crate) enum Stream {
 #[derive(Debug)]
 pub(crate) enum Stream {
     #[cfg(unix)]
-    Unix(tokio::net::UnixStream),
+    Unix(UnixStream),
     #[cfg(unix)]
     Unixexec(Command),
     Tcp(TcpStream),

zbus/src/address/transport/unix.rs

@@ -15,12 +15,10 @@ use super::encode_percents;
 #[cfg(not(feature = "tokio"))]
 use async_io::Async;
 
-#[cfg(target_os = "linux")]
-use std::os::linux::net::SocketAddrExt;
-#[cfg(unix)]
-use std::os::unix::net::SocketAddr;
-#[cfg(unix)]
+#[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;
 
@@ -46,26 +44,18 @@ impl Unix {
         self.path
     }
 
-    #[cfg(all(windows, not(feature = "tokio")))]
+    #[cfg(any(unix, not(feature = "tokio")))]
     fn take_addr(self) -> Result<PathBuf> {
-        // This is a `path` in case of Windows until uds_windows provides the needed API:
-        // https://github.com/haraldh/rust_uds_windows/issues/14
+        // 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),
-            UnixSocket::Dir(_) | UnixSocket::TmpDir(_) => {
-                // You can't connect to a unix:dir.
-                Err(Error::Unsupported)
-            }
-        }
-    }
-
-    #[cfg(unix)]
-    fn take_addr(self) -> Result<SocketAddr> {
-        match self.take_path() {
-            UnixSocket::File(path) => Ok(SocketAddr::from_pathname(path)?),
             #[cfg(target_os = "linux")]
             UnixSocket::Abstract(name) => {
-                Ok(SocketAddr::from_abstract_name(name.as_encoded_bytes())?)
+                use std::{ffi::OsString, os::unix::ffi::OsStringExt, path::PathBuf};
+                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.
@@ -103,42 +93,33 @@ impl Unix {
     pub(super) async fn connect(self) -> Result<Async<UnixStream>> {
         let addr = self.take_addr()?;
 
-        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??;
-
-        Async::new(stream).map_err(|e| Error::InputOutput(e.into()))
+        #[cfg(unix)]
+        let stream = Async::<UnixStream>::connect(addr).await;
+
+        #[cfg(not(unix))]
+        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(all(unix, feature = "tokio"))]
-    pub(super) async fn connect(self) -> Result<tokio::net::UnixStream> {
+    pub(super) async fn connect(self) -> Result<UnixStream> {
         let addr = self.take_addr()?;
-
-        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??;
-
-        tokio::net::UnixStream::from_std(stream).map_err(|e| Error::InputOutput(e.into()))
+        UnixStream::connect(addr)
+            .await
+            .map_err(|e| Error::InputOutput(e.into()))
     }
 }