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
 }