Upgrade to Tokio 1.0

Cargo.toml

@@ -16,13 +16,14 @@ tls = ["tonic/tls"]
 tls-roots = ["tls", "tonic/tls-roots"]
 
 [dependencies]
-tonic = "0.3"
-prost = "0.6"
-tokio = { version = "0.2", features = ["stream"] }
-http = "0.2"
+tonic = "0.4.0"
+prost = "0.7.0"
+tokio = "1.0.2"
+tokio-stream = "0.1.2"
+http = "0.2.3"
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "1.0.2", features = ["full"] }
 
 [build-dependencies]
-tonic-build = { version = "0.3", default-features = false, features = ["prost"] }
+tonic-build = { version = "0.4.0", default-features = false, features = ["prost"] }

examples/lock.rs

@@ -13,7 +13,7 @@ async fn main() -> Result<(), Error> {
     println!("the key is {:?}", key_str);
 
     println!("try to unlock it");
-    client.unlock(key.clone()).await?;
+    client.unlock(key).await?;
     println!("finish!");
     println!();
 
@@ -38,7 +38,7 @@ async fn main() -> Result<(), Error> {
     println!("the key is {:?}", key_str);
 
     println!("try to unlock it");
-    client.unlock(key.clone()).await?;
+    client.unlock(key).await?;
     println!("finish!");
 
     Ok(())

examples/maintenance.rs

@@ -11,27 +11,23 @@ async fn main() -> Result<(), Error> {
         .alarm(AlarmAction::Get, AlarmType::None, Some(AlarmOptions::new()))
         .await?;
     let mems = resp.alarms();
-    print!("{} members have alarm.\n", mems.len());
+    println!("{} members have alarm.", mems.len());
 
     // Get status.
     let resp = client.status().await?;
-    print!(
-        "version: {}, db_size: {} \n",
-        resp.version(),
-        resp.db_size()
-    );
+    println!("version: {}, db_size: {}", resp.version(), resp.db_size());
 
     // Defragment.
     let _resp = client.defragment().await?;
 
     // Get hash value.
     let resp = client.hash().await?;
-    print!("hash: {} \n", resp.hash());
+    println!("hash: {}", resp.hash());
 
     // Get hash key value.
     let resp = client.hash_kv(0).await?;
-    print!(
-        "hash: {} revision: {} \n",
+    println!(
+        "hash: {}, revision: {}",
         resp.hash(),
         resp.compact_version()
     );
@@ -41,7 +37,7 @@ async fn main() -> Result<(), Error> {
     loop {
         let resp = msg.message().await?;
         if let Some(r) = resp {
-            print!("Receive blob len {}\n", r.blob().len());
+            println!("Receive blob len {}", r.blob().len());
             if r.remaining_bytes() == 0 {
                 break;
             }

src/client.rs

@@ -949,7 +949,7 @@ mod tests {
         let key_str = std::str::from_utf8(key)?;
         assert!(key_str.starts_with("lock-test/"));
 
-        client.unlock(key.clone()).await?;
+        client.unlock(key).await?;
         Ok(())
     }
 
@@ -960,10 +960,7 @@ mod tests {
         client.auth_enable().await?;
 
         // after enable auth, must operate by authenticated client
-        let resp = client.put("auth-test", "value", None).await;
-        if let Ok(_) = resp {
-            assert!(false);
-        }
+        client.put("auth-test", "value", None).await.unwrap_err();
 
         // connect with authenticate, the user must already exists
         let options = Some(ConnectOptions::new().with_user(
@@ -989,34 +986,22 @@ mod tests {
         let role1 = "role1";
         let role2 = "role2";
 
-        let _resp = client.role_delete(role1).await;
-        let _resp = client.role_delete(role2).await;
+        let _ = client.role_delete(role1).await;
+        let _ = client.role_delete(role2).await;
+
         client.role_add(role1).await?;
 
-        let resp = client.role_get(role1).await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
+        client.role_get(role1).await?;
 
         client.role_delete(role1).await?;
-        let resp = client.role_get(role1).await;
-        if let Ok(_) = resp {
-            assert!(false);
-        }
+        client.role_get(role1).await.unwrap_err();
 
         client.role_add(role2).await?;
-        let resp = client.role_get(role2).await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
+        client.role_get(role2).await?;
 
-        let resp = client.role_list().await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
-
-        if let Ok(l) = resp {
-            assert!(l.roles().contains(&role2.to_string()));
+        {
+            let resp = client.role_list().await?;
+            assert!(resp.roles().contains(&role2.to_string()));
         }
 
         client
@@ -1041,12 +1026,9 @@ mod tests {
             )
             .await?;
 
-        let resp = client.role_get(role2).await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
-        if let Ok(r) = resp {
-            let permissions = r.permissions();
+        {
+            let resp = client.role_get(role2).await?;
+            let permissions = resp.permissions();
             assert!(permissions.contains(&Permission::read("123")));
             assert!(permissions.contains(&Permission::write("abc").with_from_key()));
             assert!(permissions.contains(&Permission::read_write("hi").with_range_end("hjj")));
@@ -1085,13 +1067,8 @@ mod tests {
             )
             .await?;
 
-        let resp = client.role_get(role2).await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
-        if let Ok(r) = resp {
-            assert!(r.permissions().is_empty());
-        }
+        let resp = client.role_get(role2).await?;
+        assert!(resp.permissions().is_empty());
 
         client.role_delete(role2).await?;
 
@@ -1126,49 +1103,30 @@ mod tests {
 
         client.user_add(name3, password3, None).await?;
 
-        let resp = client.user_get(name1).await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
-
-        let resp = client.user_list().await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
+        client.user_get(name1).await?;
 
-        if let Ok(l) = resp {
-            assert!(l.users().contains(&name1.to_string()));
+        {
+            let resp = client.user_list().await?;
+            assert!(resp.users().contains(&name1.to_string()));
         }
 
         client.user_delete(name2).await?;
-        let resp = client.user_get(name2).await;
-        if let Ok(_) = resp {
-            assert!(false);
-        }
+        client.user_get(name2).await.unwrap_err();
 
         client.user_change_password(name1, password2).await?;
-        let resp = client.user_get(name1).await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
+        client.user_get(name1).await?;
 
         client.role_add(role1).await?;
         client.user_grant_role(name1, role1).await?;
-        let resp = client.user_get(name1).await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
+        client.user_get(name1).await?;
 
         client.user_revoke_role(name1, role1).await?;
-        let resp = client.user_get(name1).await;
-        if let Err(_) = resp {
-            assert!(false);
-        }
+        client.user_get(name1).await?;
 
-        let _resp = client.user_delete(name1).await;
-        let _resp = client.user_delete(name2).await;
-        let _resp = client.user_delete(name3).await;
-        let _resp = client.role_delete(role1).await;
+        let _ = client.user_delete(name1).await;
+        let _ = client.user_delete(name2).await;
+        let _ = client.user_delete(name3).await;
+        let _ = client.role_delete(role1).await;
 
         Ok(())
     }
@@ -1256,7 +1214,7 @@ mod tests {
         let mut msg = client.snapshot().await?;
         loop {
             if let Some(resp) = msg.message().await? {
-                assert!(resp.blob().len() > 0);
+                assert!(!resp.blob().is_empty());
                 if resp.remaining_bytes() == 0 {
                     break;
                 }

src/rpc/election.rs

@@ -9,9 +9,9 @@ use crate::rpc::pb::v3electionpb::{
     ResignRequest as PbResignRequest, ResignResponse as PbResignResponse,
 };
 use crate::rpc::{KeyValue, ResponseHeader};
+use std::pin::Pin;
 use std::task::{Context, Poll};
-use tokio::stream::Stream;
-use tonic::codegen::Pin;
+use tokio_stream::Stream;
 use tonic::transport::Channel;
 use tonic::{Interceptor, IntoRequest, Request, Streaming};
 

src/rpc/lease.rs

@@ -15,10 +15,11 @@ use crate::rpc::pb::etcdserverpb::{
 use crate::rpc::ResponseHeader;
 use crate::Error;
 
+use std::pin::Pin;
 use std::task::{Context, Poll};
-use tokio::stream::Stream;
 use tokio::sync::mpsc::{channel, Sender};
-use tonic::codegen::Pin;
+use tokio_stream::wrappers::ReceiverStream;
+use tokio_stream::Stream;
 use tonic::transport::Channel;
 use tonic::{Interceptor, IntoRequest, Request, Streaming};
 
@@ -73,12 +74,14 @@ impl LeaseClient {
     /// to the server and streaming keep alive responses from the server to the client.
     #[inline]
     pub async fn keep_alive(&mut self, id: i64) -> Result<(LeaseKeeper, LeaseKeepAliveStream)> {
-        let (mut sender, receiver) = channel::<PbLeaseKeepAliveRequest>(100);
+        let (sender, receiver) = channel::<PbLeaseKeepAliveRequest>(100);
         sender
             .send(LeaseKeepAliveOptions::new().with_id(id).into())
             .await
             .map_err(|e| Error::LeaseKeepAliveError(e.to_string()))?;
 
+        let receiver = ReceiverStream::new(receiver);
+
         let mut stream = self.inner.lease_keep_alive(receiver).await?.into_inner();
 
         let id = match stream.message().await? {

src/rpc/watch.rs

@@ -11,10 +11,10 @@ use crate::rpc::pb::etcdserverpb::{
 };
 use crate::rpc::pb::mvccpb::Event as PbEvent;
 use crate::rpc::{KeyRange, KeyValue, ResponseHeader};
+use std::pin::Pin;
 use std::task::{Context, Poll};
-use tokio::stream::Stream;
 use tokio::sync::mpsc::{channel, Sender};
-use tonic::codegen::Pin;
+use tokio_stream::{wrappers::ReceiverStream, Stream};
 use tonic::transport::Channel;
 use tonic::{Interceptor, Streaming};
 
@@ -47,12 +47,14 @@ impl WatchClient {
         key: impl Into<Vec<u8>>,
         options: Option<WatchOptions>,
     ) -> Result<(Watcher, WatchStream)> {
-        let (mut sender, receiver) = channel::<WatchRequest>(100);
+        let (sender, receiver) = channel::<WatchRequest>(100);
         sender
             .send(options.unwrap_or_default().with_key(key).into())
             .await
             .map_err(|e| Error::WatchError(e.to_string()))?;
 
+        let receiver = ReceiverStream::new(receiver);
+
         let mut stream = self.inner.watch(receiver).await?.into_inner();
 
         let watch_id = match stream.message().await? {