feat: implement resp DEL command

src/entrystore/src/seg/resp.rs

@@ -15,6 +15,7 @@ use std::time::Duration;
 impl Execute<Request, Response> for Seg {
     fn execute(&mut self, request: &Request) -> Response {
         match request {
+            Request::Del(del) => self.del(del),
             Request::Get(get) => self.get(get),
             Request::Set(set) => self.set(set),
             _ => Response::error("not supported"),
@@ -23,6 +24,16 @@ impl Execute<Request, Response> for Seg {
 }
 
 impl Storage for Seg {
+    fn del(&mut self, delete: &Del) -> Response {
+        let count = delete
+            .keys()
+            .iter()
+            .filter(|key| self.data.delete(key))
+            .count();
+        // according to https://stackoverflow.com/questions/42911672/is-there-any-limit-on-the-number-of-arguments-that-redis-commands-such-as-zadd-o#:~:text=The%20maximum%20number%20of%20arguments,long%20meaning%20up%20to%202%2C147%2C483%2C647.
+        // the max number of arguments to a Redis command is the max value of an i32, so this cast should be safe
+        Response::integer(count as i64)
+    }
     fn get(&mut self, get: &Get) -> Response {
         if let Some(item) = self.data.get(get.key()) {
             match item.value() {
@@ -51,3 +62,49 @@ impl Storage for Seg {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::Seg;
+    use config::RdsConfig;
+    use protocol_resp::*;
+
+    #[test]
+    fn it_should_del() {
+        let config = RdsConfig::default();
+        let mut seg = Seg::new(&config).expect("could not initialize seg");
+
+        //ignores missing
+        let del_missing = Del::new(&[b"missing"]);
+        let response = seg.del(&del_missing);
+        assert_eq!(response, Response::integer(0));
+
+        //deletes multiple keys, returning the number deleted
+        let mut keys = vec![];
+        for i in 1..5 {
+            let key_string = format!("k{i}");
+            let key = key_string.as_bytes();
+            let set = Set::new(key, format!("v{i}").as_bytes(), None, SetMode::Set, false);
+            seg.set(&set);
+            keys.push(key_string);
+        }
+
+        let del = Del::new(
+            keys.iter()
+                .map(|s| s.as_bytes())
+                .collect::<Vec<&[u8]>>()
+                .as_slice(),
+        );
+
+        let response = seg.del(&del);
+        assert_eq!(response, Response::integer(keys.len() as i64));
+
+        for key in keys {
+            let bytes = key.as_bytes();
+
+            let get = Get::new(bytes);
+            let response = seg.get(&get);
+            assert_eq!(response, Response::null())
+        }
+    }
+}

src/protocol/resp/src/request/del.rs

@@ -0,0 +1,97 @@
+// Copyright 2023 Pelikan Foundation LLC.
+// Licensed under the Apache License, Version 2.0
+// http://www.apache.org/licenses/LICENSE-2.0
+
+use std::io::Error;
+use std::sync::Arc;
+
+use super::*;
+
+#[derive(Debug, PartialEq, Eq)]
+pub struct Del {
+    keys: Vec<Arc<[u8]>>,
+}
+
+impl TryFrom<Message> for Del {
+    type Error = Error;
+
+    fn try_from(value: Message) -> std::io::Result<Self> {
+        let array = match value {
+            Message::Array(array) => array,
+            _ => return Err(Error::new(ErrorKind::Other, "malformed command")),
+        };
+
+        let mut array = array.inner.unwrap();
+        if array.len() < 3 {
+            return Err(Error::new(ErrorKind::Other, "malformed command"));
+        }
+
+        let _command = take_bulk_string(&mut array)?;
+
+        let mut keys = Vec::with_capacity(array.len());
+        while !array.is_empty() {
+            keys.push(
+                take_bulk_string(&mut array)?
+                    .ok_or_else(|| Error::new(ErrorKind::Other, "malformed command"))?,
+            );
+        }
+
+        Ok(Self { keys })
+    }
+}
+
+impl Del {
+    pub fn new(keys: &[&[u8]]) -> Self {
+        Self {
+            keys: keys.iter().copied().map(From::from).collect(),
+        }
+    }
+
+    pub fn keys(&self) -> &[Arc<[u8]>] {
+        &self.keys
+    }
+}
+
+impl From<&Del> for Message {
+    fn from(value: &Del) -> Self {
+        let mut vals = Vec::with_capacity(value.keys.len() + 1);
+
+        vals.push(Message::BulkString(BulkString::new(b"DEL")));
+        vals.extend(
+            value
+                .keys()
+                .iter()
+                .map(|v| Message::BulkString(BulkString::new(v))),
+        );
+
+        Message::Array(Array { inner: Some(vals) })
+    }
+}
+
+impl Compose for Del {
+    fn compose(&self, dst: &mut dyn BufMut) -> usize {
+        Message::from(self).compose(dst)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parser() {
+        let parser = RequestParser::new();
+        assert_eq!(
+            parser.parse(b"del k1 k2 k3 k4\r\n").unwrap().into_inner(),
+            Request::Del(Del::new(&[b"k1", b"k2", b"k3", b"k4"]))
+        );
+
+        assert_eq!(
+            parser
+                .parse(b"*4\r\n$3\r\ndel\r\n$2\r\nk1\r\n$2\r\nk2\r\n$2\r\nk3\r\n")
+                .unwrap()
+                .into_inner(),
+            Request::Del(Del::new(&[b"k1", b"k2", b"k3"]))
+        );
+    }
+}

src/protocol/resp/src/request/mod.rs

@@ -14,6 +14,7 @@ use std::io::{Error, ErrorKind};
 use std::sync::Arc;
 
 mod badd;
+mod del;
 mod get;
 mod hdel;
 mod hexists;
@@ -57,6 +58,7 @@ pub use self::smembers::*;
 pub use self::srem::*;
 pub use self::sunion::*;
 pub use badd::*;
+pub use del::*;
 pub use get::*;
 pub use hdel::*;
 pub use hexists::*;
@@ -194,6 +196,7 @@ macro_rules! decl_request {
 decl_request! {
     pub enum Request {
         BtreeAdd(BtreeAdd) => "badd",
+        Del(Del) => "del",
         Get(Get) => "get",
         HashDelete(HashDelete) => "hdel",
         HashExists(HashExists) => "hexists",

src/protocol/resp/src/storage/mod.rs

@@ -5,6 +5,7 @@
 use crate::*;
 
 pub trait Storage {
+    fn del(&mut self, request: &Del) -> Response;
     fn get(&mut self, request: &Get) -> Response;
     fn set(&mut self, request: &Set) -> Response;
 }