Repository: Migorithm/duva
Pull request #72 (2024-11-30 20:25:17)

refactor: remove endecoder dependency from query manager


Patch:
diff --git a/src/lib.rs b/src/lib.rs
index dd24fc7e..3506acd8 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -24,7 +24,7 @@ pub async fn start_up(
     endec: impl TDecodeData + TEncodeData,
 ) -> Result<()> {
     let (cache_dispatcher, ttl_inbox) =
-        CacheManager::run_cache_actors(number_of_cache_actors, endec.clone());
+        CacheManager::run_cache_actors(number_of_cache_actors, endec);
 
     // Load data from the file if --dir and --dbfilename are provided
     cache_dispatcher
@@ -37,7 +37,7 @@ pub async fn start_up(
             Ok((socket, _)) =>
             // Spawn a new task to handle the connection without blocking the main thread.
             {
-                let query_manager = QueryManager::new(socket, config, endec.clone());
+                let query_manager = QueryManager::new(socket, config);
                 process_socket(query_manager, ttl_inbox.clone(), cache_dispatcher.clone())
             }
             Err(e) => eprintln!("Failed to accept connection: {:?}", e),
@@ -46,7 +46,7 @@ pub async fn start_up(
 }
 
 fn process_socket<T: TDecodeData + TEncodeData>(
-    mut query_manager: QueryManager<TcpStream, T>,
+    mut query_manager: QueryManager<TcpStream>,
     ttl_inbox: TtlSchedulerInbox,
     cache_dispatcher: CacheManager<T>,
 ) {
diff --git a/src/services/query_manager/mod.rs b/src/services/query_manager/mod.rs
index cb1c43ff..90da3f90 100644
--- a/src/services/query_manager/mod.rs
+++ b/src/services/query_manager/mod.rs
@@ -20,23 +20,20 @@ use std::str::FromStr;
 use super::interfaces::endec::{TDecodeData, TEncodeData};
 
 /// Controller is a struct that will be used to read and write values to the client.
-pub struct QueryManager<T, EnDec>
+pub struct QueryManager<T>
 where
     T: TWriteBuf + TRead,
-    EnDec: TDecodeData + TEncodeData,
 {
     pub(crate) stream: T,
     buffer: BytesMut,
     config: &'static Config,
-    endec: EnDec,
 }
 
-impl<T, EnDec> QueryManager<T, EnDec>
+impl<T> QueryManager<T>
 where
     T: TWriteBuf + TRead,
-    EnDec: TDecodeData + TEncodeData,
 {
-    pub(crate) async fn handle(
+    pub(crate) async fn handle<EnDec: TEncodeData + TDecodeData>(
         &mut self,
         persistence_router: &CacheManager<EnDec>,
         ttl_sender: TtlSchedulerInbox,
@@ -57,7 +54,7 @@ where
             }
             UserRequest::Save => {
                 // spawn save actor
-                persistence_router.run_save_actor(self.config.get_filepath(), self.endec.clone());
+                persistence_router.run_save_actor(self.config.get_filepath());
                 // TODO Set return type
                 QueryIO::Null
             }
@@ -90,17 +87,15 @@ where
     }
 }
 
-impl<T, EnDec> QueryManager<T, EnDec>
+impl<T> QueryManager<T>
 where
     T: TWriteBuf + TRead,
-    EnDec: TDecodeData + TEncodeData,
 {
-    pub fn new(stream: T, config: &'static Config, endec: EnDec) -> Self {
+    pub fn new(stream: T, config: &'static Config) -> Self {
         QueryManager {
             stream,
             buffer: BytesMut::with_capacity(512),
             config,
-            endec,
         }
     }
 
diff --git a/src/services/statefuls/routers/cache_manager.rs b/src/services/statefuls/routers/cache_manager.rs
index 05972aa9..9d829c8b 100644
--- a/src/services/statefuls/routers/cache_manager.rs
+++ b/src/services/statefuls/routers/cache_manager.rs
@@ -16,9 +16,9 @@ type OneShotReceiverJoinHandle<T> =
     tokio::task::JoinHandle<std::result::Result<T, tokio::sync::oneshot::error::RecvError>>;
 
 #[derive(Clone)]
-pub(crate) struct CacheManager<Dec> {
+pub(crate) struct CacheManager<EnDec> {
     pub(crate) inboxes: Vec<CacheCommandSender>,
-    pub(crate) decoder: Dec,
+    pub(crate) endecoder: EnDec,
 }
 
 impl<EnDec: TDecodeData + TEncodeData> CacheManager<EnDec> {
@@ -34,7 +34,7 @@ impl<EnDec: TDecodeData + TEncodeData> CacheManager<EnDec> {
 
         let data = tokio::fs::read(filepath).await?;
 
-        let database = self.decoder.decode_data(data)?;
+        let database = self.endecoder.decode_data(data)?;
 
         for kvs in database
             .key_values()
@@ -146,7 +146,7 @@ impl<EnDec: TDecodeData + TEncodeData> CacheManager<EnDec> {
             inboxes: (0..num_of_actors)
                 .map(|_| CacheActor::run())
                 .collect::<Vec<_>>(),
-            decoder,
+            endecoder: decoder,
         };
 
         let ttl_inbox = cache_dispatcher.run_ttl_actors();
@@ -157,9 +157,9 @@ impl<EnDec: TDecodeData + TEncodeData> CacheManager<EnDec> {
         TtlActor::run(self.clone())
     }
 
-    pub fn run_save_actor(&self, db_filepath: Option<String>, encoder: EnDec) {
+    pub fn run_save_actor(&self, db_filepath: Option<String>) {
         let filepath: String = db_filepath.unwrap_or_else(|| "dump.rdb".to_string());
-        let outbox = SaveActor::run(filepath, self.inboxes.len(), encoder);
+        let outbox = SaveActor::run(filepath, self.inboxes.len(), self.endecoder.clone());
 
         // get all the handlers to cache actors
         for inbox in self.inboxes.iter().map(Clone::clone) {


Test patch:
diff --git a/src/test.rs b/src/test.rs
index bc288c57..734aeb6d 100644
--- a/src/test.rs
+++ b/src/test.rs
@@ -72,7 +72,7 @@ async fn test_set() {
             .as_bytes()
             .to_vec(),
     };
-    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
+    let mut controller = QueryManager::new(stream, test_config());
 
     // WHEN
     controller
@@ -100,7 +100,7 @@ async fn test_set_with_expiry() {
 
     let (cache_dispatcher, ttl_inbox) = CacheManager::run_cache_actors(3, EnDecoder);
 
-    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
+    let mut controller = QueryManager::new(stream, test_config());
 
     // WHEN
     controller
@@ -130,7 +130,7 @@ async fn test_set_with_expire_should_expire_within_100ms() {
     };
     let (cache_dispatcher, ttl_inbox) = CacheManager::run_cache_actors(3, EnDecoder);
 
-    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
+    let mut controller = QueryManager::new(stream, test_config());
 
     // WHEN
     controller
@@ -165,7 +165,7 @@ async fn test_config_get_dir() {
     };
     let (cache_dispatcher, ttl_inbox) = CacheManager::run_cache_actors(3, EnDecoder);
 
-    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
+    let mut controller = QueryManager::new(stream, test_config());
 
     // WHEN
     controller
@@ -193,7 +193,7 @@ async fn test_keys() {
         written: "*2\r\n$4\r\nKEYS\r\n$3\r\n\"*\"\r\n".as_bytes().to_vec(),
     };
 
-    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
+    let mut controller = QueryManager::new(stream, test_config());
 
     // WHEN
     controller
@@ -225,7 +225,7 @@ async fn test_replication_info() {
             .as_bytes()
             .to_vec(),
     };
-    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
+    let mut controller = QueryManager::new(stream, test_config());
     // WHEN
     controller
         .handle(&cache_dispatcher, ttl_inbox)