feat: migrate to tokio v1

Author: jjeffcaii

examples/Cargo.toml

@@ -6,9 +6,9 @@ edition = "2018"
 publish = false
 
 [dev-dependencies]
-log = "0.4.11"
+log = "0.4.13"
 env_logger = "0.8.2"
-futures = "0.3.8"
+futures = "0.3.9"
 clap = "2.33.3"
 
 [dev-dependencies.rsocket_rust]
@@ -22,7 +22,7 @@ features = ["tls"]
 path = "../rsocket-transport-websocket"
 
 [dev-dependencies.tokio]
-version = "0.3.6"
+version = "1.0.1"
 default-features = false
 features = ["full"]
 

rsocket-messaging/Cargo.toml

@@ -10,10 +10,10 @@ homepage = "https://github.com/rsocket/rsocket-rust"
 description = "Communicate with Spring RSocket Messaging."
 
 [dependencies]
-futures = "0.3.8"
-bytes = "0.6.0"
-serde = "1.0.118"
-serde_json = "1.0.60"
+futures = "0.3.9"
+bytes = "1.0.1"
+serde = "1.0.119"
+serde_json = "1.0.61"
 serde_cbor = "0.11.1"
 hex = "0.4.2"
 url = "2.2.0"

rsocket-test/Cargo.toml

@@ -7,13 +7,13 @@ publish = false
 
 [dev-dependencies]
 log = "0.4"
-futures = "0.3.8"
+futures = "0.3.9"
 env_logger = "0.8.2"
-bytes = "0.6.0"
+bytes = "1.0.1"
 hex = "0.4.2"
-rand = "0.8.0"
-serde = "1.0.118"
-serde_derive = "1.0.118"
+rand = "0.8.1"
+serde = "1.0.119"
+serde_derive = "1.0.119"
 
 [dev-dependencies.rsocket_rust]
 path = "../rsocket"
@@ -29,6 +29,6 @@ path = "../rsocket-transport-websocket"
 path = "../rsocket-messaging"
 
 [dev-dependencies.tokio]
-version = "0.3.6"
+version = "1.0.1"
 default-features = false
 features = ["full"]

rsocket-transport-tcp/Cargo.toml

@@ -14,24 +14,24 @@ default = []
 tls = ["tokio-native-tls"]
 
 [dependencies]
-log = "0.4.11"
-futures = "0.3.8"
-bytes = "0.6.0"
+log = "0.4.13"
+futures = "0.3.9"
+bytes = "1.0.1"
 
 [dependencies.rsocket_rust]
 path = "../rsocket"
 features = ["frame"]
 
 [dependencies.tokio]
-version = "0.3.6"
+version = "1.0.1"
 default-features = false
-features = [ "rt", "rt-multi-thread", "net", "sync", "stream", "io-util", "macros" ]
+features = [ "rt", "rt-multi-thread", "net", "sync", "io-util", "macros" ]
 
 [dependencies.tokio-util]
-version = "0.5.1"
+version = "0.6.0"
 default-features = false
 features = ["codec"]
 
 [dependencies.tokio-native-tls]
 optional = true
-version = "0.2.0"
+version = "0.3.0"

rsocket-transport-wasm/Cargo.toml

@@ -10,15 +10,15 @@ homepage = "https://github.com/rsocket/rsocket-rust"
 description = "WASM Websocket RSocket transport implementation."
 
 [dependencies]
-bytes = "0.6.0"
+bytes = "1.0.1"
 wasm-bindgen-futures = "0.4.19"
 futures-channel = "0.3.9"
 futures-util = "0.3.9"
 js-sys = "0.3.46"
-serde = "1.0.118"
-serde_derive = "1.0.118"
+serde = "1.0.119"
+serde_derive = "1.0.119"
 async-trait = "0.1.42"
-log = "0.4.11"
+log = "0.4.13"
 
 [dependencies.rsocket_rust]
 path = "../rsocket"

rsocket-transport-websocket/Cargo.toml

@@ -10,17 +10,17 @@ homepage = "https://github.com/rsocket/rsocket-rust"
 description = "Websocket RSocket transport implementation."
 
 [dependencies]
-log = "0.4.11"
-futures = "0.3.8"
-bytes = "0.6.0"
+log = "0.4.13"
+futures = "0.3.9"
+bytes = "1.0.1"
 url = "2.2.0"
-tokio-tungstenite = "0.12.0"
+tokio-tungstenite = "0.13.0"
 
 [dependencies.rsocket_rust]
 path = "../rsocket"
 features = ["frame"]
 
 [dependencies.tokio]
-version = "0.3.6"
+version = "1.0.1"
 default-features = false
-features = [ "rt", "rt-multi-thread", "net", "sync", "stream" ]
+features = [ "rt", "rt-multi-thread", "net", "sync"]

rsocket/Cargo.toml

@@ -10,23 +10,23 @@ homepage = "https://github.com/rsocket/rsocket-rust"
 description = "rsocket-rust is an implementation of the RSocket protocol in Rust."
 
 [dependencies]
-log = "0.4.11"
-bytes = "0.6.0"
-futures = "0.3.8"
+log = "0.4.13"
+bytes = "1.0.1"
+futures = "0.3.9"
 lazy_static = "1.4.0"
 async-trait = "0.1.42"
 dashmap = "4.0.1"
 thiserror = "1.0.23"
-anyhow = "1.0.36"
+anyhow = "1.0.38"
 async-stream = "0.3.0"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 wasm-bindgen-futures = "0.4.19"
 
 [dependencies.tokio]
-version = "0.3.6"
+version = "1.0.1"
 default-features = false
-features = [ "rt", "rt-multi-thread", "sync", "stream", "time" ]
+features = [ "rt", "rt-multi-thread", "sync", "time" ]
 
 [features]
 default = []

rsocket/src/core/client.rs

@@ -143,7 +143,7 @@ where
         runtime::spawn(async move {
             loop {
                 // send keepalive if timeout
-                match tokio::time::timeout(tick_period, snd_rx.next()).await {
+                match tokio::time::timeout(tick_period, snd_rx.recv()).await {
                     Ok(Some(frame)) => {
                         if let frame::Body::Error(e) = frame.get_body_ref() {
                             if e.get_code() == ERR_CONN_CLOSED {
@@ -194,7 +194,7 @@ where
         });
 
         runtime::spawn(async move {
-            while let Some(next) = read_rx.next().await {
+            while let Some(next) = read_rx.recv().await {
                 if let Err(e) = cloned_socket.dispatch(next, None).await {
                     error!("dispatch frame failed: {}", e);
                     break;

rsocket/src/core/server.rs

@@ -146,7 +146,7 @@ where
             }
         });
 
-        while let Some(frame) = read_rx.next().await {
+        while let Some(frame) = read_rx.recv().await {
             if let Err(e) = socket.dispatch(frame, acceptor.as_ref().as_ref()).await {
                 error!("dispatch incoming frame failed: {}", e);
                 break;

rsocket/src/transport/socket.rs

@@ -15,7 +15,6 @@ use futures::{Sink, SinkExt, Stream, StreamExt};
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
-use tokio::prelude::*;
 use tokio::sync::{mpsc, oneshot, RwLock};
 
 #[derive(Clone)]
@@ -95,7 +94,7 @@ impl DuplexSocket {
 
     #[inline]
     async fn loop_canceller(&self, mut rx: mpsc::Receiver<u32>) {
-        while let Some(sid) = rx.next().await {
+        while let Some(sid) = rx.recv().await {
             self.handlers.remove(&sid);
         }
     }
@@ -456,12 +455,16 @@ impl DuplexSocket {
     async fn on_request_channel(&self, sid: u32, flag: u16, first: Payload) {
         let responder = self.responder.clone();
         let tx = self.tx.clone();
-        let (sender, receiver) = mpsc::channel::<Result<Payload>>(32);
+        let (sender, mut receiver) = mpsc::channel::<Result<Payload>>(32);
         sender.send(Ok(first)).await.expect("Send failed!");
         self.register_handler(sid, Handler::ReqRC(sender)).await;
         runtime::spawn(async move {
             // respond client channel
-            let mut outputs = responder.request_channel(Box::pin(receiver));
+            let mut outputs = responder.request_channel(Box::pin(stream! {
+                while let Some(it) = receiver.recv().await{
+                    yield it;
+                }
+            }));
             // TODO: support custom RequestN.
             let request_n = frame::RequestN::builder(sid, 0).build();
 
@@ -784,7 +787,7 @@ impl RSocket for DuplexSocket {
         let sid = self.seq.next();
         let tx = self.tx.clone();
         // register handler
-        let (sender, receiver) = mpsc::channel::<Result<Payload>>(32);
+        let (sender, mut receiver) = mpsc::channel::<Result<Payload>>(32);
         let handlers = self.handlers.clone();
         let splitter = self.splitter.clone();
         runtime::spawn(async move {
@@ -843,14 +846,18 @@ impl RSocket for DuplexSocket {
                 }
             }
         });
-        Box::pin(receiver)
+        Box::pin(stream! {
+            while let Some(it) = receiver.recv().await{
+                yield it;
+            }
+        })
     }
 
     fn request_channel(&self, mut reqs: Flux<Result<Payload>>) -> Flux<Result<Payload>> {
         let sid = self.seq.next();
         let mut tx = self.tx.clone();
         // register handler
-        let (sender, receiver) = mpsc::channel::<Result<Payload>>(32);
+        let (sender, mut receiver) = mpsc::channel::<Result<Payload>>(32);
         let handlers = self.handlers.clone();
         let splitter = self.splitter.clone();
         runtime::spawn(async move {
@@ -884,7 +891,11 @@ impl RSocket for DuplexSocket {
                 error!("complete REQUEST_CHANNEL failed: {}", e);
             }
         });
-        Box::pin(receiver)
+        Box::pin(stream! {
+            while let Some(it) = receiver.recv().await{
+                yield it;
+            }
+        })
     }
 }