Repository: Migorithm/duva
Pull request #91 (2024-12-05 20:17:56)

Separate Query Handler Logic
- Renamed `InputValues` to `QueryArguments`
- Moved Parsing Logic (buffer to QueryIo) under `query_io.rs`
- Extracted `RequestHandler` from `QueryManager`
  - `QueryManager` takes reading and writing logic
  - `RequestHandler` takes request handling logic
- Created `ClusterRequest` and `ClusterRequestHandler`


Patch:
diff --git a/src/lib.rs b/src/lib.rs
index 84cda5ad..d8637c9b 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,9 +1,8 @@
 pub mod adapters;
-
 pub mod macros;
 pub mod services;
+use crate::services::query_manager::user_request_handler::UserRequestHandler;
 use anyhow::Result;
-
 use services::{
     config::{config_actor::Config, config_manager::ConfigManager},
     query_manager::{
@@ -15,6 +14,7 @@ use services::{
         persist::endec::TEnDecoder,
     },
 };
+use std::str::FromStr;
 
 /// dir, dbfilename is given as follows: ./your_program.sh --dir /tmp/redis-files --dbfilename dump.rdb
 
@@ -56,13 +56,13 @@ async fn start_accepting_connections<T: TCancellationTokenFactory>(
             Ok((stream, _)) =>
             // Spawn a new task to handle the connection without blocking the main thread.
             {
-                let query_manager = QueryManager::new(
-                    stream,
-                    config_manager.clone(),
-                    &cache_manager,
-                    ttl_inbox.clone(),
-                );
-                handle_single_user_stream::<T>(query_manager);
+                let query_manager = QueryManager::new(stream);
+                let request_handler = UserRequestHandler::new(
+                        config_manager.clone(),
+                        &cache_manager,
+                        ttl_inbox.clone(),
+                    );
+                handle_single_user_stream::<T>(query_manager, request_handler);
             }
             Err(e) => eprintln!("Failed to accept connection: {:?}", e),
         }
@@ -70,7 +70,8 @@ async fn start_accepting_connections<T: TCancellationTokenFactory>(
 }
 
 fn handle_single_user_stream<U: TCancellationTokenFactory>(
-    mut query_manager: QueryManager<impl TWrite + TRead, impl TEnDecoder>,
+    mut query_manager: QueryManager<impl TWrite + TRead>,
+    mut request_handler: UserRequestHandler<impl TEnDecoder>
 ) {
     tokio::spawn(async move {
         loop {
@@ -79,6 +80,8 @@ fn handle_single_user_stream<U: TCancellationTokenFactory>(
                 break;
             };
 
+            let user_request = FromStr::from_str(&cmd).unwrap();
+
             const TIMEOUT: u64 = 100;
             let (cancellation_notifier, cancellation_watcher) = U::create(TIMEOUT).split();
 
@@ -86,9 +89,16 @@ fn handle_single_user_stream<U: TCancellationTokenFactory>(
             // Notify the cancellation notifier to cancel the query after 100 milliseconds.
             cancellation_notifier.notify();
 
-            if let Err(e) = query_manager.handle(cancellation_watcher, cmd, args).await {
-                eprintln!("Error: {:?}", e);
-                break;
+            let result = request_handler.handle(cancellation_watcher, user_request, args).await;
+            // TODO: refactoring needed
+            match result {
+                Ok(response) => {
+                    query_manager.write_value(response).await.unwrap();
+                }
+                Err(e) => {
+                    eprintln!("Error: {:?}", e);
+                    break;
+                }
             }
         }
     });
diff --git a/src/services/query_manager/cluster_request.rs b/src/services/query_manager/cluster_request.rs
new file mode 100644
index 00000000..bb938880
--- /dev/null
+++ b/src/services/query_manager/cluster_request.rs
@@ -0,0 +1,15 @@
+use std::str::FromStr;
+pub enum ClusterRequest {
+    Ping,
+}
+
+impl FromStr for ClusterRequest {
+    type Err = anyhow::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_lowercase().as_str() {
+            "ping" => Ok(ClusterRequest::Ping),
+            _ => Err(anyhow::anyhow!("Invalid command")),
+        }
+    }
+}
diff --git a/src/services/query_manager/cluster_request_handler.rs b/src/services/query_manager/cluster_request_handler.rs
new file mode 100644
index 00000000..1d59dd9d
--- /dev/null
+++ b/src/services/query_manager/cluster_request_handler.rs
@@ -0,0 +1,55 @@
+use crate::services::config::config_manager::ConfigManager;
+use crate::services::query_manager::cluster_request::ClusterRequest;
+use crate::services::query_manager::interface::TCancellationWatcher;
+use crate::services::query_manager::query_arguments::QueryArguments;
+use crate::services::query_manager::query_io::QueryIO;
+use crate::services::statefuls::cache::cache_manager::CacheManager;
+use crate::services::statefuls::cache::ttl_manager::TtlSchedulerInbox;
+use crate::services::statefuls::persist::endec::TEnDecoder;
+
+pub struct ClusterHandler<U>
+where
+    U: TEnDecoder,
+{
+    // Maybe not all of these fields are needed
+    config_manager: ConfigManager,
+    cache_manager: &'static CacheManager<U>,
+    ttl_manager: TtlSchedulerInbox,
+}
+
+impl<U> ClusterHandler<U>
+where
+    U: TEnDecoder,
+{
+    pub(crate) fn new(
+        config_manager: ConfigManager,
+        cache_manager: &'static CacheManager<U>,
+        ttl_manager: TtlSchedulerInbox,
+    ) -> Self {
+        ClusterHandler {
+            config_manager,
+            cache_manager,
+            ttl_manager,
+        }
+    }
+
+    pub(crate) async fn handle(
+        &mut self,
+        mut cancellation_token: impl TCancellationWatcher,
+        cmd: ClusterRequest,
+        args: QueryArguments,
+    ) -> anyhow::Result<QueryIO> {
+        if cancellation_token.watch() {
+            let err = QueryIO::Err(
+                "Error operation cancelled due to timeout".to_string(),
+            );
+            return Ok(err);
+        }
+
+        // TODO if it is persistence operation, get the key and hash, take the appropriate sender, send it;
+        let response = match cmd {
+            ClusterRequest::Ping => QueryIO::SimpleString("PONG".to_string()),
+        };
+        Ok(response)
+    }
+}
\ No newline at end of file
diff --git a/src/services/query_manager/mod.rs b/src/services/query_manager/mod.rs
index c432dfc7..8d752e35 100644
--- a/src/services/query_manager/mod.rs
+++ b/src/services/query_manager/mod.rs
@@ -1,145 +1,44 @@
 pub mod interface;
 pub mod query_io;
-pub mod request;
-use crate::{
-    make_smart_pointer,
-    services::statefuls::cache::{cache_manager::CacheManager, ttl_manager::TtlSchedulerInbox},
-};
+pub mod user_request;
+mod query_arguments;
+pub(crate) mod user_request_handler;
+mod cluster_request_handler;
+mod cluster_request;
+
 use anyhow::Result;
 use bytes::BytesMut;
 use interface::{TRead, TWrite};
+use query_arguments::QueryArguments;
 use query_io::QueryIO;
-use request::UserRequest;
-
-use std::str::FromStr;
-
-use super::{
-    config::{config_manager::ConfigManager, ConfigResponse},
-    statefuls::{
-        cache::CacheEntry,
-        persist::{endec::TEnDecoder, save_actor::SaveActor},
-    },
-};
 
 /// Controller is a struct that will be used to read and write values to the client.
-pub struct QueryManager<T, U>
+pub struct QueryManager<T>
 where
     T: TWrite + TRead,
-    U: TEnDecoder,
 {
     pub(crate) stream: T,
-    config_manager: ConfigManager,
-    cache_manager: &'static CacheManager<U>,
-    ttl_manager: TtlSchedulerInbox,
-}
-
-impl<T, U> QueryManager<T, U>
-where
-    T: TWrite + TRead,
-    U: TEnDecoder,
-{
-    pub(crate) async fn handle(
-        &mut self,
-        mut cancellation_token: impl interface::TCancellationWatcher,
-        cmd: UserRequest,
-        args: InputValues,
-    ) -> Result<()> {
-        if cancellation_token.watch() {
-            self.write_value(QueryIO::Err(
-                "Error opertation cancelled due to timeout".to_string(),
-            ))
-            .await?;
-            return Ok(());
-        }
-
-        // TODO if it is persistence operation, get the key and hash, take the appropriate sender, send it;
-        let response = match cmd {
-            UserRequest::Ping => QueryIO::SimpleString("PONG".to_string()),
-            UserRequest::Echo => args.first().ok_or(anyhow::anyhow!("Not exists"))?.clone(),
-            UserRequest::Set => {
-                let cache_entry = args.take_set_args()?;
-                self.cache_manager
-                    .route_set(cache_entry, self.ttl_manager.clone())
-                    .await?
-            }
-            UserRequest::Save => {
-                // spawn save actor
-                let outbox = SaveActor::run(
-                    self.config_manager.get_filepath().await?,
-                    self.cache_manager.inboxes.len(),
-                    self.cache_manager.endecoder.clone(),
-                );
-
-                self.cache_manager.route_save(outbox).await;
-
-                QueryIO::Null
-            }
-            UserRequest::Get => {
-                let key = args.take_get_args()?;
-                self.cache_manager.route_get(key).await?
-            }
-            UserRequest::Keys => {
-                let pattern = args.take_keys_pattern()?;
-                self.cache_manager.route_keys(pattern).await?
-            }
-            // modify we have to add a new command
-            UserRequest::Config => {
-                let cmd = args.take_config_args()?;
-                let rx = self.config_manager.route_get(cmd).await?;
-
-                match rx.await? {
-                    ConfigResponse::Dir(value) => QueryIO::Array(vec![
-                        QueryIO::BulkString("dir".into()),
-                        QueryIO::BulkString(value),
-                    ]),
-                    ConfigResponse::DbFileName(value) => match value {
-                        Some(value) => QueryIO::BulkString(value),
-                        None => QueryIO::Null,
-                    },
-                    _ => QueryIO::Err("Invalid operation".into()),
-                }
-            }
-            UserRequest::Delete => panic!("Not implemented"),
-
-            UserRequest::Info => {
-                QueryIO::BulkString(self.config_manager.replication_info().await?.join("\r\n"))
-            }
-        };
-
-        self.write_value(response).await?;
-        Ok(())
-    }
 }
 
-impl<T, U> QueryManager<T, U>
+impl<T> QueryManager<T>
 where
     T: TWrite + TRead,
-    U: TEnDecoder,
 {
     pub(crate) fn new(
         stream: T,
-        config_manager: ConfigManager,
-        cache_manager: &'static CacheManager<U>,
-        ttl_manager: TtlSchedulerInbox,
     ) -> Self {
         QueryManager {
             stream,
-            config_manager,
-            cache_manager,
-            ttl_manager,
         }
     }
 
     // crlf
-    pub async fn read_value(&mut self) -> Result<Option<(UserRequest, InputValues)>> {
+    pub async fn read_value(&mut self) -> Result<Option<(String, QueryArguments)>> {
         let mut buffer = BytesMut::with_capacity(512);
         self.stream.read_bytes(&mut buffer).await?;
 
-        let (user_request, _) = parse(buffer)?;
-
-        let (str_cmd, values) = Self::extract_query(user_request)?;
-
-        Ok(Some((FromStr::from_str(&str_cmd)?, values)))
+        let (user_request, _) = query_io::parse(buffer)?;
+        Ok(Some(Self::extract_query(user_request)?))
     }
 
     pub async fn write_value(&mut self, value: QueryIO) -> Result<()> {
@@ -147,223 +46,14 @@ where
         Ok(())
     }
 
-    pub(crate) fn extract_query(value: QueryIO) -> Result<(String, InputValues)> {
+    fn extract_query(value: QueryIO) -> Result<(String, QueryArguments)> {
         match value {
             QueryIO::Array(value_array) => Ok((
                 value_array.first().unwrap().clone().unpack_bulk_str()?,
-                InputValues::new(value_array.into_iter().skip(1).collect()),
+                QueryArguments::new(value_array.into_iter().skip(1).collect()),
             )),
             _ => Err(anyhow::anyhow!("Unexpected command format")),
         }
     }
 }
 
-pub(super) fn parse(buffer: BytesMut) -> Result<(QueryIO, usize)> {
-    match buffer[0] as char {
-        '+' => parse_simple_string(buffer),
-        '*' => parse_array(buffer),
-        '$' => parse_bulk_string(buffer),
-        _ => Err(anyhow::anyhow!("Not a known value type {:?}", buffer)),
-    }
-}
-
-fn read_until_crlf(buffer: &[u8]) -> Option<(&[u8], usize)> {
-    for i in 1..buffer.len() {
-        if buffer[i - 1] == b'\r' && buffer[i] == b'\n' {
-            return Some((&buffer[0..(i - 1)], i + 1));
-        }
-    }
-    None
-}
-
-// +PING\r\n
-pub(crate) fn parse_simple_string(buffer: BytesMut) -> Result<(QueryIO, usize)> {
-    let (line, len) =
-        read_until_crlf(&buffer[1..]).ok_or(anyhow::anyhow!("Invalid simple string"))?;
-    let string = String::from_utf8(line.to_vec())?;
-    Ok((QueryIO::SimpleString(string), len + 1))
-}
-
-fn parse_array(buffer: BytesMut) -> Result<(QueryIO, usize)> {
-    let (line, mut len) =
-        read_until_crlf(&buffer[1..]).ok_or(anyhow::anyhow!("Invalid bulk string"))?;
-
-    len += 1;
-
-    let len_of_array = TryInto::<usize>::try_into(ConversionWrapper(line))?;
-    let mut bulk_strings = Vec::with_capacity(len_of_array);
-
-    for _ in 0..len_of_array {
-        let (value, l) = parse(BytesMut::from(&buffer[len..]))?;
-        bulk_strings.push(value);
-        len += l;
-    }
-
-    Ok((QueryIO::Array(bulk_strings), len))
-}
-
-fn parse_bulk_string(buffer: BytesMut) -> Result<(QueryIO, usize)> {
-    let (line, mut len) =
-        read_until_crlf(&buffer[1..]).ok_or(anyhow::anyhow!("Invalid bulk string"))?;
-
-    // Adjust `len` to include the initial line and calculate `bulk_str_len`
-    len += 1;
-    let bulk_str_len = usize::try_from(ConversionWrapper(line))?;
-
-    // Extract the bulk string from the buffer
-    let bulk_str = &buffer[len..len + bulk_str_len];
-    let bulk_string_value = String::from_utf8(bulk_str.to_vec())?;
-
-    // Return the bulk string value and adjusted length to account for CRLF
-    Ok((
-        QueryIO::BulkString(bulk_string_value),
-        len + bulk_str_len + 2,
-    ))
-}
-
-#[derive(Debug, Clone)]
-pub struct InputValues(Vec<QueryIO>);
-impl InputValues {
-    pub fn new(values: Vec<QueryIO>) -> Self {
-        Self(values)
-    }
-
-    pub(crate) fn take_get_args(&self) -> Result<String> {
-        let QueryIO::BulkString(key) = self.first().ok_or(anyhow::anyhow!("Not exists"))? else {
-            return Err(anyhow::anyhow!("Invalid arguments"));
-        };
-        Ok(key.to_string())
-    }
-    pub(crate) fn take_set_args(&self) -> Result<CacheEntry> {
-        let (QueryIO::BulkString(key), QueryIO::BulkString(value)) = (
-            self.first().ok_or(anyhow::anyhow!("Not exists"))?,
-            self.get(1).ok_or(anyhow::anyhow!("No value"))?,
-        ) else {
-            return Err(anyhow::anyhow!("Invalid arguments"));
-        };
-
-        //expire sig must be px or PX
-        match (self.0.get(2), self.0.get(3)) {
-            (Some(QueryIO::BulkString(sig)), Some(expiry)) => {
-                if sig.to_lowercase() != "px" {
-                    return Err(anyhow::anyhow!("Invalid arguments"));
-                }
-                Ok(CacheEntry::KeyValueExpiry(
-                    key.to_string(),
-                    value.to_string(),
-                    expiry.extract_expiry()?,
-                ))
-            }
-            (None, _) => Ok(CacheEntry::KeyValue(key.to_owned(), value.to_string())),
-            _ => Err(anyhow::anyhow!("Invalid arguments")),
-        }
-    }
-
-    fn take_config_args(&self) -> Result<(String, String)> {
-        let sub_command = self.first().ok_or(anyhow::anyhow!("Not exists"))?;
-        let args = &self[1..];
-
-        let (QueryIO::BulkString(command), [QueryIO::BulkString(key), ..]) = (&sub_command, args)
-        else {
-            return Err(anyhow::anyhow!("Invalid arguments"));
-        };
-        Ok((command.into(), key.into()))
-    }
-
-    // Pattern is passed with escape characters \" wrapping the pattern in question.
-    fn take_keys_pattern(&self) -> Result<Option<String>> {
-        let QueryIO::BulkString(pattern) = self.first().ok_or(anyhow::anyhow!("Not exists"))?
-        else {
-            return Err(anyhow::anyhow!("Invalid arguments"));
-        };
-
-        let pattern = pattern.trim_matches('\"');
-        match pattern {
-            "*" => Ok(None),
-            "" => Err(anyhow::anyhow!("Invalid pattern")),
-            pattern => Ok(Some(pattern.to_string())),
-        }
-    }
-}
-make_smart_pointer!(InputValues, Vec<QueryIO>);
-
-pub struct ConversionWrapper<T>(pub(crate) T);
-impl TryFrom<ConversionWrapper<&[u8]>> for usize {
-    type Error = anyhow::Error;
-
-    fn try_from(value: ConversionWrapper<&[u8]>) -> Result<Self> {
-        let string = String::from_utf8(value.0.to_vec())?;
-        Ok(string.parse()?)
-    }
-}
-
-#[test]
-fn test_parse_simple_string() {
-    // GIVEN
-    let buffer = BytesMut::from("+OK\r\n");
-
-    // WHEN
-    let (value, len) = parse_simple_string(buffer).unwrap();
-
-    // THEN
-    assert_eq!(len, 5);
-    assert_eq!(value, QueryIO::SimpleString("OK".to_string()));
-}
-
-#[test]
-fn test_parse_simple_string_ping() {
-    // GIVEN
-    let buffer = BytesMut::from("+PING\r\n");
-
-    // WHEN
-    let (value, len) = parse(buffer).unwrap();
-
-    // THEN
-    assert_eq!(len, 7);
-    assert_eq!(value, QueryIO::SimpleString("PING".to_string()));
-}
-
-#[test]
-fn test_parse_bulk_string() {
-    // GIVEN
-    let buffer = BytesMut::from("$5\r\nhello\r\n");
-
-    // WHEN
-    let (value, len) = parse(buffer).unwrap();
-
-    // THEN
-    assert_eq!(len, 11);
-    assert_eq!(value, QueryIO::BulkString("hello".to_string()));
-}
-
-#[test]
-fn test_parse_bulk_string_empty() {
-    // GIVEN
-    let buffer = BytesMut::from("$0\r\n\r\n");
-
-    // WHEN
-    let (value, len) = parse(buffer).unwrap();
-
-    // THEN
-    assert_eq!(len, 6);
-    assert_eq!(value, QueryIO::BulkString("".to_string()));
-}
-
-#[test]
-fn test_parse_array() {
-    // GIVEN
-    let buffer = BytesMut::from("*2\r\n$5\r\nhello\r\n$5\r\nworld\r\n");
-
-    // WHEN
-    let (value, len) = parse(buffer).unwrap();
-
-    // THEN
-    assert_eq!(len, 26);
-    assert_eq!(
-        value,
-        QueryIO::Array(vec![
-            QueryIO::BulkString("hello".to_string()),
-            QueryIO::BulkString("world".to_string()),
-        ])
-    );
-}
diff --git a/src/services/query_manager/query_arguments.rs b/src/services/query_manager/query_arguments.rs
new file mode 100644
index 00000000..5f0d10b1
--- /dev/null
+++ b/src/services/query_manager/query_arguments.rs
@@ -0,0 +1,72 @@
+use crate::make_smart_pointer;
+use crate::services::query_manager::query_io::QueryIO;
+use crate::services::statefuls::cache::CacheEntry;
+use anyhow::Result;
+
+#[derive(Debug, Clone)]
+pub struct QueryArguments(Vec<QueryIO>);
+
+make_smart_pointer!(QueryArguments, Vec<QueryIO>);
+
+impl QueryArguments {
+    pub fn new(values: Vec<QueryIO>) -> Self {
+        Self(values)
+    }
+
+    pub(crate) fn take_get_args(&self) -> anyhow::Result<String> {
+        let QueryIO::BulkString(key) = self.first().ok_or(anyhow::anyhow!("Not exists"))? else {
+            return Err(anyhow::anyhow!("Invalid arguments"));
+        };
+        Ok(key.to_string())
+    }
+    pub(crate) fn take_set_args(&self) -> anyhow::Result<CacheEntry> {
+        let (QueryIO::BulkString(key), QueryIO::BulkString(value)) = (
+            self.first().ok_or(anyhow::anyhow!("Not exists"))?,
+            self.get(1).ok_or(anyhow::anyhow!("No value"))?,
+        ) else {
+            return Err(anyhow::anyhow!("Invalid arguments"));
+        };
+
+        //expire sig must be px or PX
+        match (self.0.get(2), self.0.get(3)) {
+            (Some(QueryIO::BulkString(sig)), Some(expiry)) => {
+                if sig.to_lowercase() != "px" {
+                    return Err(anyhow::anyhow!("Invalid arguments"));
+                }
+                Ok(CacheEntry::KeyValueExpiry(
+                    key.to_string(),
+                    value.to_string(),
+                    expiry.extract_expiry()?,
+                ))
+            }
+            (None, _) => Ok(CacheEntry::KeyValue(key.to_owned(), value.to_string())),
+            _ => Err(anyhow::anyhow!("Invalid arguments")),
+        }
+    }
+
+    pub(crate) fn take_config_args(&self) -> Result<(String, String)> {
+        let sub_command = self.first().ok_or(anyhow::anyhow!("Not exists"))?;
+        let args = &self[1..];
+
+        let (QueryIO::BulkString(command), [QueryIO::BulkString(key), ..]) = (&sub_command, args)
+        else {
+            return Err(anyhow::anyhow!("Invalid arguments"));
+        };
+        Ok((command.into(), key.into()))
+    }
+
+    // Pattern is passed with escape characters \" wrapping the pattern in question.
+    pub(crate) fn take_keys_pattern(&self) -> anyhow::Result<Option<String>> {
+        let QueryIO::BulkString(pattern) = self.first().ok_or(anyhow::anyhow!("Not exists"))?
+        else {
+            return Err(anyhow::anyhow!("Invalid arguments"));
+        };
+
+        let pattern = pattern.trim_matches('\"');
+        match pattern {
+            "*" => Ok(None),
+            "" => Err(anyhow::anyhow!("Invalid pattern")),
+            pattern => Ok(Some(pattern.to_string())),
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/services/query_manager/query_io.rs b/src/services/query_manager/query_io.rs
index da76f509..9783e2ab 100644
--- a/src/services/query_manager/query_io.rs
+++ b/src/services/query_manager/query_io.rs
@@ -1,8 +1,8 @@
 use std::time::SystemTime;
 
-use anyhow::Result;
-
 use crate::services::statefuls::cache::CacheValue;
+use anyhow::Result;
+use bytes::BytesMut;
 
 #[derive(Clone, Debug, PartialEq)]
 pub enum QueryIO {
@@ -62,3 +62,148 @@ impl From<String> for QueryIO {
         QueryIO::BulkString(v)
     }
 }
+
+pub(in crate::services) fn parse(buffer: BytesMut) -> Result<(QueryIO, usize)> {
+    match buffer[0] as char {
+        '+' => parse_simple_string(buffer),
+        '*' => parse_array(buffer),
+        '$' => parse_bulk_string(buffer),
+        _ => Err(anyhow::anyhow!("Not a known value type {:?}", buffer)),
+    }
+}
+
+// +PING\r\n
+pub(crate) fn parse_simple_string(buffer: BytesMut) -> Result<(QueryIO, usize)> {
+    let (line, len) =
+        read_until_crlf(&buffer[1..]).ok_or(anyhow::anyhow!("Invalid simple string"))?;
+    let string = String::from_utf8(line.to_vec())?;
+    Ok((QueryIO::SimpleString(string), len + 1))
+}
+
+fn parse_array(buffer: BytesMut) -> Result<(QueryIO, usize)> {
+    let (line, mut len) =
+        read_until_crlf(&buffer[1..]).ok_or(anyhow::anyhow!("Invalid bulk string"))?;
+
+    len += 1;
+
+    let len_of_array = TryInto::<usize>::try_into(ConversionWrapper(line))?;
+    let mut bulk_strings = Vec::with_capacity(len_of_array);
+
+    for _ in 0..len_of_array {
+        let (value, l) = parse(BytesMut::from(&buffer[len..]))?;
+        bulk_strings.push(value);
+        len += l;
+    }
+
+    Ok((QueryIO::Array(bulk_strings), len))
+}
+
+fn parse_bulk_string(buffer: BytesMut) -> Result<(QueryIO, usize)> {
+    let (line, mut len) =
+        read_until_crlf(&buffer[1..]).ok_or(anyhow::anyhow!("Invalid bulk string"))?;
+
+    // Adjust `len` to include the initial line and calculate `bulk_str_len`
+    len += 1;
+    let bulk_str_len = usize::try_from(ConversionWrapper(line))?;
+
+    // Extract the bulk string from the buffer
+    let bulk_str = &buffer[len..len + bulk_str_len];
+    let bulk_string_value = String::from_utf8(bulk_str.to_vec())?;
+
+    // Return the bulk string value and adjusted length to account for CRLF
+    Ok((
+        QueryIO::BulkString(bulk_string_value),
+        len + bulk_str_len + 2,
+    ))
+}
+
+fn read_until_crlf(buffer: &[u8]) -> Option<(&[u8], usize)> {
+    for i in 1..buffer.len() {
+        if buffer[i - 1] == b'\r' && buffer[i] == b'\n' {
+            return Some((&buffer[0..(i - 1)], i + 1));
+        }
+    }
+    None
+}
+
+pub struct ConversionWrapper<T>(pub(crate) T);
+
+impl TryFrom<ConversionWrapper<&[u8]>> for usize {
+    type Error = anyhow::Error;
+
+    fn try_from(value: ConversionWrapper<&[u8]>) -> Result<Self> {
+        let string = String::from_utf8(value.0.to_vec())?;
+        Ok(string.parse()?)
+    }
+}
+
+#[test]
+fn test_parse_simple_string() {
+    // GIVEN
+    let buffer = BytesMut::from("+OK\r\n");
+
+    // WHEN
+    let (value, len) = parse_simple_string(buffer).unwrap();
+
+    // THEN
+    assert_eq!(len, 5);
+    assert_eq!(value, QueryIO::SimpleString("OK".to_string()));
+}
+
+#[test]
+fn test_parse_simple_string_ping() {
+    // GIVEN
+    let buffer = BytesMut::from("+PING\r\n");
+
+    // WHEN
+    let (value, len) = parse(buffer).unwrap();
+
+    // THEN
+    assert_eq!(len, 7);
+    assert_eq!(value, QueryIO::SimpleString("PING".to_string()));
+}
+
+#[test]
+fn test_parse_bulk_string() {
+    // GIVEN
+    let buffer = BytesMut::from("$5\r\nhello\r\n");
+
+    // WHEN
+    let (value, len) = parse(buffer).unwrap();
+
+    // THEN
+    assert_eq!(len, 11);
+    assert_eq!(value, QueryIO::BulkString("hello".to_string()));
+}
+
+#[test]
+fn test_parse_bulk_string_empty() {
+    // GIVEN
+    let buffer = BytesMut::from("$0\r\n\r\n");
+
+    // WHEN
+    let (value, len) = parse(buffer).unwrap();
+
+    // THEN
+    assert_eq!(len, 6);
+    assert_eq!(value, QueryIO::BulkString("".to_string()));
+}
+
+#[test]
+fn test_parse_array() {
+    // GIVEN
+    let buffer = BytesMut::from("*2\r\n$5\r\nhello\r\n$5\r\nworld\r\n");
+
+    // WHEN
+    let (value, len) = parse(buffer).unwrap();
+
+    // THEN
+    assert_eq!(len, 26);
+    assert_eq!(
+        value,
+        QueryIO::Array(vec![
+            QueryIO::BulkString("hello".to_string()),
+            QueryIO::BulkString("world".to_string()),
+        ])
+    );
+}
\ No newline at end of file
diff --git a/src/services/query_manager/request.rs b/src/services/query_manager/user_request.rs
similarity index 100%
rename from src/services/query_manager/request.rs
rename to src/services/query_manager/user_request.rs
diff --git a/src/services/query_manager/user_request_handler.rs b/src/services/query_manager/user_request_handler.rs
new file mode 100644
index 00000000..0155298a
--- /dev/null
+++ b/src/services/query_manager/user_request_handler.rs
@@ -0,0 +1,105 @@
+use crate::services::config::config_manager::ConfigManager;
+use crate::services::config::ConfigResponse;
+use crate::services::query_manager::interface::TCancellationWatcher;
+use crate::services::query_manager::query_arguments::QueryArguments;
+use crate::services::query_manager::query_io::QueryIO;
+use crate::services::query_manager::user_request::UserRequest;
+use crate::services::statefuls::cache::cache_manager::CacheManager;
+use crate::services::statefuls::cache::ttl_manager::TtlSchedulerInbox;
+use crate::services::statefuls::persist::endec::TEnDecoder;
+use crate::services::statefuls::persist::save_actor::SaveActor;
+
+pub struct UserRequestHandler<U>
+where
+    U: TEnDecoder,
+{
+    config_manager: ConfigManager,
+    cache_manager: &'static CacheManager<U>,
+    ttl_manager: TtlSchedulerInbox,
+}
+
+impl<U> UserRequestHandler<U>
+where
+    U: TEnDecoder,
+{
+    pub(crate) fn new(
+        config_manager: ConfigManager,
+        cache_manager: &'static CacheManager<U>,
+        ttl_manager: TtlSchedulerInbox,
+    ) -> Self {
+        UserRequestHandler {
+            config_manager,
+            cache_manager,
+            ttl_manager,
+        }
+    }
+
+    pub(crate) async fn handle(
+        &mut self,
+        mut cancellation_token: impl TCancellationWatcher,
+        cmd: UserRequest,
+        args: QueryArguments,
+    ) -> anyhow::Result<QueryIO> {
+        if cancellation_token.watch() {
+            let err = QueryIO::Err(
+                "Error operation cancelled due to timeout".to_string(),
+            );
+            return Ok(err);
+        }
+
+        // TODO if it is persistence operation, get the key and hash, take the appropriate sender, send it;
+        let response = match cmd {
+            UserRequest::Ping => QueryIO::SimpleString("PONG".to_string()),
+            UserRequest::Echo => args.first().ok_or(anyhow::anyhow!("Not exists"))?.clone(),
+            UserRequest::Set => {
+                let cache_entry = args.take_set_args()?;
+                self.cache_manager
+                    .route_set(cache_entry, self.ttl_manager.clone())
+                    .await?
+            }
+            UserRequest::Save => {
+                // spawn save actor
+                let outbox = SaveActor::run(
+                    self.config_manager.get_filepath().await?,
+                    self.cache_manager.inboxes.len(),
+                    self.cache_manager.endecoder.clone(),
+                );
+
+                self.cache_manager.route_save(outbox).await;
+
+                QueryIO::Null
+            }
+            UserRequest::Get => {
+                let key = args.take_get_args()?;
+                self.cache_manager.route_get(key).await?
+            }
+            UserRequest::Keys => {
+                let pattern = args.take_keys_pattern()?;
+                self.cache_manager.route_keys(pattern).await?
+            }
+            // modify we have to add a new command
+            UserRequest::Config => {
+                let cmd = args.take_config_args()?;
+                let rx = self.config_manager.route_get(cmd).await?;
+
+                match rx.await? {
+                    ConfigResponse::Dir(value) => QueryIO::Array(vec![
+                        QueryIO::BulkString("dir".into()),
+                        QueryIO::BulkString(value),
+                    ]),
+                    ConfigResponse::DbFileName(value) => match value {
+                        Some(value) => QueryIO::BulkString(value),
+                        None => QueryIO::Null,
+                    },
+                    _ => QueryIO::Err("Invalid operation".into()),
+                }
+            }
+            UserRequest::Delete => panic!("Not implemented"),
+
+            UserRequest::Info => {
+                QueryIO::BulkString(self.config_manager.replication_info().await?.join("\r\n"))
+            }
+        };
+        Ok(response)
+    }
+}
\ No newline at end of file


Test patch:
diff --git a/tests/test_cancellation_token.rs b/tests/test_cancellation_token.rs
index 3aac581c..43d65e42 100644
--- a/tests/test_cancellation_token.rs
+++ b/tests/test_cancellation_token.rs
@@ -26,7 +26,7 @@ async fn test_cancellation_token() {
     // THEN
     assert_eq!(
         h.get_response().await,
-        "-Error opertation cancelled due to timeout\r\n"
+        "-Error operation cancelled due to timeout\r\n"
     );
 }