dapr key value implementation

Signed-off-by: David Justice <[email protected]>

Cargo.toml

@@ -17,7 +17,7 @@ path = "src/lib.rs"
 slight-blob-store = { workspace = true, features = ["aws_s3"], optional = true }
 slight-core = { workspace = true }
 slight-runtime = { workspace = true }
-slight-keyvalue = { workspace = true, features = ["filesystem", "awsdynamodb", "redis", "azblob"], optional = true}
+slight-keyvalue = { workspace = true, features = ["filesystem", "awsdynamodb", "redis", "azblob", "dapr"], optional = true}
 slight-distributed-locking = { workspace = true, features = ["etcd"], optional = true}
 slight-messaging = { workspace = true, features = ["filesystem", "mosquitto", "azsbus", "natsio"], optional = true}
 slight-runtime-configs = { workspace = true, optional = true }

crates/keyvalue/Cargo.toml

@@ -31,10 +31,13 @@ aws-config = { version = "0.54", optional = true }
 aws-sdk-dynamodb = { version = "0.24", optional = true }
 # kv.redis deps
 redis = { version = "0.22", optional = true }
+# kv.dapr deps
+dapr = { version ="0.11.0", optional = true }
 
 [features]
 default = ["filesystem"]
 filesystem = ["serde_json"]
 azblob = ["azure_storage_blobs", "azure_storage", "bytes", "futures"]
 awsdynamodb = ["aws-config", "aws-sdk-dynamodb"]
 redis = ["dep:redis"]
+dapr = ["dep:dapr"]

crates/keyvalue/src/implementors/dapr.rs

@@ -0,0 +1,73 @@
+use std::cell::RefCell;
+use std::fmt::{Debug, Formatter};
+use std::sync::{Arc};
+use tokio::sync::Mutex;
+use anyhow::{bail, Result};
+use async_trait::async_trait;
+use dapr::{Client, client::TonicClient};
+use slight_common::BasicState;
+use slight_runtime_configs::get_from_state;
+
+use super::KeyvalueImplementor;
+
+/// This is the underlying struct behind the `Dapr` variant of the `KeyvalueImplementor` enum.
+///
+/// As per its' usage in `KeyvalueImplementor`, it must `derive` `Debug`, and `Clone`.
+#[derive(Clone)]
+pub struct DaprImplementor {
+ client: Arc<Mutex<RefCell<Client<TonicClient>>>>,
+ container_name: String,
+}
+
+impl Debug for DaprImplementor {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.write_str(format!("[DaprImplementor] container_name: {}", self.container_name).as_str())
+ }
+}
+
+impl DaprImplementor {
+ pub async fn new(slight_state: &BasicState, name: &str) -> Self {
+ let connection_string = get_from_state("DAPR_ADDRESS", slight_state).await.unwrap();
+ let client = Client::connect(connection_string).await.unwrap();
+ let container_name = name.to_string();
+ let internal_mut_client = Arc::new(Mutex::new(RefCell::new(client)));
+ Self {
+ client: internal_mut_client,
+ container_name,
+ }
+ }
+}
+
+#[async_trait]
+impl KeyvalueImplementor for DaprImplementor {
+ async fn get(&self, key: &str) -> Result<Vec<u8>> {
+ let container = self.container_name.clone();
+ let mut client = self.client.lock().await;
+ let res = client.get_mut().get_state(container, key.to_string(), None).await?;
+
+ if res.data.is_empty() {
+ bail!("key not found");
+ }
+ Ok(res.data)
+ }
+
+ async fn set(&self, key: &str, value: &[u8]) -> Result<()> {
+ let container = self.container_name.clone();
+ let mut client = self.client.as_ref().lock().await;
+ client.get_mut().save_state(container, vec![(key.to_string(), value.to_vec())]).await?;
+
+ Ok(())
+ }
+
+ async fn keys(&self) -> Result<Vec<String>> {
+ bail!("not implemented");
+ }
+
+ async fn delete(&self, key: &str) -> Result<()> {
+ let container = self.container_name.clone();
+ let mut client = self.client.lock().await;
+ client.get_mut().delete_state(container, key.to_string(), None).await?;
+
+ Ok(())
+ }
+}

crates/keyvalue/src/implementors/mod.rs

@@ -9,6 +9,8 @@ pub mod azblob;
 pub mod filesystem;
 #[cfg(feature = "redis")]
 pub mod redis;
+#[cfg(feature = "dapr")]
+pub mod dapr;
 
 #[async_trait]
 pub trait KeyvalueImplementor {

crates/keyvalue/src/lib.rs

@@ -86,6 +86,10 @@ impl KeyvalueInner {
  KeyvalueImplementors::Redis => {
  Arc::new(redis::RedisImplementor::new(slight_state, name).await)
  }
+ #[cfg(feature = "dapr")]
+ KeyvalueImplementors::Dapr => {
+ Arc::new(dapr::DaprImplementor::new(slight_state, name).await)
+ }
  },
  }
  }
@@ -105,6 +109,8 @@ pub enum KeyvalueImplementors {
  AwsDynamoDb,
  #[cfg(feature = "redis")]
  Redis,
+ #[cfg(feature = "dapr")]
+ Dapr,
 }
 
 impl From<&str> for KeyvalueImplementors {