feat(transport): add support for uds, unix domain socket (#2218)

* feat(transport): add support for uds, unix domain socket (#2218)

Previously the uds support lives as an example in the `example/src/uds` folder.
Endpoint is refactored to support multiple endpoint types, including Uri and
Uds.

The supported unix domain socket URI follows RFC-3986 which is aligned with
the gRPC naming convention.
- unix:relative_path
- unix:///absolute_path

References:
- https://datatracker.ietf.org/doc/html/rfc3986
- https://github.com/grpc/grpc/blob/master/doc/naming.md

* fix feature flag error

* fix windows build

* fix windows build 2

* fix windows build 3

* fix windows build 4

* fix windows build 5

---------

Co-authored-by: Lucio Franco <[email protected]>

Author: adamcavendish

examples/Cargo.toml

@@ -118,8 +118,13 @@ path = "src/tracing/server.rs"
 required-features = ["tracing"]
 
 [[bin]]
-name = "uds-client"
-path = "src/uds/client.rs"
+name = "uds-client-standard"
+path = "src/uds/client_standard.rs"
+required-features = ["uds"]
+
+[[bin]]
+name = "uds-client-with-connector"
+path = "src/uds/client_with_connector.rs"
 required-features = ["uds"]
 
 [[bin]]

examples/src/uds/client_standard.rs

@@ -0,0 +1,34 @@
+#![cfg_attr(not(unix), allow(unused_imports))]
+
+pub mod hello_world {
+    tonic::include_proto!("helloworld");
+}
+
+use hello_world::{greeter_client::GreeterClient, HelloRequest};
+
+#[cfg(unix)]
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Unix socket URI follows [RFC-3986](https://datatracker.ietf.org/doc/html/rfc3986)
+    // which is aligned with [the gRPC naming convention](https://github.com/grpc/grpc/blob/master/doc/naming.md).
+    // - unix:relative_path
+    // - unix:///absolute_path
+    let path = "unix:///tmp/tonic/helloworld";
+
+    let mut client = GreeterClient::connect(path).await?;
+
+    let request = tonic::Request::new(HelloRequest {
+        name: "Tonic".into(),
+    });
+
+    let response = client.say_hello(request).await?;
+
+    println!("RESPONSE={:?}", response);
+
+    Ok(())
+}
+
+#[cfg(not(unix))]
+fn main() {
+    panic!("The `uds` example only works on unix systems!");
+}

examples/src/uds/client.rs renamed to examples/src/uds/client_with_connector.rs

@@ -8,6 +8,7 @@ name = "default_stubs"
 tokio = {version = "1.0", features = ["macros", "rt-multi-thread", "net"]}
 tokio-stream = {version = "0.1", features = ["net"]}
 prost = "0.13"
+rand = "0.8"
 tonic = {path = "../../tonic"}
 
 [build-dependencies]

tests/default_stubs/Cargo.toml

@@ -1,25 +1,28 @@
 #![allow(unused_imports)]
 
+use crate::test_client::TestClient;
 use crate::*;
+use rand::Rng as _;
+use std::env;
+use std::fs;
 use std::net::SocketAddr;
 use tokio::net::TcpListener;
+use tonic::transport::Channel;
 use tonic::transport::Server;
 
 #[cfg(test)]
 fn echo_requests_iter() -> impl Stream<Item = ()> {
     tokio_stream::iter(1..usize::MAX).map(|_| ())
 }
 
-#[tokio::test()]
-async fn test_default_stubs() {
+#[cfg(test)]
+async fn test_default_stubs(
+    mut client: TestClient<Channel>,
+    mut client_default_stubs: TestClient<Channel>,
+) {
     use tonic::Code;
 
-    let addrs = run_services_in_background().await;
-
     // First validate pre-existing functionality (trait has no default implementation, we explicitly return PermissionDenied in lib.rs).
-    let mut client = test_client::TestClient::connect(format!("http://{}", addrs.0))
-        .await
-        .unwrap();
     assert_eq!(
         client.unary(()).await.unwrap_err().code(),
         Code::PermissionDenied
@@ -46,9 +49,6 @@ async fn test_default_stubs() {
     );
 
     // Then validate opt-in new functionality (trait has default implementation of returning Unimplemented).
-    let mut client_default_stubs = test_client::TestClient::connect(format!("http://{}", addrs.1))
-        .await
-        .unwrap();
     assert_eq!(
         client_default_stubs.unary(()).await.unwrap_err().code(),
         Code::Unimplemented
@@ -79,6 +79,27 @@ async fn test_default_stubs() {
     );
 }
 
+#[tokio::test()]
+async fn test_default_stubs_tcp() {
+    let addrs = run_services_in_background().await;
+    let client = test_client::TestClient::connect(format!("http://{}", addrs.0))
+        .await
+        .unwrap();
+    let client_default_stubs = test_client::TestClient::connect(format!("http://{}", addrs.1))
+        .await
+        .unwrap();
+    test_default_stubs(client, client_default_stubs).await;
+}
+
+#[tokio::test()]
+#[cfg(not(target_os = "windows"))]
+async fn test_default_stubs_uds() {
+    let addrs = run_services_in_background_uds().await;
+    let client = test_client::TestClient::connect(addrs.0).await.unwrap();
+    let client_default_stubs = test_client::TestClient::connect(addrs.1).await.unwrap();
+    test_default_stubs(client, client_default_stubs).await;
+}
+
 #[cfg(test)]
 async fn run_services_in_background() -> (SocketAddr, SocketAddr) {
     let svc = test_server::TestServer::new(Svc {});
@@ -110,3 +131,48 @@ async fn run_services_in_background() -> (SocketAddr, SocketAddr) {
 
     (addr, addr_default_stubs)
 }
+
+#[cfg(all(test, not(target_os = "windows")))]
+async fn run_services_in_background_uds() -> (String, String) {
+    use tokio::net::UnixListener;
+
+    let svc = test_server::TestServer::new(Svc {});
+    let svc_default_stubs = test_default_server::TestDefaultServer::new(Svc {});
+
+    let mut rng = rand::thread_rng();
+    let suffix: String = (0..8)
+        .map(|_| rng.sample(rand::distributions::Alphanumeric) as char)
+        .collect();
+    let tmpdir = fs::canonicalize(env::temp_dir())
+        .unwrap()
+        .join(format!("tonic_test_{}", suffix));
+    fs::create_dir(&tmpdir).unwrap();
+
+    let uds_filepath = tmpdir.join("impl.sock").to_str().unwrap().to_string();
+    let listener = UnixListener::bind(uds_filepath.as_str()).unwrap();
+    let uds_addr = format!("unix://{}", uds_filepath);
+
+    let uds_default_stubs_filepath = tmpdir.join("stub.sock").to_str().unwrap().to_string();
+    let listener_default_stubs = UnixListener::bind(uds_default_stubs_filepath.as_str()).unwrap();
+    let uds_default_stubs_addr = format!("unix://{}", uds_default_stubs_filepath);
+
+    tokio::spawn(async move {
+        Server::builder()
+            .add_service(svc)
+            .serve_with_incoming(tokio_stream::wrappers::UnixListenerStream::new(listener))
+            .await
+            .unwrap();
+    });
+
+    tokio::spawn(async move {
+        Server::builder()
+            .add_service(svc_default_stubs)
+            .serve_with_incoming(tokio_stream::wrappers::UnixListenerStream::new(
+                listener_default_stubs,
+            ))
+            .await
+            .unwrap();
+    });
+
+    (uds_addr, uds_default_stubs_addr)
+}