feat: implement etcd client

Create a new internal_etcd interface witch contains all methods
related to the ETCD client.

Signed-off-by: GridexX <[email protected]>

controller/lib/Cargo.toml

@@ -6,9 +6,10 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+etcd-client = "0.9.2"
 actix-web = "4.1.0"
 serde = { version = "1.0.139", features = ["derive"] }
 tonic = "0.7.2"
 proto = { path = "../../proto" }
 log = "0.4.0"
-tokio = { version = "1.20.0", features = ["rt-multi-thread"] }
+tokio = { version = "1.20.0", features = ["rt-multi-thread", "macros"] }

controller/lib/src/etcd/mod.rs

@@ -0,0 +1,135 @@
+use etcd_client::{Client, DeleteResponse, Error, GetOptions, PutResponse};
+use log::{debug, error, info};
+
+pub struct EtcdClient {
+  inner: Client,
+}
+
+impl EtcdClient {
+  pub async fn new(address: String) -> Self {
+    info!("Starting ETCD client on {}", address);
+    EtcdClient {
+      inner: Client::connect([address], None).await.unwrap(),
+    }
+  }
+
+  pub async fn get(&mut self, key: &str) -> Result<String, Error> {
+    if let Some(kv) = self.inner.get(key, None).await?.kvs().first() {
+      let res = kv.value_str();
+      info!(
+        "Retrieving value in ETCD : Key \"{}\" is associated with value \"{}\"",
+        key,
+        res.as_ref().unwrap()
+      );
+      res.map(|s| s.to_string())
+    } else {
+      Err(Error::from(std::io::Error::new(
+        std::io::ErrorKind::NotFound,
+        "Key not found",
+      )))
+    }
+  }
+  pub async fn put(&mut self, key: &str, value: &str) -> Result<PutResponse, Error> {
+    info!(
+      "Inserting value in ETCD : Key \"{}\" associated with value \"{}\"",
+      key, value
+    );
+    self.inner.put(key, value, None).await
+  }
+  pub async fn patch(&mut self, key: &str, value: &str) -> Result<PutResponse, Error> {
+    match self.get(key).await {
+      Ok(_) => {
+        info!(
+          "Updating value in ETCD : Key \"{}\" associated with new value \"{}\"",
+          key, value
+        );
+        self.inner.put(key, value, None).await
+      }
+      Err(err) => Err(err),
+    }
+  }
+  pub async fn delete(&mut self, key: &str) -> Option<DeleteResponse> {
+    if let Some(_) = self.inner.get(key, None).await.unwrap().kvs().first() {
+      info!("Deleting value in ETCD : Key \"{}\" ", key);
+      match self.inner.delete(key, None).await {
+        Ok(delete) => {
+          Some(delete)
+        }
+        Err(_) => None
+      }
+    } else {
+      None
+    }
+  }
+
+  pub async fn get_all(&mut self) -> Result<Vec<String>, Error> {
+    info!("Retrieving all keys in ETCD");
+    let resp = self
+      .inner
+      .get("", Some(GetOptions::new().with_all_keys()))
+      .await?;
+
+    let mut values: Vec<String> = vec![];
+    for kv in resp.kvs() {
+      let value = kv.value_str()?;
+      values.push(value.to_string())
+    }
+    Ok(values)
+  }
+}
+/*
+
+#[cfg(test)]
+mod tests {
+
+  use crate::etcd::EtcdClient;
+  use etcd_client::Error;
+
+  #[tokio::test]
+  async fn test_value_insertion() -> Result<(), Error> {
+    let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await;
+    let _res = etcd_client.put("foo", "bar").await?;
+    let resp = etcd_client.get("foo").await?;
+    assert_eq!(resp, "bar");
+    Ok(())
+  }
+
+  #[tokio::test]
+  async fn test_value_modification() -> Result<(), Error> {
+    let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await;
+    etcd_client.put("foo", "bar").await?;
+    let _res = etcd_client.patch("foo", "baz").await?;
+    let resp = etcd_client.get("foo").await?;
+    assert_eq!(resp, "baz");
+    Ok(())
+  }
+  #[tokio::test]
+  async fn test_value_deletion() -> Result<(), Error> {
+    let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await;
+    let _res = etcd_client.put("foo", "bar").await?;
+    let _res = etcd_client.delete("foo").await?;
+    let err = etcd_client.get("foo").await;
+    assert!(err.is_err());
+    Ok(())
+  }
+  #[tokio::test]
+  async fn test_value_deletion_doesnt_exists() -> Result<(), Error> {
+    let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await;
+    let _res = etcd_client.put("foo", "bar").await?;
+    let err = etcd_client.delete("foo2").await;
+    assert!(err.is_err());
+    Ok(())
+  }
+
+  #[tokio::test]
+  async fn test_function_get_all() -> Result<(), Error> {
+    let mut etcd_client = EtcdClient::new("localhost:2379".to_string()).await;
+    let _res = etcd_client.put("bar", "foo").await;
+    let _res = etcd_client.put("hello", "world").await;
+    let values = etcd_client.get_all().await?;
+    assert_eq!(values[0], "foo");
+    assert_eq!(values[1], "world");
+    Ok(())
+  }
+}
+*/

controller/lib/src/lib.rs

@@ -1,2 +1,3 @@
+pub mod etcd;
 pub mod external_api;
 pub mod internal_api;