Repository: Migorithm/duva
Pull request #66 (2024-11-28 06:30:11)

feat: encoder DI applied


Patch:
diff --git a/src/adapters/persistence/encoder/mod.rs b/src/adapters/persistence/encoder/mod.rs
index 4e8f0c86..5c53ea23 100644
--- a/src/adapters/persistence/encoder/mod.rs
+++ b/src/adapters/persistence/encoder/mod.rs
@@ -1,5 +1,5 @@
 use crate::services::{
-    interfaces::endec::{Processable, TEncodeData},
+    interfaces::endec::Processable,
     statefuls::routers::{cache_actor::CacheChunk, save_actor::SaveActorCommand},
 };
 use byte_encoder::{
@@ -11,16 +11,31 @@ use tokio::io::AsyncWriteExt;
 
 pub mod byte_encoder;
 
-pub struct FileProcessor {
-    file: tokio::fs::File,
+pub struct EncodingProcessor {
+    pub(super) file: tokio::fs::File,
+    pub(super) encoding_meta: EncodingMeta,
+}
+
+pub(super) struct EncodingMeta {
     num_of_saved_table_size_actor: usize,
     total_key_value_table_size: usize,
     total_expires_table_size: usize,
     chunk_queue: VecDeque<CacheChunk>,
     num_of_cache_actors: usize,
 }
+impl EncodingMeta {
+    pub(super) fn new(num_of_cache_actors: usize) -> Self {
+        Self {
+            num_of_saved_table_size_actor: num_of_cache_actors,
+            total_key_value_table_size: 0,
+            total_expires_table_size: 0,
+            chunk_queue: VecDeque::new(),
+            num_of_cache_actors,
+        }
+    }
+}
 
-impl Processable for FileProcessor {
+impl Processable for EncodingProcessor {
     async fn add_meta(&mut self) -> anyhow::Result<()> {
         let meta = [
             encode_header("0011")?,
@@ -34,25 +49,25 @@ impl Processable for FileProcessor {
     async fn handle_cmd(&mut self, cmd: SaveActorCommand) -> anyhow::Result<bool> {
         match cmd {
             SaveActorCommand::SaveTableSize(key_value_table_size, expires_table_size) => {
-                self.num_of_saved_table_size_actor -= 1;
-                if self.num_of_saved_table_size_actor == 0 {
+                self.encoding_meta.num_of_saved_table_size_actor -= 1;
+                if self.encoding_meta.num_of_saved_table_size_actor == 0 {
                     self.file
                         .write_all(&encode_database_table_size(
-                            self.total_key_value_table_size,
-                            self.total_expires_table_size,
+                            self.encoding_meta.total_key_value_table_size,
+                            self.encoding_meta.total_expires_table_size,
                         )?)
                         .await?;
                 } else {
-                    self.total_key_value_table_size += key_value_table_size;
-                    self.total_expires_table_size += expires_table_size;
+                    self.encoding_meta.total_key_value_table_size += key_value_table_size;
+                    self.encoding_meta.total_expires_table_size += expires_table_size;
                 }
             }
             SaveActorCommand::SaveChunk(chunk) => {
-                if self.num_of_saved_table_size_actor != 0 {
-                    self.chunk_queue.push_back(chunk);
+                if self.encoding_meta.num_of_saved_table_size_actor != 0 {
+                    self.encoding_meta.chunk_queue.push_back(chunk);
                 } else {
-                    self.chunk_queue.push_back(chunk);
-                    while let Some(chunk) = self.chunk_queue.pop_front() {
+                    self.encoding_meta.chunk_queue.push_back(chunk);
+                    while let Some(chunk) = self.encoding_meta.chunk_queue.pop_front() {
                         let chunk = chunk.0;
                         for (key, value) in chunk {
                             let encoded_chunk = value.encode_with_key(&key)?;
@@ -62,9 +77,9 @@ impl Processable for FileProcessor {
                 }
             }
             SaveActorCommand::StopSentinel => {
-                self.num_of_cache_actors -= 1;
-                if self.num_of_cache_actors == 0 {
-                    while let Some(chunk) = self.chunk_queue.pop_front() {
+                self.encoding_meta.num_of_cache_actors -= 1;
+                if self.encoding_meta.num_of_cache_actors == 0 {
+                    while let Some(chunk) = self.encoding_meta.chunk_queue.pop_front() {
                         let chunk = chunk.0;
                         for (key, value) in chunk {
                             let encoded_chunk = value.encode_with_key(&key)?;
@@ -80,24 +95,3 @@ impl Processable for FileProcessor {
         Ok(false)
     }
 }
-
-impl TEncodeData for tokio::fs::OpenOptions {
-    async fn create_on_path(
-        filepath: &str,
-        num_of_cache_actors: usize,
-    ) -> anyhow::Result<FileProcessor> {
-        let file = tokio::fs::OpenOptions::new()
-            .write(true)
-            .create(true)
-            .open(filepath)
-            .await?;
-        Ok(FileProcessor {
-            file,
-            num_of_saved_table_size_actor: Default::default(),
-            total_key_value_table_size: Default::default(),
-            total_expires_table_size: Default::default(),
-            chunk_queue: Default::default(),
-            num_of_cache_actors,
-        })
-    }
-}
diff --git a/src/adapters/persistence/mod.rs b/src/adapters/persistence/mod.rs
index 5db97e76..c1f33a8c 100644
--- a/src/adapters/persistence/mod.rs
+++ b/src/adapters/persistence/mod.rs
@@ -64,11 +64,10 @@
 /// Each key-value pair is stored as follows:
 ///
 /// 1. Optional Expire Information:
-/// **Timestamp in Seconds:**
-/// FD - Expire timestamp in seconds (4-byte unsigned integer)
-///
-/// **Timestamp in Milliseconds:**
-/// FC - Expire timestamp in milliseconds (8-byte unsigned long)
+///     **Timestamp in Seconds:**
+///     FD - Expire timestamp in seconds (4-byte unsigned integer)
+///     **Timestamp in Milliseconds:**
+///     FC - Expire timestamp in milliseconds (8-byte unsigned long)
 ///
 /// 2. **Value Type:** 1-byte flag indicating the type and encoding of the value.
 /// 3. **Key:** String encoded.
@@ -81,8 +80,9 @@ pub mod encoder;
 use decoder::byte_decoder::BytesDecoder;
 use decoder::states::DecoderInit;
 pub use encoder::byte_encoder;
+use encoder::{EncodingMeta, EncodingProcessor};
 
-use crate::services::interfaces::endec::TDecodeData;
+use crate::services::interfaces::endec::{TDecodeData, TEncodeData};
 
 const HEADER_MAGIC_STRING: &str = "REDIS";
 const METADATA_SECTION_INDICATOR: u8 = 0xFA;
@@ -130,3 +130,22 @@ impl TDecodeData for EnDecoder {
         Ok(database)
     }
 }
+
+#[allow(refining_impl_trait)]
+impl TEncodeData for EnDecoder {
+    async fn create_on_path(
+        &self,
+        filepath: &str,
+        num_of_cache_actors: usize,
+    ) -> anyhow::Result<EncodingProcessor> {
+        let file = tokio::fs::OpenOptions::new()
+            .write(true)
+            .create(true)
+            .open(filepath)
+            .await?;
+        Ok(EncodingProcessor {
+            file,
+            encoding_meta: EncodingMeta::new(num_of_cache_actors),
+        })
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index d9d668e4..dd24fc7e 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -7,7 +7,7 @@ pub mod services;
 use anyhow::Result;
 use config::Config;
 use services::{
-    interfaces::endec::TDecodeData,
+    interfaces::endec::{TDecodeData, TEncodeData},
     query_manager::QueryManager,
     statefuls::routers::{cache_manager::CacheManager, ttl_manager::TtlSchedulerInbox},
 };
@@ -21,10 +21,10 @@ mod test;
 pub async fn start_up(
     config: &'static Config,
     number_of_cache_actors: usize,
-    endec: impl TDecodeData,
+    endec: impl TDecodeData + TEncodeData,
 ) -> Result<()> {
     let (cache_dispatcher, ttl_inbox) =
-        CacheManager::run_cache_actors(number_of_cache_actors, endec);
+        CacheManager::run_cache_actors(number_of_cache_actors, endec.clone());
 
     // 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);
+                let query_manager = QueryManager::new(socket, config, endec.clone());
                 process_socket(query_manager, ttl_inbox.clone(), cache_dispatcher.clone())
             }
             Err(e) => eprintln!("Failed to accept connection: {:?}", e),
@@ -45,10 +45,10 @@ pub async fn start_up(
     }
 }
 
-fn process_socket(
-    mut query_manager: QueryManager<TcpStream>,
+fn process_socket<T: TDecodeData + TEncodeData>(
+    mut query_manager: QueryManager<TcpStream, T>,
     ttl_inbox: TtlSchedulerInbox,
-    cache_dispatcher: CacheManager<impl TDecodeData>,
+    cache_dispatcher: CacheManager<T>,
 ) {
     tokio::spawn(async move {
         loop {
diff --git a/src/services/interfaces/endec.rs b/src/services/interfaces/endec.rs
index 4af6979d..6abd3ba6 100644
--- a/src/services/interfaces/endec.rs
+++ b/src/services/interfaces/endec.rs
@@ -11,31 +11,37 @@ pub trait TDecodeData: ThreadSafeCloneable {
 }
 
 pub trait TEncodeData: ThreadSafeCloneable {
-    // template method
-    async fn encode_data(
+    /// ** Template method pattern
+    ///    `create_on_path` and its return will determine where the file will be saved
+    fn encode_data(
         &self,
         filepath: &str,
         inbox: &mut Receiver<SaveActorCommand>,
         number_of_cache_actors: usize,
-    ) -> anyhow::Result<()> {
-        let mut processor = Self::create_on_path(filepath, number_of_cache_actors).await?;
-        processor.add_meta().await?;
-        while let Some(command) = inbox.recv().await {
-            match processor.handle_cmd(command).await {
-                Ok(should_break) => {
-                    if should_break {
-                        break;
+    ) -> impl std::future::Future<Output = anyhow::Result<()>> + Send {
+        async move {
+            let mut processor = self
+                .create_on_path(filepath, number_of_cache_actors)
+                .await?;
+            processor.add_meta().await?;
+            while let Some(command) = inbox.recv().await {
+                match processor.handle_cmd(command).await {
+                    Ok(should_break) => {
+                        if should_break {
+                            break;
+                        }
+                    }
+                    Err(err) => {
+                        eprintln!("error while encoding: {:?}", err);
+                        return Err(err);
                     }
-                }
-                Err(err) => {
-                    eprintln!("error while encoding: {:?}", err);
-                    return Err(err);
                 }
             }
+            Ok(())
         }
-        Ok(())
     }
     fn create_on_path(
+        &self,
         filepath: &str,
         number_of_cache_actors: usize,
     ) -> impl std::future::Future<Output = anyhow::Result<impl Processable>> + Send;
diff --git a/src/services/query_manager/mod.rs b/src/services/query_manager/mod.rs
index 6ff2ab0a..cb1c43ff 100644
--- a/src/services/query_manager/mod.rs
+++ b/src/services/query_manager/mod.rs
@@ -17,19 +17,28 @@ use request::UserRequest;
 
 use std::str::FromStr;
 
-use super::interfaces::endec::TDecodeData;
+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: TWriteBuf + TRead> {
+pub struct QueryManager<T, EnDec>
+where
+    T: TWriteBuf + TRead,
+    EnDec: TDecodeData + TEncodeData,
+{
     pub(crate) stream: T,
     buffer: BytesMut,
     config: &'static Config,
+    endec: EnDec,
 }
 
-impl<U: TWriteBuf + TRead> QueryManager<U> {
-    pub(crate) async fn handle<D: TDecodeData>(
+impl<T, EnDec> QueryManager<T, EnDec>
+where
+    T: TWriteBuf + TRead,
+    EnDec: TDecodeData + TEncodeData,
+{
+    pub(crate) async fn handle(
         &mut self,
-        persistence_router: &CacheManager<D>,
+        persistence_router: &CacheManager<EnDec>,
         ttl_sender: TtlSchedulerInbox,
     ) -> Result<()> {
         let Some((cmd, args)) = self.read_value().await? else {
@@ -48,7 +57,7 @@ impl<U: TWriteBuf + TRead> QueryManager<U> {
             }
             UserRequest::Save => {
                 // spawn save actor
-                persistence_router.run_save_actor(self.config.get_filepath());
+                persistence_router.run_save_actor(self.config.get_filepath(), self.endec.clone());
                 // TODO Set return type
                 QueryIO::Null
             }
@@ -81,12 +90,17 @@ impl<U: TWriteBuf + TRead> QueryManager<U> {
     }
 }
 
-impl<T: TWriteBuf + TRead> QueryManager<T> {
-    pub fn new(stream: T, config: &'static Config) -> Self {
+impl<T, EnDec> QueryManager<T, EnDec>
+where
+    T: TWriteBuf + TRead,
+    EnDec: TDecodeData + TEncodeData,
+{
+    pub fn new(stream: T, config: &'static Config, endec: EnDec) -> 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 b15b1f13..05972aa9 100644
--- a/src/services/statefuls/routers/cache_manager.rs
+++ b/src/services/statefuls/routers/cache_manager.rs
@@ -4,7 +4,7 @@ use super::save_actor::SaveActor;
 use super::ttl_manager::{TtlActor, TtlSchedulerInbox};
 
 use crate::config::Config;
-use crate::services::interfaces::endec::TDecodeData;
+use crate::services::interfaces::endec::{TDecodeData, TEncodeData};
 use crate::services::query_manager::query_io::QueryIO;
 use crate::services::CacheEntry;
 use anyhow::Result;
@@ -21,7 +21,7 @@ pub(crate) struct CacheManager<Dec> {
     pub(crate) decoder: Dec,
 }
 
-impl<Dec: TDecodeData> CacheManager<Dec> {
+impl<EnDec: TDecodeData + TEncodeData> CacheManager<EnDec> {
     pub(crate) async fn load_data(
         &self,
         ttl_inbox: TtlSchedulerInbox,
@@ -140,8 +140,8 @@ impl<Dec: TDecodeData> CacheManager<Dec> {
 
     pub fn run_cache_actors(
         num_of_actors: usize,
-        decoder: Dec,
-    ) -> (CacheManager<Dec>, TtlSchedulerInbox) {
+        decoder: EnDec,
+    ) -> (CacheManager<EnDec>, TtlSchedulerInbox) {
         let cache_dispatcher = CacheManager {
             inboxes: (0..num_of_actors)
                 .map(|_| CacheActor::run())
@@ -157,9 +157,9 @@ impl<Dec: TDecodeData> CacheManager<Dec> {
         TtlActor::run(self.clone())
     }
 
-    pub fn run_save_actor(&self, db_filepath: Option<String>) {
+    pub fn run_save_actor(&self, db_filepath: Option<String>, encoder: EnDec) {
         let filepath: String = db_filepath.unwrap_or_else(|| "dump.rdb".to_string());
-        let outbox = SaveActor::run(filepath, self.inboxes.len());
+        let outbox = SaveActor::run(filepath, self.inboxes.len(), encoder);
 
         // get all the handlers to cache actors
         for inbox in self.inboxes.iter().map(Clone::clone) {
diff --git a/src/services/statefuls/routers/save_actor.rs b/src/services/statefuls/routers/save_actor.rs
index 03effce7..5506264b 100644
--- a/src/services/statefuls/routers/save_actor.rs
+++ b/src/services/statefuls/routers/save_actor.rs
@@ -1,11 +1,5 @@
 use super::cache_actor::CacheChunk;
-use crate::adapters::persistence::byte_encoder::{
-    encode_checksum, encode_database_info, encode_database_table_size, encode_header,
-    encode_metadata,
-};
-use std::collections::VecDeque;
-
-use tokio::io::AsyncWriteExt;
+use crate::services::interfaces::endec::TEncodeData;
 use tokio::sync::mpsc::Sender;
 
 pub enum SaveActorCommand {
@@ -14,95 +8,35 @@ pub enum SaveActorCommand {
     StopSentinel,
 }
 
-pub struct SaveActor {
+pub struct SaveActor<T: TEncodeData> {
     filepath: String,
     pub num_of_cache_actors: usize,
     pub inbox: tokio::sync::mpsc::Receiver<SaveActorCommand>,
+    encoder: T,
 }
 
-impl SaveActor {
-    pub fn run(filepath: String, num_of_cache_actors: usize) -> Sender<SaveActorCommand> {
+impl<T: TEncodeData> SaveActor<T> {
+    pub fn run(
+        filepath: String,
+        num_of_cache_actors: usize,
+        encoder: T,
+    ) -> Sender<SaveActorCommand> {
         let (outbox, inbox) = tokio::sync::mpsc::channel(100);
         let actor = Self {
             filepath,
             inbox,
             num_of_cache_actors,
+            encoder,
         };
         tokio::spawn(actor.handle());
         outbox
     }
 
     pub async fn handle(mut self) -> anyhow::Result<()> {
-        // TODO -  eunchan
-        // USE `TEncodeData` by injecting dependency - eunchan
-        // Fire TEncodeData::encode_data(&instance,filepath, &mut self.inbox,self.number_of_actors).await?;
-
-        let mut file = tokio::fs::OpenOptions::new()
-            .write(true)
-            .create(true)
-            .open(&self.filepath)
+        self.encoder
+            .encode_data(&self.filepath, &mut self.inbox, self.num_of_cache_actors)
             .await?;
 
-        // TODO move to adapter logic as it pertains to IO
-        let meta = [
-            encode_header("0011")?,
-            encode_metadata(Vec::from([("redis-ver", "6.0.16")]))?,
-            encode_database_info(0)?,
-        ];
-
-        file.write_all(&meta.concat()).await?;
-
-        let mut num_of_saved_table_size_actor = self.num_of_cache_actors;
-        let mut total_key_value_table_size = 0;
-        let mut total_expires_table_size = 0;
-        let mut chunk_queue = VecDeque::new();
-
-        while let Some(command) = self.inbox.recv().await {
-            match command {
-                SaveActorCommand::SaveTableSize(key_value_table_size, expires_table_size) => {
-                    num_of_saved_table_size_actor -= 1;
-                    if num_of_saved_table_size_actor == 0 {
-                        file.write_all(&encode_database_table_size(
-                            total_key_value_table_size,
-                            total_expires_table_size,
-                        )?)
-                        .await?;
-                    } else {
-                        total_key_value_table_size += key_value_table_size;
-                        total_expires_table_size += expires_table_size;
-                    }
-                }
-                SaveActorCommand::SaveChunk(chunk) => {
-                    if num_of_saved_table_size_actor != 0 {
-                        chunk_queue.push_back(chunk);
-                    } else {
-                        chunk_queue.push_back(chunk);
-                        while let Some(chunk) = chunk_queue.pop_front() {
-                            let chunk = chunk.0;
-                            for (key, value) in chunk {
-                                let encoded_chunk = value.encode_with_key(&key)?;
-                                file.write_all(&encoded_chunk).await?;
-                            }
-                        }
-                    }
-                }
-                SaveActorCommand::StopSentinel => {
-                    self.num_of_cache_actors -= 1;
-                    if self.num_of_cache_actors == 0 {
-                        while let Some(chunk) = chunk_queue.pop_front() {
-                            let chunk = chunk.0;
-                            for (key, value) in chunk {
-                                let encoded_chunk = value.encode_with_key(&key)?;
-                                file.write_all(&encoded_chunk).await?;
-                            }
-                        }
-                        let checksum = encode_checksum(&[0; 8])?;
-                        file.write_all(&checksum).await?;
-                        break;
-                    }
-                }
-            }
-        }
         Ok(())
     }
 }
diff --git a/src/services/statefuls/routers/ttl_manager.rs b/src/services/statefuls/routers/ttl_manager.rs
index 5b580134..fd308f63 100644
--- a/src/services/statefuls/routers/ttl_manager.rs
+++ b/src/services/statefuls/routers/ttl_manager.rs
@@ -1,7 +1,7 @@
 use crate::{
     make_smart_pointer,
     services::{
-        interfaces::endec::TDecodeData,
+        interfaces::endec::{TDecodeData, TEncodeData},
         statefuls::routers::{cache_actor::CacheCommand, cache_manager::CacheManager},
     },
 };
@@ -21,7 +21,9 @@ use tokio::{
 pub struct TtlActor;
 
 impl TtlActor {
-    pub(crate) fn run<T: TDecodeData>(cache_dispatcher: CacheManager<T>) -> TtlSchedulerInbox {
+    pub(crate) fn run<T: TDecodeData + TEncodeData>(
+        cache_dispatcher: CacheManager<T>,
+    ) -> TtlSchedulerInbox {
         let (scheduler_outbox, inbox) = tokio::sync::mpsc::channel(100);
         tokio::spawn(Self::ttl_schedule_actor(inbox));
         tokio::spawn(Self::background_delete_actor(
@@ -33,7 +35,7 @@ impl TtlActor {
     }
 
     // Background actor keeps sending peek command to the scheduler actor to check if there is any key to delete.
-    async fn background_delete_actor<T: TDecodeData>(
+    async fn background_delete_actor<T: TDecodeData + TEncodeData>(
         cache_manager: CacheManager<T>,
         outbox: mpsc::Sender<TtlCommand>,
     ) -> anyhow::Result<()> {


Test patch:
diff --git a/src/test.rs b/src/test.rs
index 734aeb6d..bc288c57 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());
+    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
 
     // 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());
+    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
 
     // 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());
+    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
 
     // 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());
+    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
 
     // 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());
+    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
 
     // WHEN
     controller
@@ -225,7 +225,7 @@ async fn test_replication_info() {
             .as_bytes()
             .to_vec(),
     };
-    let mut controller = QueryManager::new(stream, test_config());
+    let mut controller = QueryManager::new(stream, test_config(), EnDecoder);
     // WHEN
     controller
         .handle(&cache_dispatcher, ttl_inbox)