♻️ zb: Move UnixSocket connection code to unix.rs

Move UnixSocket connection code to unix.rs from mod.rs
for better readability and maintainability.

Author: germaniuss

zbus/src/address/transport/mod.rs

@@ -23,20 +23,13 @@ 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,
 };
 
 mod unix;
-pub use unix::{Unix, UnixSocket};
+pub use unix::{Unix, UnixSocket, UnixStream};
 mod tcp;
 pub use tcp::{Tcp, TcpTransportFamily};
 #[cfg(windows)]
@@ -103,62 +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` because neither uds_windows, tokio, nor async_io provide
-                // the SocketAddrExt functions.
-                let addr = match unix.take_path() {
-                    UnixSocket::File(path) => path,
-                    #[cfg(target_os = "linux")]
-                    UnixSocket::Abstract(name) => {
-                        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);
-                    }
-                };
-
-                #[cfg(not(feature = "tokio"))]
-                {
-                    #[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()))
-                }
-
-                #[cfg(feature = "tokio")]
-                {
-                    #[cfg(unix)]
-                    {
-                        UnixStream::connect(addr)
-                            .await
-                            .map_err(|e| Error::InputOutput(e.into()))
-                    }
-
-                    #[cfg(not(unix))]
-                    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> {
+        // 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::{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.
+                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(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(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(not(unix))]
+        Err(Error::Unsupported)
+    }
 }
 
 impl Display for Unix {