Repository: Migorithm/duva
Pull request #119 (2024-12-24 05:23:52)

refactor/unify-services-structure
### Pull Request Description

This pull request focuses on refactoring and unifying the structure of the services within the codebase. The changes aim to improve organization, enhance readability, and maintainability of the code. Below is a summary of the key modifications made:

- **Module Restructuring**: 
  - Moved `ClusterCommand` to a separate module for better separation of concerns.
  - Relocated `ClusterManager` to `cluster/manager.rs`, streamlining the cluster management functionality.
  - Consolidated request controllers into a unified folder to simplify access and management.

- **Naming Improvements**: 
  - Renamed various modules to better reflect their purpose and functionality.
  - Changed the name of `Config` to `ConfigActor` to clarify its role within the system.

- **Code Corrections**: 
  - Fixed a typo in an enum variant name to ensure consistency and correctness.

- **Separation of Concerns**: 
  - Created a separate TTL manager to enhance modularity and focus on specific functionalities.


Patch:
diff --git a/src/adapters/io/tokio_stream.rs b/src/adapters/io/tokio_stream.rs
index 585aa5b4..e1388a84 100644
--- a/src/adapters/io/tokio_stream.rs
+++ b/src/adapters/io/tokio_stream.rs
@@ -1,15 +1,9 @@
 use crate::services::{
     cluster::actor::PeerAddr,
     stream_manager::{
-        client_request_controllers::{
-            arguments::ClientRequestArguments, client_request::ClientRequest,
-        },
         error::IoError,
         interface::{TConnectStreamFactory, TExtractQuery, TGetPeerIp, TRead, TStream},
         query_io::{parse, QueryIO},
-        replication_request_controllers::{
-            arguments::PeerRequestArguments, replication_request::HandShakeRequest,
-        },
     },
 };
 use anyhow::Context;
@@ -19,6 +13,12 @@ use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::{TcpListener, TcpStream},
 };
+use crate::services::stream_manager::request_controller::client::{
+    arguments::ClientRequestArguments, client_request::ClientRequest,
+};
+use crate::services::stream_manager::request_controller::replica::{
+    arguments::PeerRequestArguments, replication_request::HandShakeRequest,
+};
 
 impl TStream for tokio::net::TcpStream {
     async fn read_value(&mut self) -> anyhow::Result<QueryIO> {
diff --git a/src/lib.rs b/src/lib.rs
index 1189f335..1f548c3d 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,16 +1,17 @@
 pub mod adapters;
 pub mod macros;
 pub mod services;
-use crate::services::stream_manager::client_request_controllers::ClientRequestController;
-use adapters::io::tokio_stream::{TokioConnectStreamFactory, TokioStreamListenerFactory};
+use crate::adapters::io::tokio_stream::TokioConnectStreamFactory;
+use crate::adapters::io::tokio_stream::TokioStreamListenerFactory;
+use services::stream_manager::request_controller::client::ClientRequestController;
 use anyhow::Result;
 use services::cluster::actor::ClusterActor;
-use services::cluster::ClusterManager;
-use services::config::config_manager::ConfigManager;
+use services::cluster::manager::ClusterManager;
+use services::config::manager::ConfigManager;
 
-use services::statefuls::cache::cache_manager::CacheManager;
-use services::statefuls::cache::ttl_manager::TtlSchedulerInbox;
-use services::statefuls::persist::persist_actor::PersistActor;
+use services::statefuls::cache::manager::CacheManager;
+use services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
+use services::statefuls::persist::actor::PersistActor;
 
 use services::stream_manager::error::IoError;
 use services::stream_manager::interface::TCancellationTokenFactory;
diff --git a/src/main.rs b/src/main.rs
index 69990581..102f1296 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -2,7 +2,7 @@ use redis_starter_rust::{
     adapters::cancellation_token::CancellationTokenFactory,
     services::{
         cluster::actor::ClusterActor,
-        config::{config_actor::Config, config_manager::ConfigManager},
+        config::{actor::ConfigActor, manager::ConfigManager},
     },
     StartUpFacade,
 };
@@ -10,7 +10,7 @@ use redis_starter_rust::{
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     // bootstrap dependencies
-    let config_manager = ConfigManager::new(Config::default());
+    let config_manager = ConfigManager::new(ConfigActor::default());
     let cluster_actor = ClusterActor::new();
     let start_up_runner =
         StartUpFacade::new(CancellationTokenFactory, config_manager, cluster_actor);
diff --git a/src/services/cluster/actor.rs b/src/services/cluster/actor.rs
index f3f317b7..bfc2cec9 100644
--- a/src/services/cluster/actor.rs
+++ b/src/services/cluster/actor.rs
@@ -3,13 +3,7 @@ use std::collections::BTreeMap;
 use tokio::{net::TcpStream, sync::mpsc::Receiver};
 
 use crate::make_smart_pointer;
-
-pub enum ClusterCommand {
-    AddPeer(PeerAddr, TcpStream),
-    RemovePeer(PeerAddr),
-    GetPeer(PeerAddr),
-    GetPeers(tokio::sync::oneshot::Sender<Vec<PeerAddr>>),
-}
+use crate::services::cluster::command::ClusterCommand;
 
 pub struct ClusterActor {
     pub peers: BTreeMap<PeerAddr, Connected>,
@@ -24,7 +18,7 @@ pub enum Connected {
         peer_stream: TcpStream,
         replication_stream: TcpStream,
     },
-    ClusterMemeber {
+    ClusterMember {
         peer_stream: TcpStream,
     },
     None,
diff --git a/src/services/cluster/command.rs b/src/services/cluster/command.rs
new file mode 100644
index 00000000..31e99986
--- /dev/null
+++ b/src/services/cluster/command.rs
@@ -0,0 +1,9 @@
+use tokio::net::TcpStream;
+use crate::services::cluster::actor::PeerAddr;
+
+pub enum ClusterCommand {
+    AddPeer(PeerAddr, TcpStream),
+    RemovePeer(PeerAddr),
+    GetPeer(PeerAddr),
+    GetPeers(tokio::sync::oneshot::Sender<Vec<PeerAddr>>),
+}
\ No newline at end of file
diff --git a/src/services/cluster/manager.rs b/src/services/cluster/manager.rs
new file mode 100644
index 00000000..bc2422f9
--- /dev/null
+++ b/src/services/cluster/manager.rs
@@ -0,0 +1,21 @@
+use tokio::sync::mpsc::Sender;
+use crate::services::cluster::actor::{ClusterActor, PeerAddr};
+use crate::services::cluster::command::ClusterCommand;
+
+#[derive(Clone)]
+pub struct ClusterManager(Sender<ClusterCommand>);
+
+impl ClusterManager {
+    pub fn run(actor: ClusterActor) -> Self {
+        let (tx, rx) = tokio::sync::mpsc::channel(100);
+        tokio::spawn(actor.handle(rx));
+        Self(tx)
+    }
+
+    pub(crate) async fn get_peers(&self) -> anyhow::Result<Vec<PeerAddr>> {
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        self.0.send(ClusterCommand::GetPeers(tx)).await?;
+        let peers = rx.await?;
+        Ok(peers)
+    }
+}
\ No newline at end of file
diff --git a/src/services/cluster/mod.rs b/src/services/cluster/mod.rs
index 81dc8651..df88baa5 100644
--- a/src/services/cluster/mod.rs
+++ b/src/services/cluster/mod.rs
@@ -1,21 +1,4 @@
 pub mod actor;
-use actor::{ClusterActor, ClusterCommand, PeerAddr};
-use tokio::sync::mpsc::Sender;
+mod command;
+pub mod manager;
 
-#[derive(Clone)]
-pub struct ClusterManager(Sender<ClusterCommand>);
-
-impl ClusterManager {
-    pub fn run(actor: ClusterActor) -> Self {
-        let (tx, rx) = tokio::sync::mpsc::channel(100);
-        tokio::spawn(actor.handle(rx));
-        Self(tx)
-    }
-
-    pub(crate) async fn get_peers(&self) -> anyhow::Result<Vec<PeerAddr>> {
-        let (tx, rx) = tokio::sync::oneshot::channel();
-        self.0.send(ClusterCommand::GetPeers(tx)).await?;
-        let peers = rx.await?;
-        Ok(peers)
-    }
-}
diff --git a/src/services/config/config_actor.rs b/src/services/config/actor.rs
similarity index 96%
rename from src/services/config/config_actor.rs
rename to src/services/config/actor.rs
index 7acd10ba..074a69e4 100644
--- a/src/services/config/config_actor.rs
+++ b/src/services/config/actor.rs
@@ -7,13 +7,13 @@ use crate::{env_var, services::config::replication::Replication};
 use tokio::sync::mpsc::Receiver;
 
 #[derive(Clone)]
-pub struct Config {
+pub struct ConfigActor {
     pub(crate) dir: &'static str,
     pub dbfilename: &'static str,
     pub replication: Replication,
 }
 
-impl Config {
+impl ConfigActor {
     pub async fn handle(mut self, mut inbox: Receiver<ConfigMessage>) {
         // inner state
 
@@ -72,7 +72,7 @@ impl Config {
     }
 }
 
-impl Default for Config {
+impl Default for ConfigActor {
     fn default() -> Self {
         env_var!(
             {
@@ -92,7 +92,7 @@ impl Default for Config {
                 .collect::<(_, _)>()
         });
 
-        Config {
+        ConfigActor {
             dir: Box::leak(dir.into_boxed_str()),
             dbfilename: Box::leak(dbfilename.into_boxed_str()),
             replication: Replication::new(replicaof),
@@ -102,7 +102,7 @@ impl Default for Config {
 
 #[test]
 fn test_set_dir() {
-    let mut config = Config::default();
+    let mut config = ConfigActor::default();
     config.set_dir("test");
     assert_eq!(config.dir, "test");
     config.set_dir("test2");
diff --git a/src/services/config/config_manager.rs b/src/services/config/manager.rs
similarity index 97%
rename from src/services/config/config_manager.rs
rename to src/services/config/manager.rs
index 2badadd6..f0e68e64 100644
--- a/src/services/config/config_manager.rs
+++ b/src/services/config/manager.rs
@@ -5,7 +5,7 @@ use crate::env_var;
 use super::command::ConfigCommand;
 use super::command::ConfigMessage;
 use super::command::ConfigQuery;
-use super::config_actor::Config;
+use super::actor::ConfigActor;
 use super::ConfigResource;
 use super::ConfigResponse;
 
@@ -31,7 +31,7 @@ impl std::ops::Deref for ConfigManager {
 }
 
 impl ConfigManager {
-    pub fn new(config: Config) -> Self {
+    pub fn new(config: ConfigActor) -> Self {
         let (tx, inbox) = tokio::sync::mpsc::channel(20);
         tokio::spawn(config.handle(inbox));
 
diff --git a/src/services/config/mod.rs b/src/services/config/mod.rs
index 1deaef4b..b24a090a 100644
--- a/src/services/config/mod.rs
+++ b/src/services/config/mod.rs
@@ -1,6 +1,6 @@
 mod command;
-pub mod config_actor;
-pub mod config_manager;
+pub mod actor;
+pub mod manager;
 pub mod macros;
 pub mod replication;
 
diff --git a/src/services/statefuls/cache/cache_actor.rs b/src/services/statefuls/cache/actor.rs
similarity index 97%
rename from src/services/statefuls/cache/cache_actor.rs
rename to src/services/statefuls/cache/actor.rs
index 6641f04d..5da1291d 100644
--- a/src/services/statefuls/cache/cache_actor.rs
+++ b/src/services/statefuls/cache/actor.rs
@@ -1,8 +1,8 @@
-use super::ttl_manager::TtlSchedulerInbox;
+use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use super::CacheEntry;
 use super::CacheValue;
 use crate::make_smart_pointer;
-use crate::services::statefuls::persist::persist_actor::SaveCommand;
+use crate::services::statefuls::persist::save_command::SaveCommand;
 use crate::services::stream_manager::query_io::QueryIO;
 use anyhow::Context;
 use anyhow::Result;
diff --git a/src/services/statefuls/cache/cache_manager.rs b/src/services/statefuls/cache/manager.rs
similarity index 94%
rename from src/services/statefuls/cache/cache_manager.rs
rename to src/services/statefuls/cache/manager.rs
index 07b3f82c..20c6ca6b 100644
--- a/src/services/statefuls/cache/cache_manager.rs
+++ b/src/services/statefuls/cache/manager.rs
@@ -1,15 +1,16 @@
-use super::cache_actor::{CacheActor, CacheCommand, CacheCommandSender};
-use super::ttl_manager::{TtlActor, TtlSchedulerInbox};
+use super::actor::{CacheActor, CacheCommand, CacheCommandSender};
+use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use super::CacheEntry;
 use crate::services::stream_manager::query_io::QueryIO;
 
-use crate::services::statefuls::persist::persist_actor::SaveCommand;
+use crate::services::statefuls::persist::save_command::SaveCommand;
 use crate::services::statefuls::persist::DumpFile;
 use anyhow::Result;
 use std::time::SystemTime;
 use std::{hash::Hasher, iter::Zip};
 use tokio::sync::mpsc;
 use tokio::sync::oneshot::Sender;
+use crate::services::statefuls::cache::ttl::actor::TtlActor;
 
 type OneShotSender<T> = tokio::sync::oneshot::Sender<T>;
 type OneShotReceiverJoinHandle<T> =
diff --git a/src/services/statefuls/cache/mod.rs b/src/services/statefuls/cache/mod.rs
index 76fd752f..f6531ceb 100644
--- a/src/services/statefuls/cache/mod.rs
+++ b/src/services/statefuls/cache/mod.rs
@@ -1,9 +1,9 @@
 use std::time::SystemTime;
 
-pub mod cache_actor;
-pub mod cache_manager;
+pub mod actor;
+pub mod manager;
 
-pub mod ttl_manager;
+pub mod ttl;
 
 #[derive(Debug, Clone)]
 pub enum CacheEntry {
diff --git a/src/services/statefuls/cache/ttl_manager.rs b/src/services/statefuls/cache/ttl/actor.rs
similarity index 67%
rename from src/services/statefuls/cache/ttl_manager.rs
rename to src/services/statefuls/cache/ttl/actor.rs
index 380a818a..7ef929d4 100644
--- a/src/services/statefuls/cache/ttl_manager.rs
+++ b/src/services/statefuls/cache/ttl/actor.rs
@@ -1,19 +1,13 @@
-use crate::{
-    make_smart_pointer,
-    services::statefuls::cache::{cache_actor::CacheCommand, cache_manager::CacheManager},
-};
-use std::{
-    cmp::Reverse,
-    collections::BinaryHeap,
-    time::{Duration, SystemTime},
-};
-use tokio::{
-    sync::{
-        mpsc::{self, Receiver},
-        oneshot,
-    },
-    time::interval,
-};
+use tokio::sync::mpsc;
+use tokio::time::interval;
+use std::time::{Duration, SystemTime};
+use tokio::sync::mpsc::Receiver;
+use std::collections::BinaryHeap;
+use std::cmp::Reverse;
+use crate::services::statefuls::cache::actor::CacheCommand;
+use crate::services::statefuls::cache::manager::CacheManager;
+use crate::services::statefuls::cache::ttl::command::TtlCommand;
+use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 
 pub struct TtlActor;
 
@@ -79,32 +73,4 @@ impl TtlActor {
         }
         Ok(())
     }
-}
-
-pub enum TtlCommand {
-    ScheduleTtl { expiry: SystemTime, key: String },
-    Peek(oneshot::Sender<Option<String>>),
-    StopSentinel,
-}
-
-impl TtlCommand {
-    pub fn get_expiration(self) -> Option<(SystemTime, String)> {
-        let (expire_in_mills, key) = match self {
-            TtlCommand::ScheduleTtl { expiry, key } => (expiry, key),
-            _ => return None,
-        };
-
-        Some((expire_in_mills, key))
-    }
-}
-
-#[derive(Clone)]
-pub struct TtlSchedulerInbox(pub(crate) tokio::sync::mpsc::Sender<TtlCommand>);
-
-impl TtlSchedulerInbox {
-    pub async fn set_ttl(&self, key: String, expiry: SystemTime) {
-        let _ = self.send(TtlCommand::ScheduleTtl { key, expiry }).await;
-    }
-}
-
-make_smart_pointer!(TtlSchedulerInbox, tokio::sync::mpsc::Sender<TtlCommand>);
+}
\ No newline at end of file
diff --git a/src/services/statefuls/cache/ttl/command.rs b/src/services/statefuls/cache/ttl/command.rs
new file mode 100644
index 00000000..7069f30a
--- /dev/null
+++ b/src/services/statefuls/cache/ttl/command.rs
@@ -0,0 +1,19 @@
+use std::time::SystemTime;
+use tokio::sync::oneshot;
+
+pub enum TtlCommand {
+    ScheduleTtl { expiry: SystemTime, key: String },
+    Peek(oneshot::Sender<Option<String>>),
+    StopSentinel,
+}
+
+impl TtlCommand {
+    pub fn get_expiration(self) -> Option<(SystemTime, String)> {
+        let (expire_in_mills, key) = match self {
+            TtlCommand::ScheduleTtl { expiry, key } => (expiry, key),
+            _ => return None,
+        };
+
+        Some((expire_in_mills, key))
+    }
+}
\ No newline at end of file
diff --git a/src/services/statefuls/cache/ttl/manager.rs b/src/services/statefuls/cache/ttl/manager.rs
new file mode 100644
index 00000000..dada2660
--- /dev/null
+++ b/src/services/statefuls/cache/ttl/manager.rs
@@ -0,0 +1,14 @@
+use crate::make_smart_pointer;
+use std::time::SystemTime;
+use crate::services::statefuls::cache::ttl::command::TtlCommand;
+
+#[derive(Clone)]
+pub struct TtlSchedulerInbox(pub(crate) tokio::sync::mpsc::Sender<TtlCommand>);
+
+impl TtlSchedulerInbox {
+    pub async fn set_ttl(&self, key: String, expiry: SystemTime) {
+        let _ = self.send(TtlCommand::ScheduleTtl { key, expiry }).await;
+    }
+}
+
+make_smart_pointer!(TtlSchedulerInbox, tokio::sync::mpsc::Sender<TtlCommand>);
diff --git a/src/services/statefuls/cache/ttl/mod.rs b/src/services/statefuls/cache/ttl/mod.rs
new file mode 100644
index 00000000..8c366299
--- /dev/null
+++ b/src/services/statefuls/cache/ttl/mod.rs
@@ -0,0 +1,3 @@
+pub mod manager;
+pub(crate) mod actor;
+mod command;
\ No newline at end of file
diff --git a/src/services/statefuls/persist/persist_actor.rs b/src/services/statefuls/persist/actor.rs
similarity index 90%
rename from src/services/statefuls/persist/persist_actor.rs
rename to src/services/statefuls/persist/actor.rs
index 83c6fb7d..73e30244 100644
--- a/src/services/statefuls/persist/persist_actor.rs
+++ b/src/services/statefuls/persist/actor.rs
@@ -3,18 +3,9 @@ use super::endec::decoder::states::DecoderInit;
 use super::endec::encoder::encoding_processor::SaveMeta;
 use super::endec::encoder::encoding_processor::SavingProcessor;
 use super::DumpFile;
-use crate::services::statefuls::cache::CacheEntry;
 use crate::services::stream_manager::interface::TWriterFactory;
 use tokio::sync::mpsc::{Receiver, Sender};
-
-pub enum SaveCommand {
-    LocalShardSize {
-        table_size: usize,
-        expiry_size: usize,
-    },
-    SaveChunk(Vec<CacheEntry>),
-    StopSentinel,
-}
+use crate::services::statefuls::persist::save_command::SaveCommand;
 
 pub struct Load;
 
diff --git a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
index dfd66920..6408eac5 100644
--- a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
+++ b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
@@ -1,7 +1,7 @@
 use crate::services::statefuls::cache::CacheEntry;
 use crate::services::stream_manager::interface::TWrite;
 
-use crate::services::statefuls::persist::persist_actor::SaveCommand;
+use crate::services::statefuls::persist::save_command::SaveCommand;
 use anyhow::Result;
 
 use std::collections::VecDeque;
diff --git a/src/services/statefuls/persist/mod.rs b/src/services/statefuls/persist/mod.rs
index 79f313a5..6fa73ba5 100644
--- a/src/services/statefuls/persist/mod.rs
+++ b/src/services/statefuls/persist/mod.rs
@@ -1,5 +1,6 @@
 pub mod endec;
-pub mod persist_actor;
+pub mod actor;
+pub mod save_command;
 
 use std::collections::HashMap;
 
diff --git a/src/services/statefuls/persist/save_command.rs b/src/services/statefuls/persist/save_command.rs
new file mode 100644
index 00000000..99e9253a
--- /dev/null
+++ b/src/services/statefuls/persist/save_command.rs
@@ -0,0 +1,10 @@
+use crate::services::statefuls::cache::CacheEntry;
+
+pub enum SaveCommand {
+    LocalShardSize {
+        table_size: usize,
+        expiry_size: usize,
+    },
+    SaveChunk(Vec<CacheEntry>),
+    StopSentinel,
+}
\ No newline at end of file
diff --git a/src/services/stream_manager/mod.rs b/src/services/stream_manager/mod.rs
index 1157321d..0e60c003 100644
--- a/src/services/stream_manager/mod.rs
+++ b/src/services/stream_manager/mod.rs
@@ -1,13 +1,13 @@
-pub mod client_request_controllers;
 pub mod error;
 pub mod interface;
 pub mod query_io;
-pub mod replication_request_controllers;
+pub(crate) mod request_controller;
+
 use super::cluster::actor::PeerAddr;
-use super::cluster::ClusterManager;
-use crate::services::stream_manager::client_request_controllers::client_request::ClientRequest;
-use crate::services::stream_manager::client_request_controllers::ClientRequestController;
-use client_request_controllers::arguments::ClientRequestArguments;
+use crate::services::cluster::manager::ClusterManager;
+use request_controller::client::client_request::ClientRequest;
+use request_controller::client::ClientRequestController;
+use request_controller::client::arguments::ClientRequestArguments;
 use error::IoError;
 use interface::TCancellationNotifier;
 use interface::TCancellationTokenFactory;
@@ -16,8 +16,8 @@ use interface::TExtractQuery;
 use interface::TStream;
 use interface::TWriterFactory;
 use query_io::QueryIO;
-use replication_request_controllers::arguments::PeerRequestArguments;
-use replication_request_controllers::replication_request::HandShakeRequest;
+use request_controller::replica::arguments::PeerRequestArguments;
+use request_controller::replica::replication_request::HandShakeRequest;
 
 use std::time::Duration;
 use tokio::{task::yield_now, time::interval};
diff --git a/src/services/stream_manager/client_request_controllers/arguments.rs b/src/services/stream_manager/request_controller/client/arguments.rs
similarity index 100%
rename from src/services/stream_manager/client_request_controllers/arguments.rs
rename to src/services/stream_manager/request_controller/client/arguments.rs
diff --git a/src/services/stream_manager/client_request_controllers/client_request.rs b/src/services/stream_manager/request_controller/client/client_request.rs
similarity index 100%
rename from src/services/stream_manager/client_request_controllers/client_request.rs
rename to src/services/stream_manager/request_controller/client/client_request.rs
diff --git a/src/services/stream_manager/client_request_controllers/mod.rs b/src/services/stream_manager/request_controller/client/mod.rs
similarity index 91%
rename from src/services/stream_manager/client_request_controllers/mod.rs
rename to src/services/stream_manager/request_controller/client/mod.rs
index d1419f2f..96ae2528 100644
--- a/src/services/stream_manager/client_request_controllers/mod.rs
+++ b/src/services/stream_manager/request_controller/client/mod.rs
@@ -1,15 +1,15 @@
-use crate::services::config::config_manager::ConfigManager;
+use crate::services::config::manager::ConfigManager;
 use crate::services::config::ConfigResponse;
-use crate::services::statefuls::cache::cache_manager::CacheManager;
-use crate::services::statefuls::cache::ttl_manager::TtlSchedulerInbox;
+use crate::services::statefuls::cache::manager::CacheManager;
+use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use crate::services::statefuls::persist::endec::encoder::encoding_processor::SavingProcessor;
-use crate::services::statefuls::persist::persist_actor::PersistActor;
+use crate::services::statefuls::persist::actor::PersistActor;
 use crate::services::stream_manager::interface::TCancellationWatcher;
 use crate::services::stream_manager::query_io::QueryIO;
 use arguments::ClientRequestArguments;
 use client_request::ClientRequest;
 
-use super::interface::TWriterFactory;
+use crate::services::stream_manager::interface::TWriterFactory;
 
 pub mod arguments;
 pub mod client_request;
diff --git a/src/services/stream_manager/request_controller/mod.rs b/src/services/stream_manager/request_controller/mod.rs
new file mode 100644
index 00000000..842e582b
--- /dev/null
+++ b/src/services/stream_manager/request_controller/mod.rs
@@ -0,0 +1,2 @@
+pub mod client;
+pub mod replica;
\ No newline at end of file


Test patch:
diff --git a/src/services/stream_manager/replication_request_controllers/arguments.rs b/src/services/stream_manager/request_controller/replica/arguments.rs
similarity index 100%
rename from src/services/stream_manager/replication_request_controllers/arguments.rs
rename to src/services/stream_manager/request_controller/replica/arguments.rs
diff --git a/src/services/stream_manager/replication_request_controllers/mod.rs b/src/services/stream_manager/request_controller/replica/mod.rs
similarity index 100%
rename from src/services/stream_manager/replication_request_controllers/mod.rs
rename to src/services/stream_manager/request_controller/replica/mod.rs
diff --git a/src/services/stream_manager/replication_request_controllers/replication_request.rs b/src/services/stream_manager/request_controller/replica/replication_request.rs
similarity index 100%
rename from src/services/stream_manager/replication_request_controllers/replication_request.rs
rename to src/services/stream_manager/request_controller/replica/replication_request.rs
diff --git a/tests/common.rs b/tests/common.rs
index ed125670..d3339ff5 100644
--- a/tests/common.rs
+++ b/tests/common.rs
@@ -1,6 +1,6 @@
 use redis_starter_rust::services::cluster::actor::{ClusterActor, Connected, PeerAddr};
-use redis_starter_rust::services::config::config_actor::Config;
-use redis_starter_rust::services::config::config_manager::ConfigManager;
+use redis_starter_rust::services::config::actor::ConfigActor;
+use redis_starter_rust::services::config::manager::ConfigManager;
 use redis_starter_rust::services::stream_manager::interface::{TCancellationTokenFactory, TStream};
 use redis_starter_rust::services::stream_manager::query_io::QueryIO;
 use redis_starter_rust::{StartUpFacade, TNotifyStartUp};
@@ -53,7 +53,7 @@ impl<'a> TestStreamHandler<'a> {
 }
 
 pub async fn init_config_manager_with_free_port() -> ConfigManager {
-    let config = Config::default();
+    let config = ConfigActor::default();
     let mut manager = ConfigManager::new(config);
 
     manager.port = find_free_port_in_range(49152, 65535).await.unwrap();
@@ -61,7 +61,7 @@ pub async fn init_config_manager_with_free_port() -> ConfigManager {
 }
 
 pub async fn init_slave_config_manager_with_free_port(master_port: u16) -> ConfigManager {
-    let mut config = Config::default();
+    let mut config = ConfigActor::default();
     config.replication.master_host = Some("localhost".to_string());
     config.replication.master_port = Some(master_port);
     let mut manager = ConfigManager::new(config);
diff --git a/tests/test_disseminate_peers.rs b/tests/test_disseminate_peers.rs
index 7a4ff8b3..d89f0de3 100644
--- a/tests/test_disseminate_peers.rs
+++ b/tests/test_disseminate_peers.rs
@@ -7,7 +7,7 @@ use common::{
 use redis_starter_rust::{
     adapters::cancellation_token::CancellationTokenFactory,
     services::{
-        config::{config_actor::Config, config_manager::ConfigManager},
+        config::{actor::ConfigActor, manager::ConfigManager},
         stream_manager::interface::TStream,
     },
 };
@@ -16,7 +16,7 @@ use tokio::net::TcpStream;
 #[tokio::test]
 async fn test_disseminate_peers() {
     // GIVEN - master server configuration
-    let config = Config::default();
+    let config = ConfigActor::default();
     let peer_address_to_test = "localhost:6378";
 
     let mut manager = ConfigManager::new(config);
diff --git a/tests/test_handshake.rs b/tests/test_handshake.rs
index 24daff19..21b1055a 100644
--- a/tests/test_handshake.rs
+++ b/tests/test_handshake.rs
@@ -13,7 +13,7 @@ use redis_starter_rust::{
     adapters::cancellation_token::CancellationTokenFactory,
     services::{
         cluster::actor::ClusterActor,
-        config::{config_actor::Config, config_manager::ConfigManager},
+        config::{actor::ConfigActor, manager::ConfigManager},
     },
 };
 use tokio::net::TcpStream;
@@ -22,7 +22,7 @@ mod common;
 #[tokio::test]
 async fn test_threeway_handshake() {
     // GIVEN - master server configuration
-    let config = Config::default();
+    let config = ConfigActor::default();
     let mut manager = ConfigManager::new(config);
 
     // ! `peer_bind_addr` is bind_addr dedicated for peer connections
diff --git a/tests/test_heartbeat.rs b/tests/test_heartbeat.rs
index f5aeeddb..b669eb2c 100644
--- a/tests/test_heartbeat.rs
+++ b/tests/test_heartbeat.rs
@@ -11,7 +11,7 @@ use redis_starter_rust::{
     adapters::cancellation_token::CancellationTokenFactory,
     services::{
         cluster::actor::ClusterActor,
-        config::{config_actor::Config, config_manager::ConfigManager},
+        config::{actor::ConfigActor, manager::ConfigManager},
         stream_manager::interface::TStream,
     },
 };
@@ -39,7 +39,7 @@ async fn replica_server_helper(replica_port: u16) {
 async fn test_heartbeat() {
     // GIVEN
     // run the random server on a random port
-    let config = Config::default();
+    let config = ConfigActor::default();
     let mut manager = ConfigManager::new(config);
     manager.port = find_free_port_in_range(6000, 6553).await.unwrap();
     let master_cluster_bind_addr = manager.peer_bind_addr();
@@ -63,7 +63,7 @@ async fn test_heartbeat() {
 async fn test_heartbeat_sent_to_multiple_replicas() {
     // GIVEN
     // run the random server on a random port
-    let config = Config::default();
+    let config = ConfigActor::default();
     let mut manager = ConfigManager::new(config);
     manager.port = find_free_port_in_range(6000, 6553).await.unwrap();
     let master_cluster_bind_addr = manager.peer_bind_addr();
diff --git a/tests/test_heartbeat_broadcasting.rs b/tests/test_heartbeat_broadcasting.rs
index 6eacad65..8b77fc47 100644
--- a/tests/test_heartbeat_broadcasting.rs
+++ b/tests/test_heartbeat_broadcasting.rs
@@ -7,7 +7,7 @@ use common::{
 use redis_starter_rust::{
     adapters::cancellation_token::CancellationTokenFactory,
     services::{
-        config::{config_actor::Config, config_manager::ConfigManager},
+        config::{actor::ConfigActor, manager::ConfigManager},
         stream_manager::interface::TStream,
     },
 };
@@ -39,7 +39,7 @@ async fn test_heartbeat_sent_to_multiple_replicas() {
     let fake_repl_port = 6781;
     let fake_repl_address = "localhost:".to_string() + &fake_repl_port.to_string();
     // create master server with fake replica address as peers
-    let config = Config::default();
+    let config = ConfigActor::default();
     let mut manager = ConfigManager::new(config);
 
     manager.port = find_free_port_in_range(6000, 6553).await.unwrap();
@@ -56,7 +56,7 @@ async fn test_heartbeat_sent_to_multiple_replicas() {
     // WHEN - new replica is connecting to master
     {
         let connecting_replica_port = 6782;
-        let mut config = Config::default();
+        let mut config = ConfigActor::default();
         config.replication.master_port = Some(manager.port);
         config.replication.master_host = Some("localhost".to_string());
         let mut manager = ConfigManager::new(config);