feat: support tokio v0.3.x (#23)

jjeffcaii · web-flow · commit 396653f5cf99 · 2020-11-08T22:06:41.000+08:00
diff --git a/README.md b/README.md
@@ -1,5 +1,6 @@
 # rsocket-rust
 
+![GitHub Workflow Status](https://github.com/rsocket/rsocket-rust/workflows/Rust/badge.svg)
 [![Build Status](https://travis-ci.com/rsocket/rsocket-rust.svg?branch=master)](https://travis-ci.com/rsocket/rsocket-rust)
 [![Crates.io](https://img.shields.io/crates/v/rsocket_rust)](https://crates.io/crates/rsocket_rust)
 [![Crates.io](https://img.shields.io/crates/d/rsocket_rust)](https://crates.io/crates/rsocket_rust)
diff --git a/examples/Cargo.toml b/examples/Cargo.toml
@@ -7,10 +7,10 @@ publish = false
 
 [dev-dependencies]
 log = "0.4.11"
-env_logger = "0.7.1"
-futures = "0.3.5"
+env_logger = "0.8.1"
+futures = "0.3.7"
 clap = "2.33.3"
-tokio-postgres = "0.5.5"
+tokio-postgres = "0.6.0"
 redis = "0.17.0"
 
 [dev-dependencies.rsocket_rust]
@@ -23,7 +23,7 @@ path = "../rsocket-transport-tcp"
 path = "../rsocket-transport-websocket"
 
 [dev-dependencies.tokio]
-version = "0.2.22"
+version = "0.3.3"
 default-features = false
 features = ["full"]
 
diff --git a/examples/qps.rs b/examples/qps.rs
@@ -48,7 +48,7 @@ where
             }
             let current = counter.fetch_add(1, Ordering::SeqCst) + 1;
             if current >= count {
-                notify.notify();
+                notify.notify_one();
             }
         });
     }
diff --git a/rsocket-messaging/Cargo.toml b/rsocket-messaging/Cargo.toml
@@ -10,13 +10,13 @@ homepage = "https://github.com/rsocket/rsocket-rust"
 description = "Communicate with Spring RSocket Messaging."
 
 [dependencies]
-futures = "0.3.5"
-bytes = "0.5.6"
-serde = "1.0.116"
-serde_json = "1.0.57"
+futures = "0.3.7"
+bytes = "0.6.0"
+serde = "1.0.117"
+serde_json = "1.0.59"
 serde_cbor = "0.11.1"
 hex = "0.4.2"
-url = "2.1.1"
+url = "2.2.0"
 
 [dependencies.rsocket_rust]
 path = "../rsocket"
diff --git a/rsocket-test/Cargo.toml b/rsocket-test/Cargo.toml
@@ -7,13 +7,13 @@ publish = false
 
 [dev-dependencies]
 log = "0.4"
-futures = "0.3.5"
-env_logger = "0.7.1"
-bytes = "0.5.6"
+futures = "0.3.7"
+env_logger = "0.8.1"
+bytes = "0.6.0"
 hex = "0.4.2"
 rand = "0.7.3"
-serde = "1.0.116"
-serde_derive = "1.0.116"
+serde = "1.0.117"
+serde_derive = "1.0.117"
 
 [dev-dependencies.rsocket_rust]
 path = "../rsocket"
@@ -29,6 +29,6 @@ path = "../rsocket-transport-websocket"
 path = "../rsocket-messaging"
 
 [dev-dependencies.tokio]
-version = "0.2.22"
+version = "0.3.3"
 default-features = false
 features = ["full"]
diff --git a/rsocket-test/tests/test_clients.rs b/rsocket-test/tests/test_clients.rs
@@ -54,7 +54,7 @@ fn test_websocket() {
 
     sleep(Duration::from_millis(500));
 
-    let mut client_runtime = Runtime::new().unwrap();
+    let client_runtime = Runtime::new().unwrap();
 
     client_runtime.block_on(async {
         let cli = RSocketFactory::connect()
@@ -102,7 +102,7 @@ fn test_tcp() {
 
     sleep(Duration::from_millis(500));
 
-    let mut client_runtime = Runtime::new().unwrap();
+    let client_runtime = Runtime::new().unwrap();
 
     client_runtime.block_on(async {
         let cli = RSocketFactory::connect()
@@ -158,7 +158,7 @@ fn test_unix() {
 
     sleep(Duration::from_millis(500));
 
-    let mut client_runtime = Runtime::new().unwrap();
+    let client_runtime = Runtime::new().unwrap();
 
     client_runtime.block_on(async {
         let cli = RSocketFactory::connect()
diff --git a/rsocket-transport-tcp/Cargo.toml b/rsocket-transport-tcp/Cargo.toml
@@ -11,20 +11,20 @@ description = "TCP RSocket transport implementation."
 
 [dependencies]
 log = "0.4.11"
-futures = "0.3.5"
-bytes = "0.5.6"
-async-trait = "0.1.40"
+futures = "0.3.7"
+bytes = "0.6.0"
+async-trait = "0.1.41"
 
 [dependencies.rsocket_rust]
 path = "../rsocket"
 features = ["frame"]
 
 [dependencies.tokio]
-version = "0.2.22"
+version = "0.3.3"
 default-features = false
-features = [ "rt-core", "rt-threaded", "tcp", "uds", "sync", "dns", "stream", "io-util", "macros" ]
+features = [ "rt", "rt-multi-thread", "net", "sync", "stream", "io-util", "macros" ]
 
 [dependencies.tokio-util]
-version = "0.3.1"
+version = "0.5.0"
 default-features = false
 features = ["codec"]
diff --git a/rsocket-transport-wasm/Cargo.toml b/rsocket-transport-wasm/Cargo.toml
@@ -10,14 +10,14 @@ homepage = "https://github.com/rsocket/rsocket-rust"
 description = "WASM Websocket RSocket transport implementation."
 
 [dependencies]
-bytes = "0.5.6"
+bytes = "0.6.0"
 wasm-bindgen-futures = "0.4.18"
-futures-channel = "0.3.5"
-futures-util = "0.3.5"
+futures-channel = "0.3.7"
+futures-util = "0.3.7"
 js-sys = "0.3.45"
-serde = "1.0.116"
-serde_derive = "1.0.116"
-async-trait = "0.1.40"
+serde = "1.0.117"
+serde_derive = "1.0.117"
+async-trait = "0.1.41"
 
 [dependencies.rsocket_rust]
 path = "../rsocket"
diff --git a/rsocket-transport-websocket/Cargo.toml b/rsocket-transport-websocket/Cargo.toml
@@ -11,17 +11,17 @@ description = "Websocket RSocket transport implementation."
 
 [dependencies]
 log = "0.4.11"
-futures = "0.3.5"
-bytes = "0.5.6"
-url = "2.1.1"
-tokio-tungstenite = "0.11.0"
-async-trait = "0.1.40"
+futures = "0.3.7"
+bytes = "0.6.0"
+url = "2.2.0"
+tokio-tungstenite = "0.12.0"
+async-trait = "0.1.41"
 
 [dependencies.rsocket_rust]
 path = "../rsocket"
 features = ["frame"]
 
 [dependencies.tokio]
-version = "0.2.22"
+version = "0.3.3"
 default-features = false
-features = [ "rt-core", "rt-threaded", "tcp", "sync", "stream" ]
+features = [ "rt", "rt-multi-thread", "net", "sync", "stream" ]
diff --git a/rsocket/Cargo.toml b/rsocket/Cargo.toml
@@ -11,21 +11,21 @@ description = "rsocket-rust is an implementation of the RSocket protocol in Rust
 
 [dependencies]
 log = "0.4.11"
-bytes = "0.5.6"
-futures = "0.3.5"
+bytes = "0.6.0"
+futures = "0.3.7"
 lazy_static = "1.4.0"
-async-trait = "0.1.40"
+async-trait = "0.1.41"
 dashmap = "3.11.10"
-thiserror = "1.0.20"
-anyhow = "1.0.32"
+thiserror = "1.0.22"
+anyhow = "1.0.34"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 wasm-bindgen-futures = "0.4.18"
 
 [dependencies.tokio]
-version = "0.2.22"
+version = "0.3.3"
 default-features = false
-features = [ "rt-core", "rt-threaded", "sync", "stream" ]
+features = [ "rt", "rt-multi-thread", "sync", "stream" ]
 
 [features]
 default = []
diff --git a/rsocket/src/frame/error.rs b/rsocket/src/frame/error.rs
@@ -50,7 +50,7 @@ impl Error {
         } else {
             let code = bf.get_u32();
             let data: Option<Bytes> = if !bf.is_empty() {
-                Some(bf.to_bytes())
+                Some(Bytes::from(bf.to_vec()))
             } else {
                 None
             };
diff --git a/rsocket/src/frame/keepalive.rs b/rsocket/src/frame/keepalive.rs
@@ -51,7 +51,7 @@ impl Keepalive {
         let data = if bf.is_empty() {
             None
         } else {
-            Some(bf.to_bytes())
+            Some(Bytes::from(bf.to_vec()))
         };
         Ok(Keepalive {
             last_received_position: position,
diff --git a/rsocket/src/frame/lease.rs b/rsocket/src/frame/lease.rs
@@ -58,7 +58,7 @@ impl Lease {
         let ttl = bf.get_u32();
         let n = bf.get_u32();
         let m = if flag & Frame::FLAG_METADATA != 0 {
-            Some(bf.to_bytes())
+            Some(Bytes::from(bf.to_vec()))
         } else {
             None
         };
diff --git a/rsocket/src/transport/socket.rs b/rsocket/src/transport/socket.rs
@@ -259,8 +259,8 @@ impl DuplexSocket {
             match handler {
                 Handler::ReqRR(tx) => tx.send(e).expect("Send RR failed"),
                 Handler::ResRR(_) => unreachable!(),
-                Handler::ReqRS(mut tx) => tx.send(e).await.expect("Send RS failed"),
-                Handler::ReqRC(mut tx) => tx.send(e).await.expect("Send RC failed"),
+                Handler::ReqRS(tx) => tx.send(e).await.expect("Send RS failed"),
+                Handler::ReqRC(tx) => tx.send(e).await.expect("Send RC failed"),
             }
         }
     }
@@ -368,7 +368,7 @@ impl DuplexSocket {
     #[inline]
     async fn on_request_response(&mut self, sid: u32, _flag: u16, input: Payload) {
         let responder = self.responder.clone();
-        let mut canceller = self.canceller.clone();
+        let canceller = self.canceller.clone();
         let mut tx = self.tx.clone();
         let splitter = self.splitter.clone();
         let counter = Counter::new(2);
@@ -441,8 +441,8 @@ impl DuplexSocket {
     #[inline]
     async fn on_request_channel(&self, sid: u32, flag: u16, first: Payload) {
         let responder = self.responder.clone();
-        let mut tx = self.tx.clone();
-        let (mut sender, receiver) = mpsc::channel::<Result<Payload>>(32);
+        let tx = self.tx.clone();
+        let (sender, 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 {
@@ -620,7 +620,7 @@ impl DuplexSocket {
 impl RSocket for DuplexSocket {
     fn metadata_push(&self, req: Payload) -> Mono<()> {
         let sid = self.seq.next();
-        let mut tx = self.tx.clone();
+        let tx = self.tx.clone();
         Box::pin(async move {
             let (_d, m) = req.split();
             let mut bu = frame::MetadataPush::builder(sid, 0);
@@ -634,7 +634,7 @@ impl RSocket for DuplexSocket {
     }
     fn fire_and_forget(&self, req: Payload) -> Mono<()> {
         let sid = self.seq.next();
-        let mut tx = self.tx.clone();
+        let tx = self.tx.clone();
         let splitter = self.splitter.clone();
         Box::pin(async move {
             match splitter {
@@ -696,7 +696,7 @@ impl RSocket for DuplexSocket {
         let (tx, rx) = oneshot::channel::<Result<Payload>>();
         let sid = self.seq.next();
         let handlers = self.handlers.clone();
-        let mut sender = self.tx.clone();
+        let sender = self.tx.clone();
 
         let splitter = self.splitter.clone();
 
@@ -770,7 +770,7 @@ impl RSocket for DuplexSocket {
 
     fn request_stream(&self, input: Payload) -> Flux<Result<Payload>> {
         let sid = self.seq.next();
-        let mut tx = self.tx.clone();
+        let tx = self.tx.clone();
         // register handler
         let (sender, receiver) = mpsc::channel::<Result<Payload>>(32);
         let handlers = self.handlers.clone();

Original file line number	Diff line number	Diff line change
`@@ -48,7 +48,7 @@ where`
`48`	`48`	`}`
`49`	`49`	`let current = counter.fetch_add(1, Ordering::SeqCst) + 1;`
`50`	`50`	`if current >= count {`
`51`		`- notify.notify();`
	`51`	`+ notify.notify_one();`
`52`	`52`	`}`
`53`	`53`	`});`
`54`	`54`	`}`