feat: allow caller to initialize the channel (#93)

In some scenarios, such as when using a Deterministic Simulation
Testing environment such as `turmoil`, when mocking the behaviour of
etcd itself, or using non-Tokio sockets for connecting to etcd, it is
desirable to construct a Tonic channel externally, then "wrap" it in a
`Client` object. This patch enables this use case by adding a
`raw-channel` Cargo feature, which enables the `Client::from_channel`
function.

Co-authored-by: Nathan Moos <nmoos@cisco.com>

Author: nmooscisco

Cargo.toml

@@ -19,6 +19,7 @@ tls-openssl-vendored = ["tls-openssl", "openssl/vendored"]
 tls-roots = ["tls", "tonic/tls-roots"]
 pub-response-field = ["visible"]
 build-server = ["pub-response-field"]
+raw-channel = []
 
 [dependencies]
 tonic = "0.12.3"

README.md

@@ -73,6 +73,7 @@ Examples can be found in [`examples`](./examples).
 - `tls-openssl`: Enables the `openssl`-based TLS connections. This would make your binary dynamically link to `libssl`.
 - `tls-openssl-vendored`: Like `tls-openssl`, however compile openssl from source code and statically link to it.
 - `build-server`: Builds a server variant of the etcd protobuf and re-exports it under the same `proto` package as the `pub-response-field` feature does.
+- `raw-channel`: Allows the caller to construct the underlying Tonic channel used by the client.
 
 ## Test
 

src/client.rs

@@ -1,5 +1,7 @@
 //! Asynchronous client & synchronous client.
 
+#[cfg(feature = "raw-channel")]
+use crate::channel::Channel;
 use crate::error::{Error, Result};
 use crate::intercept::{InterceptedChannel, Interceptor};
 use crate::lock::RwLockExt;
@@ -66,7 +68,7 @@ pub struct Client {
     cluster: ClusterClient,
     election: ElectionClient,
     options: Option<ConnectOptions>,
-    tx: Sender<Change<Uri, Endpoint>>,
+    tx: Option<Sender<Change<Uri, Endpoint>>>,
 }
 
 impl Client {
@@ -130,7 +132,24 @@ impl Client {
         let auth_token = Arc::new(RwLock::new(None));
         Self::auth(channel.clone(), &mut options, &auth_token).await?;
 
-        Ok(Self::build_client(channel, tx, auth_token, options))
+        Ok(Self::build_client(channel, Some(tx), auth_token, options))
+    }
+
+    #[cfg(feature = "raw-channel")]
+    /// Connect to `etcd` servers represented by the given `channel`.
+    pub async fn from_channel(channel: Channel, options: Option<ConnectOptions>) -> Result<Self> {
+        let channel = InterceptedChannel::new(
+            channel,
+            Interceptor {
+                require_leader: options.as_ref().map(|o| o.require_leader).unwrap_or(false),
+            },
+        );
+        let mut options = options;
+
+        let auth_token = Arc::new(RwLock::new(None));
+        Self::auth(channel.clone(), &mut options, &auth_token).await?;
+
+        Ok(Self::build_client(channel, None, auth_token, options))
     }
 
     fn build_endpoint(url: &str, options: &Option<ConnectOptions>) -> Result<Endpoint> {
@@ -255,7 +274,7 @@ impl Client {
 
     fn build_client(
         channel: InterceptedChannel,
-        tx: Sender<Change<Uri, Endpoint>>,
+        tx: Option<Sender<Change<Uri, Endpoint>>>,
         auth_token: Arc<RwLock<Option<HeaderValue>>>,
         options: Option<ConnectOptions>,
     ) -> Self {
@@ -295,7 +314,9 @@ impl Client {
     #[inline]
     pub async fn add_endpoint<E: AsRef<str>>(&self, endpoint: E) -> Result<()> {
         let endpoint = Self::build_endpoint(endpoint.as_ref(), &self.options)?;
-        let tx = &self.tx;
+        let Some(tx) = &self.tx else {
+            return Err(Error::EndpointsNotManaged);
+        };
         tx.send(Change::Insert(endpoint.uri().clone(), endpoint))
             .await
             .map_err(|e| Error::EndpointError(format!("failed to add endpoint because of {}", e)))
@@ -309,7 +330,9 @@ impl Client {
     #[inline]
     pub async fn remove_endpoint<E: AsRef<str>>(&self, endpoint: E) -> Result<()> {
         let uri = http::Uri::from_str(endpoint.as_ref())?;
-        let tx = &self.tx;
+        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))
         })

src/error.rs

@@ -41,6 +41,9 @@ pub enum Error {
     /// Endpoint error
     EndpointError(String),
 
+    /// Endpoint set is not managed by this client
+    EndpointsNotManaged,
+
     /// OpenSSL errors.
     #[cfg(feature = "tls-openssl")]
     OpenSsl(openssl::error::ErrorStack),
@@ -61,6 +64,7 @@ impl Display for Error {
             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),
         }

src/lib.rs

@@ -54,6 +54,7 @@
 //! - `tls-openssl`: Enables the `openssl`-based TLS connections. This would make your binary dynamically link to `libssl`.
 //! - `tls-openssl-vendored`: Like `tls-openssl`, however compile openssl from source code and statically link to it.
 //! - `build-server`: Builds a server variant of the etcd protobuf and re-exports it under the same `proto` package as the `pub-response-field` feature does.
+//! - `raw-channel`: Allows the caller to construct the underlying Tonic channel used by the client.
 
 #![cfg_attr(docsrs, feature(doc_cfg))]