Merge pull request #16 from seal90/develop

support for unix domain socket

Author: jjeffcaii

.gitignore

@@ -1,5 +1,5 @@
 target/
 **/*.rs.bk
 Cargo.lock
-
+.vscode/
 

Cargo.toml

@@ -5,6 +5,7 @@ members = [
 "rsocket",
 # transports
 "rsocket-transport-tcp",
+"rsocket-transport-unix",
 "rsocket-transport-websocket",
 "rsocket-transport-wasm",
 # extra

examples/Cargo.toml

@@ -19,6 +19,9 @@ path = "../rsocket"
 [dev-dependencies.rsocket_rust_transport_tcp]
 path = "../rsocket-transport-tcp"
 
+[dev-dependencies.rsocket_rust_transport_unix]
+path = "../rsocket-transport-unix"
+
 [dev-dependencies.rsocket_rust_transport_websocket]
 version = "0.5.2"
 path = "../rsocket-transport-websocket"
@@ -51,3 +54,11 @@ path = "postgres.rs"
 [[example]]
 name = "redis"
 path = "redis.rs"
+
+[[example]]
+name = "echo-uds"
+path = "echo-uds.rs"
+
+[[example]]
+name = "echo-uds-client"
+path = "echo-uds-client.rs"

examples/echo-uds-client.rs

@@ -0,0 +1,32 @@
+use log::info;
+use rsocket_rust::prelude::{ClientResponder, EchoRSocket, Payload, RSocket, RSocketFactory};
+use rsocket_rust_transport_unix::UnixClientTransport;
+use std::error::Error;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    
+    let responder: ClientResponder = Box::new(|| Box::new(EchoRSocket));
+
+    let client = RSocketFactory::connect()
+        .acceptor(responder)
+        .transport(UnixClientTransport::from("/tmp/rsocket-uds.sock"))
+        .setup(Payload::from("READY!"))
+        .mime_type("text/plain", "text/plain")
+        .start()
+        .await
+        .unwrap();
+
+    let request_payload: Payload = Payload::builder()
+        .set_data_utf8("Hello World!")
+        .set_metadata_utf8("Rust")
+        .build();
+
+    let res = client.request_response(request_payload).await.unwrap();
+
+    info!("got: {:?}", res);
+
+    client.close();
+
+    Ok(())
+}

examples/echo-uds.rs

@@ -0,0 +1,29 @@
+use log::info;
+use rsocket_rust::prelude::{EchoRSocket, RSocketFactory, ServerResponder};
+use rsocket_rust_transport_unix::UnixServerTransport;
+use std::error::Error;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+
+    let transport: UnixServerTransport = UnixServerTransport::from("/tmp/rsocket-uds.sock");
+
+    let responder: ServerResponder = Box::new(|setup, _socket| {
+        info!("accept setup: {:?}", setup);
+        Ok(Box::new(EchoRSocket))
+        // Or you can reject setup
+        // Err(From::from("SETUP_NOT_ALLOW"))
+    });
+
+    let on_start: Box<dyn FnMut() + Send + Sync> =
+        Box::new(|| info!("+++++++ echo server started! +++++++"));
+
+    RSocketFactory::receive()
+        .transport(transport)
+        .acceptor(responder)
+        .on_start(on_start)
+        .serve()
+        .await?;
+
+    Ok(())
+}

rsocket-test/Cargo.toml

@@ -22,6 +22,9 @@ features = ["frame"]
 [dev-dependencies.rsocket_rust_transport_tcp]
 path = "../rsocket-transport-tcp"
 
+[dev-dependencies.rsocket_rust_transport_unix]
+path = "../rsocket-transport-unix"
+
 [dev-dependencies.rsocket_rust_transport_websocket]
 path = "../rsocket-transport-websocket"
 

rsocket-test/tests/test_clients.rs

@@ -4,6 +4,7 @@ extern crate log;
 use futures::stream;
 use rsocket_rust::prelude::*;
 use rsocket_rust_transport_tcp::{TcpClientTransport, TcpServerTransport};
+use rsocket_rust_transport_unix::{UnixClientTransport, UnixServerTransport};
 use rsocket_rust_transport_websocket::{WebsocketClientTransport, WebsocketServerTransport};
 use std::thread::sleep;
 use std::time::Duration;
@@ -120,6 +121,52 @@ fn test_tcp() {
     });
 }
 
+#[test]
+fn test_unix() {
+    init();
+
+    let addr = "/tmp/rsocket-uds.sock";
+
+    let server_runtime = Runtime::new().unwrap();
+
+    // spawn a server
+    server_runtime.spawn(async move {
+        RSocketFactory::receive()
+            .transport(UnixServerTransport::from(addr))
+            .acceptor(Box::new(|setup, _socket| {
+                info!("accept setup: {:?}", setup);
+                Ok(Box::new(EchoRSocket))
+            }))
+            .on_start(Box::new(|| {
+                info!("+++++++ unix echo server started! +++++++")
+            }))
+            .serve()
+            .await
+    });
+
+    sleep(Duration::from_millis(500));
+
+    let mut client_runtime = Runtime::new().unwrap();
+
+    client_runtime.block_on(async {
+        let cli = RSocketFactory::connect()
+            .acceptor(Box::new(|| Box::new(EchoRSocket)))
+            .transport(UnixClientTransport::from(addr))
+            .setup(Payload::from("READY!"))
+            .mime_type("text/plain", "text/plain")
+            .start()
+            .await
+            .unwrap();
+
+        exec_metadata_push(&cli).await;
+        exec_fire_and_forget(&cli).await;
+        exec_request_response(&cli).await;
+        exec_request_stream(&cli).await;
+        exec_request_channel(&cli).await;
+        cli.close();
+    });
+}
+
 #[tokio::main]
 #[test]
 #[ignore]

rsocket-transport-unix/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "rsocket_rust_transport_unix"
+version = "0.5.3"
+authors = ["seal <[email protected]>"]
+edition = "2018"
+license = "Apache-2.0"
+readme = "README.md"
+repository = "https://github.com/rsocket/rsocket-rust"
+homepage = "https://github.com/rsocket/rsocket-rust"
+description = "Unix Domain Socket RSocket transport implementation."
+
+[dependencies]
+log = "0.4.8"
+futures = "0.3.5"
+bytes = "0.5.4"
+
+[dependencies.rsocket_rust]
+path = "../rsocket"
+features = ["frame"]
+
+[dependencies.tokio]
+version = "0.2.21"
+default-features = false
+features = [ "rt-core", "rt-threaded", "uds", "sync", "stream" ]
+
+[dependencies.tokio-util]
+version = "0.3.1"
+default-features = false
+features = ["codec"]

rsocket-transport-unix/README.md

@@ -0,0 +1,85 @@
+# RSocket Transport For Unix Domain Socket
+
+## Example
+
+Add dependencies in your `Cargo.toml`.
+
+```toml
+[dependencies]
+tokio = "0.2.21"
+rsocket_rust = "0.5.3"
+rsocket_rust_transport_unix = "0.5.3"
+```
+
+### Server
+
+```rust
+use log::info;
+use rsocket_rust::prelude::{EchoRSocket, RSocketFactory, ServerResponder};
+use rsocket_rust_transport_unix::UnixServerTransport;
+use std::error::Error;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+
+    let transport: UnixServerTransport = UnixServerTransport::from("/tmp/rsocket-uds.sock");
+
+    let responder: ServerResponder = Box::new(|setup, _socket| {
+        info!("accept setup: {:?}", setup);
+        Ok(Box::new(EchoRSocket))
+        // Or you can reject setup
+        // Err(From::from("SETUP_NOT_ALLOW"))
+    });
+
+    let on_start: Box<dyn FnMut() + Send + Sync> =
+        Box::new(|| info!("+++++++ echo server started! +++++++"));
+
+    RSocketFactory::receive()
+        .transport(transport)
+        .acceptor(responder)
+        .on_start(on_start)
+        .serve()
+        .await?;
+
+    Ok(())
+}
+
+```
+
+### Client
+
+```rust
+use log::info;
+use rsocket_rust::prelude::{ClientResponder, EchoRSocket, Payload, RSocket, RSocketFactory};
+use rsocket_rust_transport_unix::UnixClientTransport;
+use std::error::Error;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    
+    let responder: ClientResponder = Box::new(|| Box::new(EchoRSocket));
+
+    let client = RSocketFactory::connect()
+        .acceptor(responder)
+        .transport(UnixClientTransport::from("/tmp/rsocket-uds.sock"))
+        .setup(Payload::from("READY!"))
+        .mime_type("text/plain", "text/plain")
+        .start()
+        .await
+        .unwrap();
+
+    let request_payload: Payload = Payload::builder()
+        .set_data_utf8("Hello World!")
+        .set_metadata_utf8("Rust")
+        .build();
+
+    let res = client.request_response(request_payload).await.unwrap();
+
+    info!("got: {:?}", res);
+
+    client.close();
+
+    Ok(())
+}
+
+```