Support tokio v0.2.x and async-await.

Author: jjeffcaii

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,31 @@ 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.1"
+futures = "0.3.1"
+# 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"
 
 [[example]]
 name = "echo"
 path = "examples/echo/main.rs"
 
-[[example]]
-name = "proxy"
-path = "examples/proxy/main.rs"
+# [[example]]
+# name = "proxy"
+# path = "examples/proxy/main.rs"

examples/echo/main.rs

@@ -1,42 +1,48 @@
-#![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 futures::{SinkExt, StreamExt};
+use rsocket_rust::frame::Frame;
 use rsocket_rust::prelude::*;
+use rsocket_rust::transport::DuplexSocket;
+use rsocket_rust::transport::{self, Rx, Tx};
+use std::env;
+use std::error::Error;
+use std::sync::Arc;
+use tokio::net::TcpListener;
+use tokio::prelude::*;
+use tokio::sync::mpsc;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    env_logger::builder().init();
+    RSocketFactory::receive()
+        .transport(URI::Tcp("127.0.0.1:7878"))
+        .acceptor(|setup, sending_socket| {
+            info!("accept setup: {:?}", setup);
+            Box::new(EchoRSocket)
+        })
+        .serve()
+        .await
+
+    // let addr = env::args().nth(1).unwrap_or("127.0.0.1:7878".to_string());
+    // let mut listener = TcpListener::bind(&addr).await?;
+    // println!("Listening on: {}", addr);
+
+    // loop {
+    //     let (mut socket, _) = listener.accept().await?;
+    //     let (rcv_tx, mut rcv_rx) = mpsc::unbounded_channel::<Frame>();
+    //     let (snd_tx, snd_rx) = mpsc::unbounded_channel::<Frame>();
 
-fn main() {
-  env_logger::builder()
-    .default_format_timestamp_nanos(true)
-    .init();
+    //     tokio::spawn(
+    //         async move { rsocket_rust::transport::tcp::process(socket, snd_rx, rcv_tx).await },
+    //     );
 
-  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);
+    //     let ds = DuplexSocket::new(0, snd_tx.clone());
+    //     tokio::spawn(async move {
+    //         let acceptor = Acceptor::Generate(Arc::new(|setup, socket| Box::new(EchoRSocket)));
+    //         ds.event_loop(acceptor, rcv_rx).await;
+    //     });
+    // }
 }