♻️ 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

@@ -12,21 +12,24 @@ 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)]
 mod unixexec;
 #[cfg(unix)]
 pub use unixexec::Unixexec;
 
+#[cfg(all(unix, not(feature = "tokio")))]
+use std::os::unix::net::UnixStream;
+#[cfg(all(unix, feature = "tokio"))]
+use tokio::net::UnixStream;
+#[cfg(windows)]
+use uds_windows::UnixStream;
+
 use std::{
     fmt::{Display, Formatter},
     str::from_utf8_unchecked,
@@ -55,8 +58,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"
@@ -103,38 +104,44 @@ impl Transport {
     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
+                // This is a `path` because neither uds_windows, tokio, nor async_io provide
+                // the SocketAddrExt functions.
                 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())?
+                        use std::{ffi::OsString, os::unix::ffi::OsStringExt, path::PathBuf};
+                        let mut v = name.into_vec();
+                        v.insert(0, 0);
+                        PathBuf::from(OsString::from_vec(v))
                     }
                     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)
+                    #[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(Stream::Unix)
                         .map_err(|e| Error::InputOutput(e.into()))
                 }
@@ -143,16 +150,13 @@ impl Transport {
                 {
                     #[cfg(unix)]
                     {
-                        tokio::net::UnixStream::from_std(stream)
-                            .map(Stream::Unix)
+                        UnixStream::connect(addr)
+                            .await
                             .map_err(|e| Error::InputOutput(e.into()))
                     }
 
                     #[cfg(not(unix))]
-                    {
-                        let _ = stream;
-                        Err(Error::Unsupported)
-                    }
+                    Err(Error::Unsupported)
                 }
             }
             #[cfg(unix)]