Merge pull request #438 from Migorithm/feat/435

feat: exists command on cli

Author: Migorithm

duva-client/src/command.rs

@@ -48,6 +48,14 @@ pub fn take_input(action: &str, args: &[&str]) -> Result<ClientInputKind, String
             }
             Ok(ClientInputKind::Del)
         },
+        "EXISTS" => {
+            if args.len() < 1 {
+                return Err(
+                    "(error) ERR wrong number of arguments for 'exists' command".to_string()
+                );
+            }
+            Ok(ClientInputKind::Exists)
+        },
 
         "PING" => Ok(ClientInputKind::Ping),
         "ECHO" => {
@@ -106,4 +114,5 @@ pub enum ClientInputKind {
     ClusterInfo,
     ClusterNodes,
     ClusterForget,
+    Exists,
 }

duva-client/src/controller.rs

@@ -132,14 +132,13 @@ impl<T> ClientController<T> {
                     return Err("Unexpected response format".to_string());
                 },
             },
-            Del => {
+            Del | Exists => {
                 let QueryIO::SimpleString(value) = query_io else {
                     return Err("Unexpected response format".to_string());
                 };
                 let deleted_count = value.parse::<u64>().unwrap();
                 println!("(integer) {}", deleted_count);
             },
-
             Set => {
                 let v = match query_io {
                     QueryIO::SimpleString(value) => value,
@@ -154,7 +153,6 @@ impl<T> ClientController<T> {
                 self.latest_known_index = rindex.parse::<u64>().unwrap();
                 println!("OK");
             },
-
             ClusterNodes => {
                 let QueryIO::Array(value) = query_io else {
                     return Err("Unexpected response format".to_string());

duva/src/domains/caches/actor.rs

@@ -60,6 +60,13 @@ impl CacheActor {
             let _ = callback.send(false);
         }
     }
+    pub(crate) fn exists(&self, key: String, callback: oneshot::Sender<bool>) {
+        if self.cache.get(&key).is_some() {
+            let _ = callback.send(true);
+        } else {
+            let _ = callback.send(false);
+        }
+    }
     pub(crate) fn get(&self, key: &str, callback: oneshot::Sender<QueryIO>) {
         let _ = callback.send(self.cache.get(key).cloned().into());
     }

duva/src/domains/caches/cache_manager.rs

@@ -14,8 +14,11 @@ use chrono::Utc;
 use futures::StreamExt;
 use futures::future::join_all;
 use futures::stream::FuturesUnordered;
+use tokio::sync::oneshot::error::RecvError;
+
 use std::sync::Arc;
 use std::sync::atomic::AtomicU64;
+
 use std::{hash::Hasher, iter::Zip};
 use tokio::sync::oneshot::Sender;
 use tokio::task::JoinHandle;
@@ -169,27 +172,43 @@ impl CacheManager {
     }
 
     pub(crate) async fn route_delete(&self, keys: Vec<String>) -> Result<u64> {
+        let closure = |key, callback| -> CacheCommand { CacheCommand::Delete { key, callback } };
         // Create futures for all delete operations at once
-        let results = FuturesUnordered::from_iter(keys.into_iter().map(|key| {
-            let (tx, rx) = tokio::sync::oneshot::channel();
-            async move {
-                let _ =
-                    self.select_shard(&key).send(CacheCommand::Delete { key, callback: tx }).await;
-                rx.await
-            }
-        }))
-        .collect::<Vec<_>>()
-        .await;
+        let results = self.send_selectively(keys, closure).await;
 
         let deleted = results.into_iter().filter_map(|r| r.ok().filter(|&success| success)).count();
-
         Ok(deleted as u64)
     }
+    pub(crate) async fn route_exists(&self, keys: Vec<String>) -> Result<u64> {
+        let closure = |key, callback| -> CacheCommand { CacheCommand::Exists { key, callback } };
+        // Create futures for all delete operations at once
+        let results = self.send_selectively(keys, closure).await;
+
+        let found = results.into_iter().filter_map(|r| r.ok().filter(|&success| success)).count();
+        Ok(found as u64)
+    }
+
     pub(crate) fn select_shard(&self, key: &str) -> &CacheCommandSender {
         let shard_key = self.take_shard_key_from_str(key);
         &self.inboxes[shard_key]
     }
 
+    async fn send_selectively<T>(
+        &self,
+        keys: Vec<String>,
+        func: fn(String, Sender<T>) -> CacheCommand,
+    ) -> Vec<Result<T, RecvError>> {
+        FuturesUnordered::from_iter(keys.into_iter().map(|key| {
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            async move {
+                let _ = self.select_shard(&key).send(func(key, tx)).await;
+                rx.await
+            }
+        }))
+        .collect::<Vec<_>>()
+        .await
+    }
+
     fn take_shard_key_from_str(&self, s: &str) -> usize {
         let mut hasher = std::hash::DefaultHasher::new();
         std::hash::Hash::hash(&s, &mut hasher);

duva/src/domains/caches/command.rs

@@ -13,4 +13,5 @@ pub enum CacheCommand {
     IndexGet { key: String, read_idx: u64, callback: oneshot::Sender<QueryIO> },
     Ping,
     StopSentinel,
+    Exists { key: String, callback: oneshot::Sender<bool> },
 }

duva/src/presentation/clients/controllers/handler.rs

@@ -68,6 +68,10 @@ impl ClientController<Handler> {
             ClientAction::Delete { keys } => {
                 QueryIO::SimpleString(self.cache_manager.route_delete(keys).await?.to_string())
             },
+
+            ClientAction::Exists { keys } => {
+                QueryIO::SimpleString(self.cache_manager.route_exists(keys).await?.to_string())
+            },
             ClientAction::Info => QueryIO::BulkString(
                 self.cluster_communication_manager
                     .replication_info()

duva/src/presentation/clients/parser.rs

@@ -28,6 +28,7 @@ pub fn parse_query(
             }
         },
         ("del", keys) => ClientAction::Delete { keys: keys.to_vec() },
+        ("exists", keys) => ClientAction::Exists { keys: keys.to_vec() },
         ("echo", [value]) => ClientAction::Echo(value.to_string()),
         ("config", [key, value]) => {
             ClientAction::Config { key: key.to_string(), value: value.to_string() }

duva/src/presentation/clients/request.rs

@@ -20,6 +20,7 @@ pub(crate) enum ClientAction {
     ClusterInfo,
     ClusterNodes,
     ClusterForget(PeerIdentifier),
+    Exists { keys: Vec<String> },
 }
 
 impl ClientAction {

duva/src/services/handlers/cache.rs

@@ -38,6 +38,9 @@ impl CacheActor {
                 CacheCommand::Delete { key, callback } => {
                     self.delete(key, callback);
                 },
+                CacheCommand::Exists { key, callback } => {
+                    self.exists(key, callback);
+                },
                 CacheCommand::Save { outbox } => {
                     outbox
                         .send(SaveCommand::LocalShardSize {

duva/tests/test_exists.rs

@@ -0,0 +1,36 @@
+/// The following is to test out the set operation with expiry
+/// Firstly, we set a key with a value and an expiry of 300ms
+/// Then we get the key and check if the value is returned
+/// After 300ms, we get the key again and check if the value is not returned (-1)
+mod common;
+use common::{ServerEnv, array, spawn_server_process};
+
+use duva::{clients::ClientStreamHandler, domains::query_parsers::query_io::QueryIO};
+
+#[tokio::test]
+async fn test_exists() {
+    // GIVEN
+    let env = ServerEnv::default();
+    let process = spawn_server_process(&env);
+
+    let mut h = ClientStreamHandler::new(process.bind_addr()).await;
+    assert_eq!(
+        h.send_and_get(&array(vec!["SET", "a", "b"])).await,
+        QueryIO::SimpleString("OK RINDEX 1".into()).serialize()
+    );
+    assert_eq!(
+        h.send_and_get(&array(vec!["SET", "c", "d"])).await,
+        QueryIO::SimpleString("OK RINDEX 2".into()).serialize()
+    );
+
+    // WHEN & THEN
+    assert_eq!(
+        h.send_and_get(&array(vec!["exists", "a", "c", "d"])).await,
+        QueryIO::SimpleString("2".into()).serialize() // 2 means 2 keys deleted
+    );
+
+    assert_eq!(
+        h.send_and_get(&array(vec!["exists", "x"])).await,
+        QueryIO::SimpleString("0".into()).serialize() // 2 means 2 keys deleted
+    );
+}