etcdv3 · davidli2010 · Jul 11, 2021 · Jul 10, 2021 · Jul 11, 2021 · Jul 11, 2021
diff --git a/Cargo.toml b/Cargo.toml
@@ -16,17 +16,18 @@ tls = ["tonic/tls"]
 tls-roots = ["tls", "tonic/tls-roots"]
 
 [dependencies]
-tonic = "0.4.3"
-prost = "0.7.0"
+tonic = "0.5.0"
+prost = "0.8.0"
 tokio = "1.8.1"
 tokio-stream = "0.1.7"
+tower-service = "0.3.1"
 http = "0.2.4"
 
 [dev-dependencies]
 tokio = { version = "1.8.1", features = ["full"] }
 
 [build-dependencies]
-tonic-build = { version = "0.4.2", default-features = false, features = ["prost"] }
+tonic-build = { version = "0.5.0", default-features = false, features = ["prost"] }
 
 [package.metadata.docs.rs]
 all-features = true

diff --git a/src/auth.rs b/src/auth.rs
@@ -0,0 +1,44 @@
+//! Authentication service.
+
+use http::{header::AUTHORIZATION, HeaderValue, Request};
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use tower_service::Service;
+
+#[derive(Debug, Clone)]
+pub struct AuthService<S> {
+    inner: S,
+    token: Option<Arc<HeaderValue>>,
+}
+
+impl<S> AuthService<S> {
+    #[inline]
+    pub fn new(inner: S, token: Option<Arc<HeaderValue>>) -> Self {
+        Self { inner, token }
+    }
+}
+
+impl<S, Body, Response> Service<Request<Body>> for AuthService<S>
+where
+    S: Service<Request<Body>, Response = Response>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    #[inline]
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    #[inline]
+    fn call(&mut self, mut request: Request<Body>) -> Self::Future {
+        if let Some(token) = &self.token {
+            request
+                .headers_mut()
+                .insert(AUTHORIZATION, token.as_ref().clone());
+        }
+
+        self.inner.call(request)
+    }
+}
diff --git a/src/client.rs b/src/client.rs
@@ -33,9 +33,8 @@ use crate::rpc::maintenance::{
 use crate::rpc::watch::{WatchClient, WatchOptions, WatchStream, Watcher};
 #[cfg(feature = "tls")]
 use crate::TlsOptions;
-use tonic::metadata::{Ascii, MetadataValue};
+use std::sync::Arc;
 use tonic::transport::Channel;
-use tonic::Interceptor;
 
 const HTTP_PREFIX: &str = "http://";
 const HTTPS_PREFIX: &str = "https://";
@@ -129,33 +128,23 @@ impl Client {
             _ => Channel::balance_list(endpoints.into_iter()),
         };
 
-        let interceptor = if let Some(connect_options) = options {
-            if let Some((name, password)) = connect_options.user {
+        let auth_token: Option<Arc<http::HeaderValue>> =
+            if let Some((name, password)) = options.and_then(|options| options.user) {
                 let mut tmp_auth = AuthClient::new(channel.clone(), None);
                 let resp = tmp_auth.authenticate(name, password).await?;
-                let token: MetadataValue<Ascii> = resp.token().parse()?;
-
-                Some(Interceptor::new(move |mut request| {
-                    let metadata = request.metadata_mut();
-                    // authorization for http::header::AUTHORIZATION
-                    metadata.insert("authorization", token.clone());
-                    Ok(request)
-                }))
+                Some(Arc::new(resp.token().parse()?))
             } else {
                 None
-            }
-        } else {
-            None
-        };
+            };
 
-        let kv = KvClient::new(channel.clone(), interceptor.clone());
-        let watch = WatchClient::new(channel.clone(), interceptor.clone());
-        let lease = LeaseClient::new(channel.clone(), interceptor.clone());
-        let lock = LockClient::new(channel.clone(), interceptor.clone());
-        let auth = AuthClient::new(channel.clone(), interceptor.clone());
-        let cluster = ClusterClient::new(channel.clone(), interceptor.clone());
-        let maintenance = MaintenanceClient::new(channel.clone(), interceptor.clone());
-        let election = ElectionClient::new(channel, interceptor);
+        let kv = KvClient::new(channel.clone(), auth_token.clone());
+        let watch = WatchClient::new(channel.clone(), auth_token.clone());
+        let lease = LeaseClient::new(channel.clone(), auth_token.clone());
+        let lock = LockClient::new(channel.clone(), auth_token.clone());
+        let auth = AuthClient::new(channel.clone(), auth_token.clone());
+        let cluster = ClusterClient::new(channel.clone(), auth_token.clone());
+        let maintenance = MaintenanceClient::new(channel.clone(), auth_token.clone());
+        let election = ElectionClient::new(channel, auth_token);
 
         Ok(Self {
             kv,

diff --git a/src/error.rs b/src/error.rs
@@ -35,8 +35,8 @@ pub enum Error {
     /// Election error
     ElectError(String),
 
-    /// Invalid metadata value
-    InvalidMetadataValue(tonic::metadata::errors::InvalidMetadataValue),
+    /// Invalid header value
+    InvalidHeaderValue(http::header::InvalidHeaderValue),
 }
 
 impl Display for Error {
@@ -52,7 +52,7 @@ impl Display for Error {
             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::InvalidMetadataValue(e) => write!(f, "invalid metadata value: {}", e),
+            Error::InvalidHeaderValue(e) => write!(f, "invalid metadata value: {}", e),
         }
     }
 }
@@ -94,9 +94,9 @@ impl From<Utf8Error> for Error {
     }
 }
 
-impl From<tonic::metadata::errors::InvalidMetadataValue> for Error {
+impl From<http::header::InvalidHeaderValue> for Error {
     #[inline]
-    fn from(e: tonic::metadata::errors::InvalidMetadataValue) -> Self {
-        Error::InvalidMetadataValue(e)
+    fn from(e: http::header::InvalidHeaderValue) -> Self {
+        Error::InvalidHeaderValue(e)
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
@@ -55,6 +55,7 @@
 
 #![cfg_attr(docsrs, feature(doc_cfg))]
 
+mod auth;
 mod client;
 mod error;
 mod rpc;

diff --git a/src/rpc/auth.rs b/src/rpc/auth.rs
@@ -2,6 +2,7 @@
 
 pub use crate::rpc::pb::authpb::permission::Type as PermissionType;
 
+use crate::auth::AuthService;
 use crate::error::Result;
 use crate::rpc::pb::authpb::{Permission as PbPermission, UserAddOptions as PbUserAddOptions};
 use crate::rpc::pb::etcdserverpb::auth_client::AuthClient as PbAuthClient;
@@ -32,26 +33,23 @@ use crate::rpc::pb::etcdserverpb::{
 };
 use crate::rpc::ResponseHeader;
 use crate::rpc::{get_prefix, KeyRange};
-use std::string::String;
+use http::HeaderValue;
+use std::{string::String, sync::Arc};
 use tonic::transport::Channel;
-use tonic::{Interceptor, IntoRequest, Request};
+use tonic::{IntoRequest, Request};
 
 /// Client for Auth operations.
 #[repr(transparent)]
 #[derive(Clone)]
 pub struct AuthClient {
-    inner: PbAuthClient<Channel>,
+    inner: PbAuthClient<AuthService<Channel>>,
 }
 
 impl AuthClient {
     /// Creates an auth client.
     #[inline]
-    pub(crate) fn new(channel: Channel, interceptor: Option<Interceptor>) -> Self {
-        let inner = match interceptor {
-            Some(it) => PbAuthClient::with_interceptor(channel, it),
-            None => PbAuthClient::new(channel),
-        };
-
+    pub(crate) fn new(channel: Channel, auth_token: Option<Arc<HeaderValue>>) -> Self {
+        let inner = PbAuthClient::new(AuthService::new(channel, auth_token));
         Self { inner }
     }
 

diff --git a/src/rpc/cluster.rs b/src/rpc/cluster.rs
@@ -1,5 +1,6 @@
 //! Etcd Cluster RPC.
 
+use crate::auth::AuthService;
 use crate::error::Result;
 use crate::rpc::pb::etcdserverpb::cluster_client::ClusterClient as PbClusterClient;
 use crate::rpc::pb::etcdserverpb::{
@@ -11,26 +12,23 @@ use crate::rpc::pb::etcdserverpb::{
     MemberUpdateResponse as PbMemberUpdateResponse,
 };
 use crate::rpc::ResponseHeader;
-use std::string::String;
+use http::HeaderValue;
+use std::{string::String, sync::Arc};
 use tonic::transport::Channel;
-use tonic::{Interceptor, IntoRequest, Request};
+use tonic::{IntoRequest, Request};
 
 /// Client for Cluster operations.
 #[repr(transparent)]
 #[derive(Clone)]
 pub struct ClusterClient {
-    inner: PbClusterClient<Channel>,
+    inner: PbClusterClient<AuthService<Channel>>,
 }
 
 impl ClusterClient {
     /// Creates an Cluster client.
     #[inline]
-    pub(crate) fn new(channel: Channel, interceptor: Option<Interceptor>) -> Self {
-        let inner = match interceptor {
-            Some(it) => PbClusterClient::with_interceptor(channel, it),
-            None => PbClusterClient::new(channel),
-        };
-
+    pub(crate) fn new(channel: Channel, auth_token: Option<Arc<HeaderValue>>) -> Self {
+        let inner = PbClusterClient::new(AuthService::new(channel, auth_token));
         Self { inner }
     }
 

diff --git a/src/rpc/election.rs b/src/rpc/election.rs
@@ -1,5 +1,6 @@
 //! Etcd Election RPC.
 
+use crate::auth::AuthService;
 use crate::error::Result;
 use crate::rpc::pb::v3electionpb::election_client::ElectionClient as PbElectionClient;
 use crate::rpc::pb::v3electionpb::{
@@ -9,17 +10,18 @@ use crate::rpc::pb::v3electionpb::{
     ResignRequest as PbResignRequest, ResignResponse as PbResignResponse,
 };
 use crate::rpc::{KeyValue, ResponseHeader};
-use std::pin::Pin;
+use http::HeaderValue;
 use std::task::{Context, Poll};
+use std::{pin::Pin, sync::Arc};
 use tokio_stream::Stream;
 use tonic::transport::Channel;
-use tonic::{Interceptor, IntoRequest, Request, Streaming};
+use tonic::{IntoRequest, Request, Streaming};
 
 /// Client for Elect operations.
 #[repr(transparent)]
 #[derive(Clone)]
 pub struct ElectionClient {
-    inner: PbElectionClient<Channel>,
+    inner: PbElectionClient<AuthService<Channel>>,
 }
 
 /// Options for `campaign` operation.
@@ -472,15 +474,9 @@ impl From<&PbLeaderKey> for &LeaderKey {
 impl ElectionClient {
     /// Creates a election
     #[inline]
-    pub(crate) fn new(channel: Channel, interceptor: Option<Interceptor>) -> Self {
-        let inner_channel = match interceptor {
-            Some(it) => PbElectionClient::with_interceptor(channel, it),
-            None => PbElectionClient::new(channel),
-        };
-
-        Self {
-            inner: inner_channel,
-        }
+    pub(crate) fn new(channel: Channel, auth_token: Option<Arc<HeaderValue>>) -> Self {
+        let inner = PbElectionClient::new(AuthService::new(channel, auth_token));
+        Self { inner }
     }
 
     /// Puts a value as eligible for the election on the prefix key.

diff --git a/src/rpc/kv.rs b/src/rpc/kv.rs
@@ -3,6 +3,7 @@
 pub use crate::rpc::pb::etcdserverpb::compare::CompareResult as CompareOp;
 pub use crate::rpc::pb::etcdserverpb::range_request::{SortOrder, SortTarget};
 
+use crate::auth::AuthService;
 use crate::error::Result;
 use crate::rpc::pb::etcdserverpb::compare::{CompareTarget, TargetUnion};
 use crate::rpc::pb::etcdserverpb::kv_client::KvClient as PbKvClient;
@@ -17,25 +18,23 @@ use crate::rpc::pb::etcdserverpb::{
     RequestOp as PbTxnRequestOp, TxnRequest as PbTxnRequest, TxnResponse as PbTxnResponse,
 };
 use crate::rpc::{get_prefix, KeyRange, KeyValue, ResponseHeader};
+use http::HeaderValue;
+use std::sync::Arc;
 use tonic::transport::Channel;
-use tonic::{Interceptor, IntoRequest, Request};
+use tonic::{IntoRequest, Request};
 
 /// Client for KV operations.
 #[repr(transparent)]
 #[derive(Clone)]
 pub struct KvClient {
-    inner: PbKvClient<Channel>,
+    inner: PbKvClient<AuthService<Channel>>,
 }
 
 impl KvClient {
     /// Creates a kv client.
     #[inline]
-    pub(crate) fn new(channel: Channel, interceptor: Option<Interceptor>) -> Self {
-        let inner = match interceptor {
-            Some(it) => PbKvClient::with_interceptor(channel, it),
-            None => PbKvClient::new(channel),
-        };
-
+    pub(crate) fn new(channel: Channel, auth_token: Option<Arc<HeaderValue>>) -> Self {
+        let inner = PbKvClient::new(AuthService::new(channel, auth_token));
         Self { inner }
     }
 

diff --git a/src/rpc/lease.rs b/src/rpc/lease.rs
@@ -1,5 +1,6 @@
 //! Etcd Lease RPC.
 
+use crate::auth::AuthService;
 use crate::error::Result;
 use crate::rpc::pb::etcdserverpb::lease_client::LeaseClient as PbLeaseClient;
 use crate::rpc::pb::etcdserverpb::{
@@ -11,34 +12,30 @@ use crate::rpc::pb::etcdserverpb::{
     LeaseTimeToLiveRequest as PbLeaseTimeToLiveRequest,
     LeaseTimeToLiveResponse as PbLeaseTimeToLiveResponse,
 };
-
 use crate::rpc::ResponseHeader;
 use crate::Error;
-
+use http::HeaderValue;
 use std::pin::Pin;
+use std::sync::Arc;
 use std::task::{Context, Poll};
 use tokio::sync::mpsc::{channel, Sender};
 use tokio_stream::wrappers::ReceiverStream;
 use tokio_stream::Stream;
 use tonic::transport::Channel;
-use tonic::{Interceptor, IntoRequest, Request, Streaming};
+use tonic::{IntoRequest, Request, Streaming};
 
 /// Client for lease operations.
 #[repr(transparent)]
 #[derive(Clone)]
 pub struct LeaseClient {
-    inner: PbLeaseClient<Channel>,
+    inner: PbLeaseClient<AuthService<Channel>>,
 }
 
 impl LeaseClient {
     /// Creates a `LeaseClient`.
     #[inline]
-    pub(crate) fn new(channel: Channel, interceptor: Option<Interceptor>) -> Self {
-        let inner = match interceptor {
-            Some(it) => PbLeaseClient::with_interceptor(channel, it),
-            None => PbLeaseClient::new(channel),
-        };
-
+    pub(crate) fn new(channel: Channel, auth_token: Option<Arc<HeaderValue>>) -> Self {
+        let inner = PbLeaseClient::new(AuthService::new(channel, auth_token));
         Self { inner }
     }