Add windows support on async

Add support for windows named pipes to the async codebase.
This refactor the codes to use a transport abstraction instead of file descriptors.

Signed-off-by: Jorge Prendes <jorge.prendes@gmail.com>

Author: jprendes

Cargo.toml

@@ -19,6 +19,7 @@ log = "0.4"
 byteorder = "1.3.2"
 thiserror = "1.0"
 async-trait = { version = "0.1.31", optional = true }
+async-stream = { version = "0.3.6", optional = true }
 tokio = { version = "1", features = ["rt", "sync", "io-util", "macros", "time"], optional = true }
 futures = { version = "0.3", optional = true }
 crossbeam = "0.8.0"
@@ -27,7 +28,7 @@ crossbeam = "0.8.0"
 windows-sys = {version = "0.48", features = [ "Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_IO", "Win32_System_Pipes", "Win32_Security", "Win32_System_Threading"]}
 
 [target.'cfg(any(target_os = "linux", target_os = "android"))'.dependencies]
-tokio-vsock = { version = "0.4.0", optional = true }
+tokio-vsock = { version = "0.7.0", optional = true }
 
 [build-dependencies]
 # lock home to avoid conflict with latest version
@@ -36,7 +37,7 @@ protobuf-codegen = "3.1.0"
 
 [features]
 default = ["sync"]
-async = ["async-trait", "tokio", "futures", "tokio-vsock"]
+async = ["async-trait", "async-stream", "tokio", "futures", "tokio-vsock"]
 sync = []
 
 [package.metadata.docs.rs]

Makefile

@@ -21,12 +21,7 @@ build: debug
 
 .PHONY: test
 test:
-ifeq ($OS,Windows_NT)
-	# async isn't enabled for windows, don't test that feature
-	cargo test --verbose
-else
 	cargo test --all-features --verbose
-endif
 
 .PHONY: check
 check:

example/async-client.rs

@@ -5,21 +5,13 @@
 
 mod protocols;
 mod utils;
-#[cfg(unix)]
 use protocols::asynchronous::{agent, agent_ttrpc, health, health_ttrpc};
 use ttrpc::context::{self, Context};
-#[cfg(unix)]
 use ttrpc::r#async::Client;
 
-#[cfg(windows)]
-fn main() {
-    println!("This example only works on Unix-like OSes");
-}
-
-#[cfg(unix)]
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
-    let c = Client::connect(utils::SOCK_ADDR).unwrap();
+    let c = Client::connect(utils::SOCK_ADDR).await.unwrap();
     let hc = health_ttrpc::HealthClient::new(c.clone());
     let ac = agent_ttrpc::AgentServiceClient::new(c);
 

example/async-server.rs

@@ -10,20 +10,14 @@ use std::sync::Arc;
 
 use log::LevelFilter;
 
-#[cfg(unix)]
 use protocols::asynchronous::{agent, agent_ttrpc, health, health_ttrpc};
-#[cfg(unix)]
 use ttrpc::asynchronous::Server;
 
-#[cfg(unix)]
 use async_trait::async_trait;
-#[cfg(unix)]
-use tokio::signal::unix::{signal, SignalKind};
 use tokio::time::sleep;
 
 struct HealthService;
 
-#[cfg(unix)]
 #[async_trait]
 impl health_ttrpc::Health for HealthService {
     async fn check(
@@ -46,7 +40,6 @@ impl health_ttrpc::Health for HealthService {
 }
 
 struct AgentService;
-#[cfg(unix)]
 #[async_trait]
 impl agent_ttrpc::AgentService for AgentService {
     async fn list_interfaces(
@@ -58,12 +51,6 @@ impl agent_ttrpc::AgentService for AgentService {
     }
 }
 
-#[cfg(windows)]
-fn main() {
-    println!("This example only works on Unix-like OSes");
-}
-
-#[cfg(unix)]
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
     simple_logging::log_to_stderr(LevelFilter::Trace);
@@ -79,12 +66,10 @@ async fn main() {
         .register_service(hservice)
         .register_service(aservice);
 
-    let mut hangup = signal(SignalKind::hangup()).unwrap();
-    let mut interrupt = signal(SignalKind::interrupt()).unwrap();
     server.start().await.unwrap();
 
     tokio::select! {
-        _ = hangup.recv() => {
+        _ = utils::hangup() => {
             // test stop_listen -> start
             println!("stop listen");
             server.stop_listen().await;
@@ -94,7 +79,7 @@ async fn main() {
             // hold some time for the new test connection.
             sleep(std::time::Duration::from_secs(100)).await;
         }
-        _ = interrupt.recv() => {
+        _ = utils::interrupt() => {
             // test graceful shutdown
             println!("graceful shutdown");
             server.shutdown().await.unwrap();

example/async-stream-client.rs

@@ -5,23 +5,15 @@
 
 mod protocols;
 mod utils;
-#[cfg(unix)]
 use protocols::asynchronous::{empty, streaming, streaming_ttrpc};
 use ttrpc::context::{self, Context};
-#[cfg(unix)]
 use ttrpc::r#async::Client;
 
-#[cfg(windows)]
-fn main() {
-    println!("This example only works on Unix-like OSes");
-}
-
-#[cfg(unix)]
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
     simple_logging::log_to_stderr(log::LevelFilter::Info);
 
-    let c = Client::connect(utils::SOCK_ADDR).unwrap();
+    let c = Client::connect(utils::SOCK_ADDR).await.unwrap();
     let sc = streaming_ttrpc::StreamingClient::new(c);
 
     let _now = std::time::Instant::now();
@@ -75,7 +67,6 @@ fn default_ctx() -> Context {
     ctx
 }
 
-#[cfg(unix)]
 async fn echo_request(cli: streaming_ttrpc::StreamingClient) {
     let echo1 = streaming::EchoPayload {
         seq: 1,
@@ -87,7 +78,6 @@ async fn echo_request(cli: streaming_ttrpc::StreamingClient) {
     assert_eq!(resp.seq, echo1.seq + 1);
 }
 
-#[cfg(unix)]
 async fn echo_stream(cli: streaming_ttrpc::StreamingClient) {
     let mut stream = cli.echo_stream(default_ctx()).await.unwrap();
 
@@ -110,7 +100,6 @@ async fn echo_stream(cli: streaming_ttrpc::StreamingClient) {
     assert!(matches!(ret, Err(ttrpc::Error::Eof)));
 }
 
-#[cfg(unix)]
 async fn sum_stream(cli: streaming_ttrpc::StreamingClient) {
     let mut stream = cli.sum_stream(default_ctx()).await.unwrap();
 
@@ -138,7 +127,6 @@ async fn sum_stream(cli: streaming_ttrpc::StreamingClient) {
     assert_eq!(ssum.num, sum.num);
 }
 
-#[cfg(unix)]
 async fn divide_stream(cli: streaming_ttrpc::StreamingClient) {
     let expected = streaming::Sum {
         sum: 392,
@@ -158,7 +146,6 @@ async fn divide_stream(cli: streaming_ttrpc::StreamingClient) {
     assert_eq!(actual.num, expected.num);
 }
 
-#[cfg(unix)]
 async fn echo_null(cli: streaming_ttrpc::StreamingClient) {
     let mut stream = cli.echo_null(default_ctx()).await.unwrap();
 
@@ -174,7 +161,6 @@ async fn echo_null(cli: streaming_ttrpc::StreamingClient) {
     assert_eq!(res, empty::Empty::new());
 }
 
-#[cfg(unix)]
 async fn echo_null_stream(cli: streaming_ttrpc::StreamingClient) {
     let stream = cli.echo_null_stream(default_ctx()).await.unwrap();
 
@@ -206,7 +192,6 @@ async fn echo_null_stream(cli: streaming_ttrpc::StreamingClient) {
         .unwrap();
 }
 
-#[cfg(unix)]
 async fn echo_default_value(cli: streaming_ttrpc::StreamingClient) {
     let mut stream = cli
         .echo_default_value(default_ctx(), &Default::default()) // send default value to verify #208
@@ -219,7 +204,6 @@ async fn echo_default_value(cli: streaming_ttrpc::StreamingClient) {
     assert_eq!(received.msg, "");
 }
 
-#[cfg(unix)]
 async fn server_send_stream(cli: streaming_ttrpc::StreamingClient) {
     let mut stream = cli
         .server_send_stream(default_ctx(), &Default::default())

example/async-stream-server.rs

@@ -10,20 +10,14 @@ use std::sync::Arc;
 
 use log::{info, LevelFilter};
 
-#[cfg(unix)]
 use protocols::asynchronous::{empty, streaming, streaming_ttrpc};
-#[cfg(unix)]
 use ttrpc::{asynchronous::Server, Error};
 
-#[cfg(unix)]
 use async_trait::async_trait;
-#[cfg(unix)]
-use tokio::signal::unix::{signal, SignalKind};
 use tokio::time::sleep;
 
 struct StreamingService;
 
-#[cfg(unix)]
 #[async_trait]
 impl streaming_ttrpc::Streaming for StreamingService {
     async fn echo(
@@ -173,12 +167,6 @@ impl streaming_ttrpc::Streaming for StreamingService {
     }
 }
 
-#[cfg(windows)]
-fn main() {
-    println!("This example only works on Unix-like OSes");
-}
-
-#[cfg(unix)]
 #[tokio::main(flavor = "current_thread")]
 async fn main() {
     simple_logging::log_to_stderr(LevelFilter::Info);
@@ -190,12 +178,10 @@ async fn main() {
         .unwrap()
         .register_service(service);
 
-    let mut hangup = signal(SignalKind::hangup()).unwrap();
-    let mut interrupt = signal(SignalKind::interrupt()).unwrap();
     server.start().await.unwrap();
 
     tokio::select! {
-        _ = hangup.recv() => {
+        _ = utils::hangup() => {
             // test stop_listen -> start
             info!("stop listen");
             server.stop_listen().await;
@@ -205,7 +191,7 @@ async fn main() {
             // hold some time for the new test connection.
             sleep(std::time::Duration::from_secs(100)).await;
         }
-        _ = interrupt.recv() => {
+        _ = utils::interrupt() => {
             // test graceful shutdown
             info!("graceful shutdown");
             server.shutdown().await.unwrap();

example/protocols/mod.rs

@@ -2,6 +2,5 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 //
-#[cfg(unix)]
 pub mod asynchronous;
 pub mod sync;

example/utils.rs

@@ -65,7 +65,6 @@ pub mod resp {
         }
     }
 
-    #[cfg(unix)]
     pub mod asynchronous {
         use crate::protocols::asynchronous::{
             agent::Interfaces, health::VersionCheckResponse, types::Interface,
@@ -96,3 +95,19 @@ pub mod resp {
         }
     }
 }
+
+pub async fn hangup() {
+    #[cfg(unix)]
+    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup())
+        .unwrap()
+        .recv()
+        .await
+        .unwrap();
+
+    #[cfg(not(unix))]
+    std::future::pending::<()>().await;
+}
+
+pub async fn interrupt() {
+    tokio::signal::ctrl_c().await.unwrap();
+}

src/asynchronous/client.rs

@@ -6,14 +6,14 @@
 
 use std::collections::HashMap;
 use std::convert::TryInto;
+#[cfg(unix)]
 use std::os::unix::io::RawFd;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, Mutex};
 
 use async_trait::async_trait;
 use tokio::{self, sync::mpsc, task};
 
-use crate::common::client_connect;
 use crate::error::{get_rpc_status, Error, Result};
 use crate::proto::{
     Code, Codec, GenMessage, Message, MessageHeader, Request, Response, FLAG_NO_DATA,
@@ -24,9 +24,9 @@ use crate::r#async::shutdown;
 use crate::r#async::stream::{
     Kind, MessageReceiver, MessageSender, ResultReceiver, ResultSender, StreamInner,
 };
-use crate::r#async::utils;
 
 use super::stream::SendingMessage;
+use super::transport::Socket;
 
 /// A ttrpc Client (async).
 #[derive(Clone)]
@@ -37,15 +37,23 @@ pub struct Client {
 }
 
 impl Client {
-    pub fn connect(sockaddr: &str) -> Result<Client> {
-        let fd = unsafe { client_connect(sockaddr)? };
-        Ok(Self::new(fd))
+    pub async fn connect(sockaddr: &str) -> Result<Client> {
+        let socket = Socket::connect(sockaddr)
+            .await
+            .map_err(err_to_others_err!(e, "Socket::connect error "))?;
+        Ok(Self::new(socket))
     }
 
-    /// Initialize a new [`Client`].
-    pub fn new(fd: RawFd) -> Client {
-        let stream = utils::new_unix_stream_from_raw_fd(fd);
+    #[cfg(unix)]
+    /// # Safety
+    /// The file descriptor must represent a unix socket.
+    pub unsafe fn from_raw_unix_socket_fd(fd: RawFd) -> Client {
+        let stream = unsafe { Socket::from_raw_unix_socket_fd(fd) }.unwrap();
+        Self::new(stream)
+    }
 
+    /// Initialize a new [`Client`].
+    pub fn new(stream: Socket) -> Client {
         let (req_tx, rx): (MessageSender, MessageReceiver) = mpsc::channel(100);
 
         let req_map = Arc::new(Mutex::new(HashMap::new()));