Merge pull request #2 from rsocket/feature/async_await

Feature/async await

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rsocket_rust"
-version = "0.1.5"
+version = "0.2.0"
 authors = ["Jeffsky <jjeffcaii@outlook.com>"]
 edition = "2018"
 license = "Apache-2.0"
@@ -10,19 +10,32 @@ homepage = "https://github.com/rsocket/rsocket-rust"
 description = "rsocket-rust is an implementation of the RSocket protocol in Rust."
 
 [dependencies]
+matches = "0.1.8"
 log = "0.4.8"
-bytes = "0.4.12"
-futures = "0.1.29"
-tokio = "0.1.22"
+bytes = "0.5.2"
+futures = "0.3.1"
+lazy_static = "1.4.0"
+# reactor_rs = {git = "https://github.com/jjeffcaii/reactor-rust", branch = "develop"}
+
+[dependencies.tokio]
+version = "0.2.1"
+default-features = false
+features = ["full"]
+
+[dependencies.tokio-util]
+version = "0.2.0"
+default-features = false
+features = ["codec"]
 
 [dev-dependencies]
-env_logger = "0.6.2"
-hex = "0.3.2"
+env_logger = "0.7.1"
+hex = "0.4.0"
+rand = "0.7.2"
 
 [[example]]
 name = "echo"
 path = "examples/echo/main.rs"
 
-[[example]]
-name = "proxy"
-path = "examples/proxy/main.rs"
+# [[example]]
+# name = "proxy"
+# path = "examples/proxy/main.rs"

README.md

@@ -5,7 +5,7 @@
 [![License](https://img.shields.io/github/license/rsocket/rsocket-rust.svg)](https://github.com/rsocket/rsocket-rust/blob/master/LICENSE)
 [![GitHub Release](https://img.shields.io/github/release-pre/rsocket/rsocket-rust.svg)](https://github.com/rsocket/rsocket-rust/releases)
 
-> rsocket-rust is an implementation of the RSocket protocol in Rust.
+> rsocket-rust is an implementation of the RSocket protocol in Rust(1.39+).
 It's an **alpha** version and still under active development. **Do not use it in a production environment!**
 
 ## Example
@@ -15,66 +15,57 @@ It's an **alpha** version and still under active development. **Do not use it in
 ### Server
 
 ```rust
-extern crate bytes;
-extern crate futures;
 extern crate rsocket_rust;
-
-use bytes::Bytes;
-use futures::prelude::*;
+extern crate tokio;
+#[macro_use]
+extern crate log;
 use rsocket_rust::prelude::*;
-
-#[test]
-fn test_serve() {
-  RSocketFactory::receive()
-    .transport(URI::Tcp("127.0.0.1:7878"))
-    .acceptor(|setup, sending_socket| {
-      println!("accept setup: {:?}", setup);
-      // TODO: use tokio runtime?
-      std::thread::spawn(move || {
-        let resp = sending_socket
-          .request_response(
-            Payload::builder()
-              .set_data(Bytes::from("Hello Client!"))
-              .build(),
-          )
-          .wait()
-          .unwrap();
-        println!(">>>>> response success: {:?}", resp);
-      });
-      Box::new(MockResponder)
-    })
-    .serve()
-    .wait()
-    .unwrap();
+use std::env;
+use std::error::Error;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    env_logger::builder().init();
+    let addr = env::args().nth(1).unwrap_or("127.0.0.1:7878".to_string());
+
+    RSocketFactory::receive()
+        .transport(URI::Tcp(addr))
+        .acceptor(|setup, _socket| {
+            info!("accept setup: {:?}", setup);
+            Box::new(EchoRSocket)
+        })
+        .serve()
+        .await
 }
-
 ```
 
 ### Client
 
 ```rust
-extern crate futures;
 extern crate rsocket_rust;
 
-use futures::prelude::*;
 use rsocket_rust::prelude::*;
 
+#[tokio::main]
 #[test]
-fn test_client() {
-  let cli = RSocketFactory::connect()
-    .acceptor(||Box::new(MockResponder))
-    .transport(URI::Tcp("127.0.0.1:7878"))
-    .setup(Payload::from("READY!"))
-    .mime_type("text/plain", "text/plain")
-    .start()
-    .unwrap();
-  let pa = Payload::builder()
-    .set_data_utf8("Hello World!")
-    .set_metadata_utf8("Rust!")
-    .build();
-  let resp = cli.request_response(pa).wait().unwrap();
-  println!("******* response: {:?}", resp);
+async fn test() {
+    let cli = RSocketFactory::connect()
+        .acceptor(|| Box::new(EchoRSocket))
+        .transport(URI::Tcp("127.0.0.1:7878".to_string()))
+        .setup(Payload::from("READY!"))
+        .mime_type("text/plain", "text/plain")
+        .start()
+        .await
+        .unwrap();
+    let req = Payload::builder()
+        .set_data_utf8("Hello World!")
+        .set_metadata_utf8("Rust")
+        .build();
+    let res = cli.request_response(req).await.unwrap();
+    println!("got: {:?}", res);
+    cli.close();
 }
+
 ```
 
 ## Dependencies

examples/echo/main.rs

@@ -1,42 +1,22 @@
-#![allow(unused_variables)]
-#![allow(unused_imports)]
-#![allow(dead_code)]
-
-#[macro_use]
-extern crate log;
-extern crate bytes;
-extern crate env_logger;
-extern crate futures;
 extern crate rsocket_rust;
 extern crate tokio;
-
-use bytes::Bytes;
-use futures::prelude::*;
+#[macro_use]
+extern crate log;
 use rsocket_rust::prelude::*;
+use std::env;
+use std::error::Error;
 
-fn main() {
-  env_logger::builder()
-    .default_format_timestamp_nanos(true)
-    .init();
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    env_logger::builder().init();
+    let addr = env::args().nth(1).unwrap_or("127.0.0.1:7878".to_string());
 
-  let server = RSocketFactory::receive()
-    .transport(URI::Tcp("127.0.0.1:7878"))
-    .acceptor(|setup, sending_socket| {
-      info!("accept setup: {:?}", setup);
-      // TODO: use tokio runtime?
-      // std::thread::spawn(move || {
-      //   let resp = sending_socket
-      //     .request_response(
-      //       Payload::builder()
-      //         .set_data(Bytes::from("Hello Client!"))
-      //         .build(),
-      //     )
-      //     .wait()
-      //     .unwrap();
-      //   println!(">>>>> response success: {:?}", resp);
-      // });
-      Box::new(MockResponder)
-    })
-    .serve();
-  tokio::run(server);
+    RSocketFactory::receive()
+        .transport(URI::Tcp(addr))
+        .acceptor(|setup, _socket| {
+            info!("accept setup: {:?}", setup);
+            Box::new(EchoRSocket)
+        })
+        .serve()
+        .await
 }