Update tonic to 0.13.1 and tower to 0.5

belltoy · belltoy · commit 723b19b2fb57 · 2025-07-13T09:07:29.000+08:00
- Update tonic to 0.13.1 and tower to 0.5
- Use string interpolation in build.rs to pass clippy
- Fix the broken Buffer API since 0.5 and bridge tower Change and tonic Change
diff --git a/Cargo.toml b/Cargo.toml
@@ -13,23 +13,23 @@ documentation = "https://docs.rs/etcd-client/"
 keywords = ["etcd", "v3", "api", "client", "async"]
 
 [features]
-tls = ["tonic/tls"]
+tls = ["tonic/tls-ring"]
 tls-openssl = ["openssl", "hyper-openssl", "hyper", "hyper-util"]
 tls-openssl-vendored = ["tls-openssl", "openssl/vendored"]
-tls-roots = ["tls", "tonic/tls-roots"]
+tls-roots = ["tls", "tonic/tls-native-roots"]
 pub-response-field = ["visible"]
 build-server = ["pub-response-field"]
 raw-channel = []
 
 [dependencies]
-tonic = "0.12.3"
+tonic = "0.13.1"
 prost = "0.13"
 tokio = "1.38"
 tokio-stream = "0.1"
 tower-service = "0.3"
 http = "1.1"
 visible = { version = "0.0.1", optional = true }
-tower = { version = "0.4", default-features = false }
+tower = { version = "0.5", default-features = false }
 openssl = { version = "0.10", optional = true }
 hyper = { version = "1.4", features = ["client"], optional = true }
 hyper-openssl = { version = "0.10", features = ["client-legacy", "tokio"], optional = true }
@@ -39,7 +39,7 @@ hyper-util = { version = "0.1", features = ["client-legacy"], optional = true }
 tokio = { version = "1.38", features = ["full"] }
 
 [build-dependencies]
-tonic-build = { version = "0.12.3", default-features = false, features = ["prost"] }
+tonic-build = { version = "0.13.1", default-features = false, features = ["prost"] }
 
 [package.metadata.docs.rs]
 features = ["tls", "tls-roots"]
diff --git a/build.rs b/build.rs
@@ -10,7 +10,7 @@ fn should_build_server() -> bool {
 
 fn main() {
     let proto_root = "proto";
-    println!("cargo:rerun-if-changed={}", proto_root);
+    println!("cargo:rerun-if-changed={proto_root}");
 
     tonic_build::configure()
         .build_server(should_build_server())
diff --git a/examples/auth_role.rs b/examples/auth_role.rs
@@ -45,7 +45,7 @@ async fn main() -> Result<(), Error> {
 
     let resp = client.role_get(role1).await?;
     //show the result
-    println!("Role {:?}\n{:?}", role1, resp);
+    println!("Role {role1:?}\n{resp:?}");
 
     let resp = client.role_revoke_permission(role1, "abc", None).await?;
     println!("{:?}", resp.header());
@@ -70,7 +70,7 @@ async fn main() -> Result<(), Error> {
 
     let resp = client.role_get(role1).await?;
     //show the result
-    println!("Role {:?}\n{:?}", role1, resp);
+    println!("Role {role1:?}\n{resp:?}");
 
     let resp = client.role_list().await?;
     //show the result
diff --git a/examples/auth_user.rs b/examples/auth_user.rs
@@ -40,14 +40,14 @@ async fn main() -> Result<(), Error> {
 
     let resp = client.user_get(name1).await?;
     //show the result
-    println!("User {:?}\n{:?}", name1, resp);
+    println!("User {name1:?}\n{resp:?}");
 
     client.user_revoke_role(name1, role1).await?;
     println!("revoke role1 from usr1 successfully");
 
     let resp = client.user_get(name1).await?;
     //show the result
-    println!("User {:?}\n{:?}", name1, resp);
+    println!("User {name1:?}\n{resp:?}");
 
     let resp = client.user_list().await?;
     //show the result
diff --git a/examples/cluster.rs b/examples/cluster.rs
@@ -37,6 +37,6 @@ async fn main() -> Result<(), Error> {
 
     // remove the added member
     let _resp = client.member_remove(id).await?;
-    println!("remove a member with id {:?}", id);
+    println!("remove a member with id {id:?}");
     Ok(())
 }
diff --git a/examples/kv.rs b/examples/kv.rs
@@ -38,12 +38,12 @@ async fn main() -> Result<(), Error> {
 
     // put kv
     let resp = client.put(alice.key(), alice.value(), None).await?;
-    println!("put kv: {:?}", alice);
+    println!("put kv: {alice:?}");
     let revision = resp.header().unwrap().revision();
     client.put(bob.key(), bob.value(), None).await?;
-    println!("put kv: {:?}", bob);
+    println!("put kv: {bob:?}");
     client.put(chris.key(), chris.value(), None).await?;
-    println!("put kv: {:?}", chris);
+    println!("put kv: {chris:?}");
     println!();
 
     // get kv
@@ -92,10 +92,10 @@ async fn main() -> Result<(), Error> {
             "18",
             Some(PutOptions::new().with_ignore_lease()),
         )]);
-    println!("transaction: {:?}", txn);
+    println!("transaction: {txn:?}");
     let resp = client.txn(txn).await?;
     for op_resp in resp.op_responses() {
-        println!("transaction resp: {:?}", op_resp);
+        println!("transaction resp: {op_resp:?}");
     }
     let resp = client.get(alice.key(), None).await?;
     if let Some(kv) = resp.kvs().first() {
@@ -111,7 +111,7 @@ async fn main() -> Result<(), Error> {
     client
         .compact(revision, Some(CompactionOptions::new().with_physical()))
         .await?;
-    println!("Compact to revision {}.", revision);
+    println!("Compact to revision {revision}.");
 
     Ok(())
 }
diff --git a/examples/lease.rs b/examples/lease.rs
@@ -26,7 +26,7 @@ async fn main() -> Result<(), Error> {
 
     // keep alive
     let (mut keeper, mut stream) = client.lease_keep_alive(id).await?;
-    println!("lease {:?} keep alive start", id);
+    println!("lease {id:?} keep alive start");
     keeper.keep_alive().await?;
     if let Some(resp) = stream.message().await? {
         println!("lease {:?} keep alive, new ttl {:?}", resp.id(), resp.ttl());
@@ -39,11 +39,11 @@ async fn main() -> Result<(), Error> {
 
     // revoke a lease
     let _resp = client.lease_revoke(id).await?;
-    println!("revoke a lease with id {:?}", id);
+    println!("revoke a lease with id {id:?}");
 
     // keep alive a revoked lease returns error
     if let Err(err) = client.lease_keep_alive(id).await {
-        println!("revoked lease {:?} keep alive error: {:?}", id, err);
+        println!("revoked lease {id:?} keep alive error: {err:?}");
     }
     Ok(())
 }
diff --git a/examples/lock.rs b/examples/lock.rs
@@ -10,7 +10,7 @@ async fn main() -> Result<(), Error> {
     let resp = client.lock("lock-test", None).await?;
     let key = resp.key();
     let key_str = std::str::from_utf8(key)?;
-    println!("the key is {:?}", key_str);
+    println!("the key is {key_str:?}");
 
     println!("try to unlock it");
     client.unlock(key).await?;
@@ -27,15 +27,12 @@ async fn main() -> Result<(), Error> {
     let lease_id = resp.id();
 
     // lock with lease
-    println!(
-        "try to lock with name \'lock-test2\' and lease {:?}",
-        lease_id
-    );
+    println!("try to lock with name \'lock-test2\' and lease {lease_id:?}");
     let lock_options = LockOptions::new().with_lease(lease_id);
     let resp = client.lock("lock-test2", Some(lock_options)).await?;
     let key = resp.key();
     let key_str = std::str::from_utf8(key);
-    println!("the key is {:?}", key_str);
+    println!("the key is {key_str:?}");
 
     println!("try to unlock it");
     client.unlock(key).await?;
diff --git a/examples/maintenance.rs b/examples/maintenance.rs
@@ -75,7 +75,7 @@ async fn main() -> Result<(), Error> {
     if member_id == leader_id {
         assert!(header.is_none());
     } else {
-        println!("move_leader header {:?}", header);
+        println!("move_leader header {header:?}");
     }
 
     Ok(())
diff --git a/src/channel.rs b/src/channel.rs
@@ -2,8 +2,8 @@ use std::{future::Future, pin::Pin, task::ready};
 
 use http::Uri;
 use tokio::sync::mpsc::Sender;
-use tonic::transport::Endpoint;
-use tower::{discover::Change, util::BoxCloneService, Service};
+use tonic::transport::{channel::Change, Endpoint};
+use tower::{util::BoxCloneService, Service};
 
 /// A type alias to make the below types easier to represent.
 pub type EndpointUpdater = Sender<Change<Uri, Endpoint>>;
@@ -52,8 +52,8 @@ impl BalancedChannelBuilder for Openssl {
     }
 }
 
-type TonicRequest = http::Request<tonic::body::BoxBody>;
-type TonicResponse = http::Response<tonic::body::BoxBody>;
+type TonicRequest = http::Request<tonic::body::Body>;
+type TonicResponse = http::Response<tonic::body::Body>;
 pub type CustomChannel = BoxCloneService<TonicRequest, TonicResponse, tower::BoxError>;
 
 /// Represents a channel that can be created by a BalancedChannelBuilder
diff --git a/src/client.rs b/src/client.rs
@@ -49,9 +49,7 @@ use std::sync::{Arc, RwLock};
 use std::time::Duration;
 use tokio::sync::mpsc::Sender;
 
-use tonic::transport::Endpoint;
-
-use tower::discover::Change;
+use tonic::transport::{channel::Change, Endpoint};
 
 const HTTP_PREFIX: &str = "http://";
 const HTTPS_PREFIX: &str = "https://";
@@ -319,7 +317,7 @@ impl Client {
         };
         tx.send(Change::Insert(endpoint.uri().clone(), endpoint))
             .await
-            .map_err(|e| Error::EndpointError(format!("failed to add endpoint because of {}", e)))
+            .map_err(|e| Error::EndpointError(format!("failed to add endpoint because of {e}")))
     }
 
     /// Dynamically remove an endpoint from the client.
@@ -333,9 +331,9 @@ impl Client {
         let Some(tx) = &self.tx else {
             return Err(Error::EndpointsNotManaged);
         };
-        tx.send(Change::Remove(uri)).await.map_err(|e| {
-            Error::EndpointError(format!("failed to remove endpoint because of {}", e))
-        })
+        tx.send(Change::Remove(uri))
+            .await
+            .map_err(|e| Error::EndpointError(format!("failed to remove endpoint because of {e}")))
     }
 
     /// Gets a KV client.
diff --git a/src/error.rs b/src/error.rs
@@ -53,20 +53,20 @@ impl Display for Error {
     #[inline]
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {
-            Error::InvalidArgs(e) => write!(f, "invalid arguments: {}", e),
-            Error::InvalidUri(e) => write!(f, "invalid uri: {}", e),
-            Error::IoError(e) => write!(f, "io error: {}", e),
-            Error::TransportError(e) => write!(f, "transport error: {}", e),
-            Error::GRpcStatus(e) => write!(f, "grpc request error: {}", e),
-            Error::WatchError(e) => write!(f, "watch error: {}", e),
-            Error::Utf8Error(e) => write!(f, "utf8 error: {}", e),
-            Error::LeaseKeepAliveError(e) => write!(f, "lease keep alive error: {}", e),
-            Error::ElectError(e) => write!(f, "election error: {}", e),
-            Error::InvalidHeaderValue(e) => write!(f, "invalid metadata value: {}", e),
-            Error::EndpointError(e) => write!(f, "endpoint error: {}", e),
+            Error::InvalidArgs(e) => write!(f, "invalid arguments: {e}"),
+            Error::InvalidUri(e) => write!(f, "invalid uri: {e}"),
+            Error::IoError(e) => write!(f, "io error: {e}"),
+            Error::TransportError(e) => write!(f, "transport error: {e}"),
+            Error::GRpcStatus(e) => write!(f, "grpc request error: {e}"),
+            Error::WatchError(e) => write!(f, "watch error: {e}"),
+            Error::Utf8Error(e) => write!(f, "utf8 error: {e}"),
+            Error::LeaseKeepAliveError(e) => write!(f, "lease keep alive error: {e}"),
+            Error::ElectError(e) => write!(f, "election error: {e}"),
+            Error::InvalidHeaderValue(e) => write!(f, "invalid metadata value: {e}"),
+            Error::EndpointError(e) => write!(f, "endpoint error: {e}"),
             Error::EndpointsNotManaged => write!(f, "endpoints not managed by this client"),
             #[cfg(feature = "tls-openssl")]
-            Error::OpenSsl(e) => write!(f, "open ssl error: {}", e),
+            Error::OpenSsl(e) => write!(f, "open ssl error: {e}"),
         }
     }
 }
diff --git a/src/openssl_tls/backoff.rs b/src/openssl_tls/backoff.rs
@@ -172,18 +172,16 @@ mod test {
     fn test_back_off() {
         let mut state = BackOffStatus::new(Duration::from_secs(1), Duration::from_secs(4));
         let assert_state = |state: &mut BackOffStatus, failed, curr, next, desc| {
-            assert_eq!(state.failed(), failed, "{}: success status not match", desc);
+            assert_eq!(state.failed(), failed, "{desc}: success status not match");
             assert_eq!(
                 state.current_backoff_dur,
                 Duration::from_secs(curr),
-                "{}: current_backoff_dur not match",
-                desc
+                "{desc}: current_backoff_dur not match"
             );
             assert_eq!(
                 state.next_backoff_dur,
                 Duration::from_secs(next),
-                "{}: next_backoff_dur not match",
-                desc
+                "{desc}: next_backoff_dur not match"
             );
         };
         assert!(!state.failed());
diff --git a/src/openssl_tls/transport.rs b/src/openssl_tls/transport.rs
@@ -12,24 +12,22 @@ use openssl::{
 };
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio_stream::wrappers::ReceiverStream;
-use tonic::{
-    body::BoxBody,
-    transport::{Channel, Endpoint},
-};
-use tower::{balance::p2c::Balance, buffer::Buffer, discover::Change};
+use tonic::transport::{channel::Change, Channel, Endpoint};
+use tower::Service;
+use tower::{balance::p2c::Balance, buffer::Buffer, discover::Change as TowerChange};
 
 use crate::error::Result;
 
 pub type SslConnectorBuilder = openssl::ssl::SslConnectorBuilder;
 pub type OpenSslResult<T> = std::result::Result<T, ErrorStack>;
 // Below are some type alias for make clearer types.
-pub type TonicRequest = Request<BoxBody>;
-pub type Buffered<T> = Buffer<T, TonicRequest>;
+pub type TonicRequest = Request<tonic::body::Body>;
 pub type Balanced<T> = Balance<T, TonicRequest>;
-pub type OpenSslChannel = Buffered<Balanced<OpenSslDiscover<Uri>>>;
+pub type OpenSslChannel =
+    Buffer<TonicRequest, <Balanced<OpenSslDiscover<Uri>> as Service<TonicRequest>>::Future>;
 /// OpenSslDiscover is the backend for balanced channel based on OpenSSL transports.
 /// Because `Channel::balance` doesn't allow us to provide custom connector, we must implement ourselves' balancer...
-pub type OpenSslDiscover<K> = ReceiverStream<Result<Change<K, BackOffWhenFail<Channel>>>>;
+pub type OpenSslDiscover<K> = ReceiverStream<Result<TowerChange<K, BackOffWhenFail<Channel>>>>;
 
 #[derive(Clone)]
 pub struct OpenSslConnector(HttpsConnector<HttpConnector>);
@@ -91,7 +89,7 @@ fn create_openssl_discover<K: Send + 'static>(
                 match x {
                     Change::Insert(name, e) => {
                         let chan = e.connect_with_connector_lazy(connector.clone().0);
-                        Ok(Change::Insert(
+                        Ok(TowerChange::Insert(
                             name,
                             BackOffWhenFail::new(
                                 chan,
@@ -102,7 +100,7 @@ fn create_openssl_discover<K: Send + 'static>(
                             ),
                         ))
                     }
-                    Change::Remove(name) => Ok(Change::Remove(name)),
+                    Change::Remove(name) => Ok(TowerChange::Remove(name)),
                 }
             }
             .await;
diff --git a/src/rpc/watch.rs b/src/rpc/watch.rs
@@ -331,7 +331,7 @@ impl Event {
         match self.0.r#type {
             0 => EventType::Put,
             1 => EventType::Delete,
-            i => panic!("unknown event {}", i),
+            i => panic!("unknown event {i}"),
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -37,6 +37,6 @@ async fn main() -> Result<(), Error> {`
`37`	`37`
`38`	`38`	`// remove the added member`
`39`	`39`	`let _resp = client.member_remove(id).await?;`
`40`		`- println!("remove a member with id {:?}", id);`
	`40`	`+ println!("remove a member with id {id:?}");`
`41`	`41`	`Ok(())`
`42`	`42`	`}`
Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ async fn main() -> Result<(), Error> {`
`75`	`75`	`if member_id == leader_id {`
`76`	`76`	`assert!(header.is_none());`
`77`	`77`	`} else {`
`78`		`- println!("move_leader header {:?}", header);`
	`78`	`+ println!("move_leader header {header:?}");`
`79`	`79`	`}`
`80`	`80`
`81`	`81`	`Ok(())`