net: add SocketAddr methods to Unix sockets (#8144)

tokio/src/net/unix/listener.rs

@@ -90,7 +90,23 @@ impl UnixListener {
         #[cfg(not(any(target_os = "linux", target_os = "android")))]
         let addr = StdSocketAddr::from_pathname(path)?;
 
-        let listener = mio::net::UnixListener::bind_addr(&addr)?;
+        let addr = SocketAddr::from(addr);
+        UnixListener::bind_addr(&addr)
+    }
+
+    /// Creates a new `UnixListener` bound to the specified address.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if it is not called from within a runtime with
+    /// IO enabled.
+    ///
+    /// The runtime is usually set implicitly when this function is called
+    /// from a future driven by a tokio runtime, otherwise runtime can be set
+    /// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
+    #[track_caller]
+    pub fn bind_addr(socket_addr: &SocketAddr) -> io::Result<UnixListener> {
+        let listener = mio::net::UnixListener::bind_addr(&socket_addr.0)?;
         let io = PollEvented::new(listener)?;
         Ok(UnixListener { io })
     }

tokio/src/net/unix/stream.rs

@@ -69,6 +69,15 @@ impl UnixStream {
     /// This function will create a new Unix socket and connect to the path
     /// specified, associating the returned stream with the default event loop's
     /// handle.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if it is not called from within a runtime with
+    /// IO enabled.
+    ///
+    /// The runtime is usually set implicitly when this function is called
+    /// from a future driven by a tokio runtime, otherwise runtime can be set
+    /// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
     pub async fn connect<P>(path: P) -> io::Result<UnixStream>
     where
         P: AsRef<Path>,
@@ -86,7 +95,26 @@ impl UnixStream {
         #[cfg(not(any(target_os = "linux", target_os = "android")))]
         let addr = StdSocketAddr::from_pathname(path)?;
 
-        let stream = mio::net::UnixStream::connect_addr(&addr)?;
+        let addr = SocketAddr::from(addr);
+        UnixStream::connect_addr(&addr).await
+    }
+
+    /// Connects to the socket named by `socket_addr`.
+    ///
+    /// This function will create a new Unix socket and connect to the address
+    /// specified, associating the returned stream with the default event
+    /// loop's handle.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if it is not called from within a runtime with
+    /// IO enabled.
+    ///
+    /// The runtime is usually set implicitly when this function is called
+    /// from a future driven by a tokio runtime, otherwise runtime can be set
+    /// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
+    pub async fn connect_addr(socket_addr: &SocketAddr) -> io::Result<UnixStream> {
+        let stream = mio::net::UnixStream::connect_addr(&socket_addr.0)?;
         let stream = UnixStream::new(stream)?;
 
         poll_fn(|cx| stream.io.registration().poll_write_ready(cx)).await?;

tokio/tests/uds_stream.rs

@@ -8,6 +8,7 @@ use std::io;
 use std::os::android::net::SocketAddrExt;
 #[cfg(target_os = "linux")]
 use std::os::linux::net::SocketAddrExt;
+use std::os::unix::net::SocketAddr;
 use std::task::Poll;
 
 use tokio::io::{AsyncReadExt, AsyncWriteExt, Interest};
@@ -43,6 +44,34 @@ async fn accept_read_write() -> std::io::Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn accept_read_write_socketaddr() -> std::io::Result<()> {
+    let dir = tempfile::Builder::new()
+        .prefix("tokio-uds-tests")
+        .tempdir()
+        .unwrap();
+    let sock_path = dir.path().join("connect.sock");
+    let addr = SocketAddr::from_pathname(&sock_path)?.into();
+
+    let listener = UnixListener::bind_addr(&addr)?;
+
+    let accept = listener.accept();
+    let connect = UnixStream::connect_addr(&addr);
+    let ((mut server, _), mut client) = try_join(accept, connect).await?;
+
+    // Write to the client.
+    client.write_all(b"hello").await?;
+    drop(client);
+
+    // Read from the server.
+    let mut buf = vec![];
+    server.read_to_end(&mut buf).await?;
+    assert_eq!(&buf, b"hello");
+    let len = server.read(&mut buf).await?;
+    assert_eq!(len, 0);
+    Ok(())
+}
+
 #[tokio::test]
 async fn shutdown() -> std::io::Result<()> {
     let dir = tempfile::Builder::new()