Merge pull request #19 from rsocket/develop

Transport refinement.

Author: jjeffcaii

Cargo.toml

@@ -5,13 +5,12 @@ members = [
 "rsocket",
 # transports
 "rsocket-transport-tcp",
-"rsocket-transport-unix",
 "rsocket-transport-websocket",
 "rsocket-transport-wasm",
 # extra
 "rsocket-messaging",
 # internal
 "examples",
 "rsocket-test",
-"rsocket-benchmark",
+# "rsocket-benchmark",
 ]

examples/Cargo.toml

@@ -6,27 +6,24 @@ edition = "2018"
 publish = false
 
 [dev-dependencies]
-log = "0.4.8"
+log = "0.4.11"
 env_logger = "0.7.1"
 futures = "0.3.5"
-clap = "2.33.1"
-tokio-postgres = "0.5.4"
-redis = "0.16.0"
+clap = "2.33.3"
+tokio-postgres = "0.5.5"
+redis = "0.17.0"
 
 [dev-dependencies.rsocket_rust]
-version = "0.5.3"
+path = "../rsocket"
 
 [dev-dependencies.rsocket_rust_transport_tcp]
-version = "0.5.3"
-
-[dev-dependencies.rsocket_rust_transport_unix]
-version = "0.5.3"
+path = "../rsocket-transport-tcp"
 
 [dev-dependencies.rsocket_rust_transport_websocket]
-version = "0.5.3"
+path = "../rsocket-transport-websocket"
 
 [dev-dependencies.tokio]
-version = "0.2.21"
+version = "0.2.22"
 default-features = false
 features = ["full"]
 
@@ -53,11 +50,3 @@ 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.rs

@@ -3,19 +3,92 @@ extern crate log;
 
 use clap::{App, Arg, SubCommand};
 use rsocket_rust::prelude::*;
-use rsocket_rust_transport_tcp::{TcpClientTransport, TcpServerTransport};
-use std::error::Error;
+use rsocket_rust::transport::Connection;
+use rsocket_rust_transport_tcp::{
+    TcpClientTransport, TcpServerTransport, UnixClientTransport, UnixServerTransport,
+};
+use rsocket_rust_transport_websocket::{WebsocketClientTransport, WebsocketServerTransport};
 use std::fs;
 
+type Result<T> = rsocket_rust::Result<T>;
+
 enum RequestMode {
     FNF,
     REQUEST,
     STREAM,
     CHANNEL,
 }
 
+async fn serve<A, B>(transport: A, mtu: usize) -> Result<()>
+where
+    A: Send + Sync + ServerTransport<Item = B> + 'static,
+    B: Send + Sync + Transport + 'static,
+{
+    RSocketFactory::receive()
+        .transport(transport)
+        .fragment(mtu)
+        .acceptor(Box::new(|setup, _socket| {
+            info!("accept setup: {:?}", setup);
+            Ok(Box::new(EchoRSocket))
+            // Or you can reject setup
+            // Err(From::from("SETUP_NOT_ALLOW"))
+        }))
+        .on_start(Box::new(|| info!("+++++++ echo server started! +++++++")))
+        .serve()
+        .await
+}
+
+async fn connect<A, B>(transport: A, mtu: usize, req: Payload, mode: RequestMode) -> Result<()>
+where
+    A: Send + Sync + Transport<Conn = B> + 'static,
+    B: Send + Sync + Connection + 'static,
+{
+    let cli = RSocketFactory::connect()
+        .fragment(mtu)
+        .transport(transport)
+        .start()
+        .await?;
+
+    match mode {
+        RequestMode::FNF => {
+            cli.fire_and_forget(req).await;
+        }
+        RequestMode::STREAM => {
+            let mut results = cli.request_stream(req);
+            loop {
+                match results.next().await {
+                    Some(Ok(v)) => info!("{:?}", v),
+                    Some(Err(e)) => {
+                        error!("STREAM_RESPONSE FAILED: {:?}", e);
+                        break;
+                    }
+                    None => break,
+                }
+            }
+        }
+        RequestMode::CHANNEL => {
+            let mut results = cli.request_channel(Box::pin(futures::stream::iter(vec![Ok(req)])));
+            loop {
+                match results.next().await {
+                    Some(Ok(v)) => info!("{:?}", v),
+                    Some(Err(e)) => {
+                        error!("CHANNEL_RESPONSE FAILED: {:?}", e);
+                        break;
+                    }
+                    None => break,
+                }
+            }
+        }
+        RequestMode::REQUEST => {
+            let res = cli.request_response(req).await.expect("Request failed!");
+            info!("{:?}", res);
+        }
+    }
+    Ok(())
+}
+
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+async fn main() -> Result<()> {
     env_logger::builder().format_timestamp_millis().init();
 
     let cli = App::new("echo")
@@ -106,18 +179,25 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
                 .value_of("mtu")
                 .map(|it| it.parse().expect("Invalid mtu string!"))
                 .unwrap_or(0);
-            RSocketFactory::receive()
-                .transport(TcpServerTransport::from(addr))
-                .fragment(mtu)
-                .acceptor(Box::new(|setup, _socket| {
-                    info!("accept setup: {:?}", setup);
-                    Ok(Box::new(EchoRSocket))
-                    // Or you can reject setup
-                    // Err(From::from("SETUP_NOT_ALLOW"))
-                }))
-                .on_start(Box::new(|| info!("+++++++ echo server started! +++++++")))
-                .serve()
-                .await
+
+            if addr.starts_with("ws://") {
+                serve(WebsocketServerTransport::from(addr), mtu).await
+            } else if addr.starts_with("unix://") {
+                let addr_owned = addr.to_owned();
+                tokio::spawn(async move {
+                    let _ = serve(UnixServerTransport::from(addr_owned), mtu).await;
+                });
+                let sockfile = addr.chars().skip(7).collect::<String>();
+                // Watch signal
+                tokio::signal::ctrl_c().await?;
+                info!("ctrl-c received!");
+                if let Err(e) = std::fs::remove_file(&sockfile) {
+                    error!("remove unix sock file failed: {}", e);
+                }
+                Ok(())
+            } else {
+                serve(TcpServerTransport::from(addr), mtu).await
+            }
         }
         ("connect", Some(flags)) => {
             let mut modes: Vec<RequestMode> = vec![];
@@ -147,12 +227,6 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
                 .unwrap_or(0);
 
             let addr = flags.value_of("URL").expect("Missing URL");
-            let cli = RSocketFactory::connect()
-                .fragment(mtu)
-                .transport(TcpClientTransport::from(addr))
-                .start()
-                .await
-                .expect("Connect failed!");
             let mut bu = Payload::builder();
             if let Some(data) = flags.value_of("input") {
                 if data.starts_with("@") {
@@ -164,45 +238,14 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
                 }
             }
             let req = bu.build();
-
-            match modes.pop().unwrap_or(RequestMode::REQUEST) {
-                RequestMode::FNF => {
-                    cli.fire_and_forget(req).await;
-                }
-                RequestMode::STREAM => {
-                    let mut results = cli.request_stream(req);
-                    loop {
-                        match results.next().await {
-                            Some(Ok(v)) => info!("{:?}", v),
-                            Some(Err(e)) => {
-                                error!("STREAM_RESPONSE FAILED: {:?}", e);
-                                break;
-                            }
-                            None => break,
-                        }
-                    }
-                }
-                RequestMode::CHANNEL => {
-                    let mut results =
-                        cli.request_channel(Box::pin(futures::stream::iter(vec![Ok(req)])));
-                    loop {
-                        match results.next().await {
-                            Some(Ok(v)) => info!("{:?}", v),
-                            Some(Err(e)) => {
-                                error!("CHANNEL_RESPONSE FAILED: {:?}", e);
-                                break;
-                            }
-                            None => break,
-                        }
-                    }
-                }
-                RequestMode::REQUEST => {
-                    let res = cli.request_response(req).await.expect("Request failed!");
-                    info!("{:?}", res);
-                }
+            let mode = modes.pop().unwrap_or(RequestMode::REQUEST);
+            if addr.starts_with("ws://") {
+                connect(WebsocketClientTransport::from(addr), mtu, req, mode).await
+            } else if addr.starts_with("unix://") {
+                connect(UnixClientTransport::from(addr), mtu, req, mode).await
+            } else {
+                connect(TcpClientTransport::from(addr), mtu, req, mode).await
             }
-
-            Ok(())
         }
         _ => Ok(()),
     }

examples/echo_uds.rs

@@ -1,14 +1,14 @@
 #[macro_use]
 extern crate log;
 
-use rsocket_rust::{error::RSocketError, prelude::*};
+use rsocket_rust::prelude::*;
+use rsocket_rust::Result;
 use rsocket_rust_transport_tcp::TcpServerTransport;
-use std::error::Error;
 use std::sync::Arc;
 use tokio_postgres::{Client as PgClient, NoTls};
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+async fn main() -> Result<()> {
     env_logger::builder().format_timestamp_millis().init();
     let dao = Dao::try_new().await?;
     RSocketFactory::receive()
@@ -25,7 +25,7 @@ struct Dao {
 }
 
 impl RSocket for Dao {
-    fn request_response(&self, _: Payload) -> Mono<Result<Payload, RSocketError>> {
+    fn request_response(&self, _: Payload) -> Mono<Result<Payload>> {
         let client = self.client.clone();
         Box::pin(async move {
             let row = client
@@ -45,20 +45,17 @@ impl RSocket for Dao {
         unimplemented!()
     }
 
-    fn request_stream(&self, _: Payload) -> Flux<Result<Payload, RSocketError>> {
+    fn request_stream(&self, _: Payload) -> Flux<Result<Payload>> {
         unimplemented!()
     }
 
-    fn request_channel(
-        &self,
-        _: Flux<Result<Payload, RSocketError>>,
-    ) -> Flux<Result<Payload, RSocketError>> {
+    fn request_channel(&self, _: Flux<Result<Payload>>) -> Flux<Result<Payload>> {
         unimplemented!()
     }
 }
 
 impl Dao {
-    async fn try_new() -> Result<Dao, Box<dyn Error + Sync + Send>> {
+    async fn try_new() -> Result<Dao> {
         let (client, connection) =
             tokio_postgres::connect("host=localhost user=postgres password=postgres", NoTls)
                 .await?;