Repository: Migorithm/duva
Pull request #134 (2025-01-06 20:27:52)

refactor: simplifying role-specific handlers


Patch:
diff --git a/src/adapters/io/tokio_stream.rs b/src/adapters/io/tokio_stream.rs
index 09796ffd..deac6a5d 100644
--- a/src/adapters/io/tokio_stream.rs
+++ b/src/adapters/io/tokio_stream.rs
@@ -3,11 +3,8 @@ use crate::services::interface::{TGetPeerIp, TRead, TStream, TWrite};
 use crate::services::query_io::{parse, QueryIO};
 use bytes::BytesMut;
 use std::io::ErrorKind;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::tcp::OwnedWriteHalf;
-use tokio::{
-    io::{AsyncReadExt, AsyncWriteExt},
-    net::TcpListener,
-};
 
 impl TStream for tokio::net::TcpStream {
     async fn read_value(&mut self) -> anyhow::Result<QueryIO> {
@@ -99,13 +96,6 @@ impl TGetPeerIp for tokio::net::TcpStream {
     }
 }
 
-pub struct TokioStreamListenerFactory;
-impl TokioStreamListenerFactory {
-    pub async fn create_listner(&self, bind_addr: &str) -> TcpListener {
-        TcpListener::bind(bind_addr).await.expect("failed to bind")
-    }
-}
-
 impl From<ErrorKind> for IoError {
     fn from(value: ErrorKind) -> Self {
         match value {
diff --git a/src/lib.rs b/src/lib.rs
index 08060b15..fffdcee8 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,23 +1,19 @@
 pub mod adapters;
 pub mod macros;
 pub mod services;
-use crate::adapters::io::tokio_stream::TokioStreamListenerFactory;
 use anyhow::Result;
+use services::client_request_controller::ClientRequestController;
 use services::cluster::actor::ClusterActor;
 use services::cluster::inbound_stream::InboundStream;
 use services::cluster::manager::ClusterManager;
-use services::cluster::outbound_stream::OutboundStream;
 use services::config::manager::ConfigManager;
-use services::config::{ConfigResource, ConfigResponse};
-
-use services::client_request_controller::ClientRequestController;
 use services::error::IoError;
 use services::interface::TCancellationTokenFactory;
 use services::statefuls::cache::manager::CacheManager;
 use services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use services::statefuls::persist::actor::PersistActor;
 use std::time::Duration;
-use tokio::net::TcpStream;
+use tokio::net::TcpListener;
 
 // * StartUp Facade that manages invokes subsystems
 pub struct StartUpFacade<V>
@@ -90,9 +86,10 @@ where
         peer_bind_addr: String,
         cluster_manager: &'static ClusterManager,
     ) {
-        let peer_listener = TokioStreamListenerFactory
-            .create_listner(&peer_bind_addr)
-            .await;
+        let peer_listener = TcpListener::bind(&peer_bind_addr)
+            .await
+            .expect("[ERROR] Failed to bind to peer address for listening");
+
         println!("Starting to accept peer connections");
         println!("listening on {}...", peer_bind_addr);
 
@@ -118,32 +115,23 @@ where
             let (stop_sentinel_tx, stop_sentinel_recv) = tokio::sync::oneshot::channel::<()>();
 
             if is_master_mode {
-                tokio::spawn(
-                    self.client_request_controller
-                        .run_client_connection_handling_actor(
-                            self.config_manager.bind_addr(),
-                            self.cancellation_factory,
-                            stop_sentinel_recv,
-                        ),
-                );
+                let client_stream_listener =
+                    TcpListener::bind(&self.config_manager.bind_addr()).await?;
+                tokio::spawn(self.client_request_controller.receive_clients(
+                    self.cancellation_factory,
+                    stop_sentinel_recv,
+                    client_stream_listener,
+                ));
             } else {
                 // Cancel all client connections only IF the cluster mode has changes to slave
-                stop_sentinel_tx.send(()).unwrap();
-
-                let ConfigResponse::ReplicationInfo(repl_info) = self
-                    .config_manager
-                    .route_query(ConfigResource::ReplicationInfo)
-                    .await?
-                else {
-                    unreachable!()
-                };
-
-                tokio::spawn(self.cluster_manager.join_peer(
-                    OutboundStream(TcpStream::connect(repl_info.master_cluster_bind_addr()).await?),
-                    repl_info,
-                    self.config_manager.port,
-                ));
+                let _ = stop_sentinel_tx.send(());
+                let repl_info = self.config_manager.replication_info().await?;
+                tokio::spawn(
+                    self.cluster_manager
+                        .join_master(repl_info, self.config_manager.port),
+                );
             }
+
             self.config_manager
                 .wait_until_cluster_mode_changed()
                 .await?;
diff --git a/src/main.rs b/src/main.rs
index b7f9ca32..dcedb51e 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -11,7 +11,7 @@ use redis_starter_rust::{
 async fn main() -> anyhow::Result<()> {
     // bootstrap dependencies
     let config_manager = ConfigManager::new(ConfigActor::default());
-    let cluster_actor = ClusterActor::new();
+    let cluster_actor = ClusterActor::default();
     let mut start_up_runner =
         StartUpFacade::new(CancellationTokenFactory, config_manager, cluster_actor);
     start_up_runner.run(()).await
diff --git a/src/services/client_request_controller/mod.rs b/src/services/client_request_controller/mod.rs
index 576fa572..18197b8c 100644
--- a/src/services/client_request_controller/mod.rs
+++ b/src/services/client_request_controller/mod.rs
@@ -1,23 +1,22 @@
-use crate::adapters::io::tokio_stream::TokioStreamListenerFactory;
 use crate::services::config::manager::ConfigManager;
 use crate::services::config::ConfigResponse;
-
-use crate::services::interface::TWriterFactory;
-use crate::services::interface::{TCancellationTokenFactory, TCancellationWatcher};
+use crate::services::interface::{
+    TCancellationNotifier, TCancellationTokenFactory, TCancellationWatcher, TStream,
+};
 use crate::services::query_io::QueryIO;
 use crate::services::statefuls::cache::manager::CacheManager;
 use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use crate::services::statefuls::persist::actor::PersistActor;
 use crate::services::statefuls::persist::endec::encoder::encoding_processor::SavingProcessor;
-
 use arguments::ClientRequestArguments;
 use request::ClientRequest;
-use stream_manager::{ClientStream, ClientStreamManager};
+use stream::ClientStream;
+use tokio::net::TcpListener;
 use tokio::select;
 
 pub mod arguments;
 pub mod request;
-pub mod stream_manager;
+pub mod stream;
 
 pub(crate) struct ClientRequestController {
     config_manager: ConfigManager,
@@ -38,7 +37,7 @@ impl ClientRequestController {
         }
     }
 
-    pub(crate) async fn handle<T: TWriterFactory>(
+    pub(crate) async fn handle(
         &self,
         mut cancellation_token: impl TCancellationWatcher,
         cmd: ClientRequest,
@@ -61,7 +60,7 @@ impl ClientRequestController {
             }
             ClientRequest::Save => {
                 // spawn save actor
-                let outbox = PersistActor::<SavingProcessor<T>>::run(
+                let outbox = PersistActor::<SavingProcessor>::run(
                     self.config_manager.get_filepath().await?,
                     self.cache_manager.inboxes.len(),
                 )
@@ -106,34 +105,58 @@ impl ClientRequestController {
         Ok(response)
     }
 
-    pub async fn run_client_connection_handling_actor(
+    pub async fn receive_clients(
         &'static self,
-        bind_addr: String,
         cancellation_factory: impl TCancellationTokenFactory,
         stop_sentinel_recv: tokio::sync::oneshot::Receiver<()>,
+        client_stream_listener: TcpListener,
     ) {
-        let client_stream_listener = TokioStreamListenerFactory.create_listner(&bind_addr).await;
         let mut conn_handlers: Vec<tokio::task::JoinHandle<()>> = Vec::with_capacity(100);
+
         select! {
-            _ = stop_sentinel_recv => {
-                // Reconnection logic should be implemented by client?
-                conn_handlers.iter().for_each(|handle| handle.abort());
-            },
+            // The following closure doesn't take the ownership of the `conn_handlers` which enables us to abort the tasks
+            // when the sentinel is received.
             _ = async {
                     while let Ok((stream, _)) = client_stream_listener.accept().await {
-                        // TODO refactoring
-                        let query_manager = ClientStreamManager::new(
-                            ClientStream(stream),
-                            self,
-                        );
-
                         conn_handlers.push(tokio::spawn(
-                            query_manager.handle_single_client_stream::<tokio::fs::File>(
-                                cancellation_factory.clone()
-                            ),
+                           async move{
+                                const TIMEOUT: u64 = 100;
+                                let mut stream =  ClientStream(stream);
+                                loop {
+                                    let Ok((request, query_args)) = stream.extract_query().await else {
+                                        eprintln!("invalid user request");
+                                        continue;
+                                    };
+
+                                    let (cancellation_notifier, cancellation_token) =
+                                        cancellation_factory.create(TIMEOUT);
+
+                                    // TODO subject to change - more to dynamic
+                                    // Notify the cancellation notifier to cancel the query after 100 milliseconds.
+                                    cancellation_notifier.notify();
+
+                                    let res = match self.handle(cancellation_token, request, query_args).await {
+                                        Ok(response) => stream.write(response).await,
+                                        Err(e) => stream.write(QueryIO::Err(e.to_string())).await,
+                                    };
+
+                                    if let Err(e) = res {
+                                        if e.should_break() {
+                                            break;
+                                        }
+                                    }
+                                }
+                           }
                         ));
                     }
                 } =>{}
+
+
+            _ = stop_sentinel_recv => {
+                // Reconnection logic should be implemented by client?
+                conn_handlers.iter().for_each(|handle| handle.abort());
+                },
+
         };
     }
 }
diff --git a/src/services/client_request_controller/stream.rs b/src/services/client_request_controller/stream.rs
new file mode 100644
index 00000000..1c627d41
--- /dev/null
+++ b/src/services/client_request_controller/stream.rs
@@ -0,0 +1,32 @@
+use crate::{
+    make_smart_pointer,
+    services::{interface::TStream, query_io::QueryIO},
+};
+use anyhow::Context;
+
+use tokio::net::TcpStream;
+
+use super::{arguments::ClientRequestArguments, request::ClientRequest};
+
+pub struct ClientStream(pub(crate) TcpStream);
+make_smart_pointer!(ClientStream, TcpStream);
+
+impl ClientStream {
+    pub(crate) async fn extract_query(
+        &mut self,
+    ) -> anyhow::Result<(ClientRequest, ClientRequestArguments)> {
+        let query_io = self.read_value().await?;
+        match query_io {
+            QueryIO::Array(value_array) => Ok((
+                value_array
+                    .first()
+                    .context("request not given")?
+                    .clone()
+                    .unpack_bulk_str()?
+                    .try_into()?,
+                ClientRequestArguments::new(value_array.into_iter().skip(1).collect()),
+            )),
+            _ => Err(anyhow::anyhow!("Unexpected command format")),
+        }
+    }
+}
diff --git a/src/services/client_request_controller/stream_manager.rs b/src/services/client_request_controller/stream_manager.rs
deleted file mode 100644
index fe835978..00000000
--- a/src/services/client_request_controller/stream_manager.rs
+++ /dev/null
@@ -1,89 +0,0 @@
-use crate::{
-    make_smart_pointer,
-    services::{
-        error::IoError,
-        interface::{TCancellationNotifier, TCancellationTokenFactory, TStream},
-        query_io::QueryIO,
-    },
-};
-use anyhow::Context;
-
-use tokio::net::TcpStream;
-
-use super::{
-    arguments::ClientRequestArguments, request::ClientRequest, ClientRequestController,
-    TWriterFactory,
-};
-
-/// Controller is a struct that will be used to read and write values to the client.
-pub struct ClientStreamManager<U> {
-    pub(crate) stream: ClientStream,
-    pub(crate) controller: U,
-}
-
-pub struct ClientStream(pub(crate) TcpStream);
-make_smart_pointer!(ClientStream, TcpStream);
-
-impl<U> ClientStreamManager<U> {
-    pub(crate) fn new(stream: ClientStream, controller: U) -> Self {
-        ClientStreamManager { stream, controller }
-    }
-
-    pub async fn send(&mut self, value: QueryIO) -> Result<(), IoError> {
-        self.stream.write(value).await
-    }
-}
-
-impl ClientStreamManager<&'static ClientRequestController> {
-    pub(crate) async fn handle_single_client_stream<F: TWriterFactory>(
-        mut self,
-        cancellation_token_factory: impl TCancellationTokenFactory,
-    ) {
-        const TIMEOUT: u64 = 100;
-
-        loop {
-            let Ok((request, query_args)) = self.extract_query().await else {
-                eprintln!("invalid user request");
-                continue;
-            };
-
-            let (cancellation_notifier, cancellation_token) =
-                cancellation_token_factory.create(TIMEOUT);
-
-            // TODO subject to change - more to dynamic
-            // Notify the cancellation notifier to cancel the query after 100 milliseconds.
-            cancellation_notifier.notify();
-
-            let res = match self
-                .controller
-                .handle::<F>(cancellation_token, request, query_args)
-                .await
-            {
-                Ok(response) => self.send(response).await,
-                Err(e) => self.send(QueryIO::Err(e.to_string())).await,
-            };
-
-            if let Err(e) = res {
-                if e.should_break() {
-                    break;
-                }
-            }
-        }
-    }
-
-    async fn extract_query(&mut self) -> anyhow::Result<(ClientRequest, ClientRequestArguments)> {
-        let query_io = self.stream.read_value().await?;
-        match query_io {
-            QueryIO::Array(value_array) => Ok((
-                value_array
-                    .first()
-                    .context("request not given")?
-                    .clone()
-                    .unpack_bulk_str()?
-                    .try_into()?,
-                ClientRequestArguments::new(value_array.into_iter().skip(1).collect()),
-            )),
-            _ => Err(anyhow::anyhow!("Unexpected command format")),
-        }
-    }
-}
diff --git a/src/services/cluster/actor.rs b/src/services/cluster/actor.rs
index f28c9f15..44fc60a3 100644
--- a/src/services/cluster/actor.rs
+++ b/src/services/cluster/actor.rs
@@ -10,6 +10,7 @@ use tokio::sync::mpsc::Sender;
 use tokio::time::interval;
 use tokio::{net::TcpStream, sync::mpsc::Receiver};
 
+#[derive(Debug, Default)]
 pub struct ClusterActor {
     // TODO change PeerAddr to PeerIdentifier
     pub peers: HashSet<PeerAddr>,
@@ -19,12 +20,6 @@ pub struct PeerAddr(pub String);
 make_smart_pointer!(PeerAddr, String);
 
 impl ClusterActor {
-    pub fn new() -> Self {
-        Self {
-            peers: HashSet::new(),
-        }
-    }
-
     // * Add peer to the cluster
     // * This function is called when a new peer is connected to peer listener
     // * Some are replicas and some are cluster members
diff --git a/src/services/cluster/establishment/inbound.rs b/src/services/cluster/establishment/inbound.rs
index c6c1eb8d..5a021ffe 100644
--- a/src/services/cluster/establishment/inbound.rs
+++ b/src/services/cluster/establishment/inbound.rs
@@ -79,7 +79,7 @@ impl HandShakeRequest {
 
         let replica_id = self
             .args
-            .get(0)
+            .first()
             .map(|v| v.clone().unpack_bulk_str())
             .ok_or(anyhow::anyhow!("No replica id"))??;
         let offset = self
diff --git a/src/services/cluster/manager.rs b/src/services/cluster/manager.rs
index 2c7d1fd6..8a3b2bfd 100644
--- a/src/services/cluster/manager.rs
+++ b/src/services/cluster/manager.rs
@@ -63,15 +63,17 @@ impl ClusterManager {
         Ok(())
     }
 
-    pub(crate) async fn join_peer(
-        &self,
-        mut stream: OutboundStream,
+    pub async fn join_master(
+        &'static self,
         repl_info: Replication,
         self_port: u16,
-    ) {
-        stream
+    ) -> anyhow::Result<()> {
+        let mut outbound_stream =
+            OutboundStream(TcpStream::connect(repl_info.master_cluster_bind_addr()).await?);
+        outbound_stream
             .estabilish_handshake(repl_info, self_port)
-            .await
-            .expect("joining handshake failed");
+            .await?;
+
+        Ok(())
     }
 }
diff --git a/src/services/cluster/outbound_stream.rs b/src/services/cluster/outbound_stream.rs
index 8c4d8b61..30ec83f0 100644
--- a/src/services/cluster/outbound_stream.rs
+++ b/src/services/cluster/outbound_stream.rs
@@ -1,7 +1,7 @@
 use super::establishment::outbound::HandShakeResponse;
 use crate::services::config::replication::Replication;
-use crate::services::query_io::QueryIO;
 use crate::services::interface::TStream;
+use crate::services::query_io::QueryIO;
 use crate::{make_smart_pointer, write_array};
 use tokio::net::TcpStream;
 
diff --git a/src/services/config/actor.rs b/src/services/config/actor.rs
index f9156408..f55a19f8 100644
--- a/src/services/config/actor.rs
+++ b/src/services/config/actor.rs
@@ -32,18 +32,16 @@ impl ConfigActor {
                 match msg {
                     ConfigMessage::Query(query) => match query.resource {
                         ConfigResource::Dir => {
-                            let _ = query.respond_with(ConfigResponse::Dir(self.dir.into()));
+                            query.respond_with(ConfigResponse::Dir(self.dir.into()));
                         }
                         ConfigResource::DbFileName => {
-                            let _ = query
-                                .respond_with(ConfigResponse::DbFileName(self.dbfilename.into()));
+                            query.respond_with(ConfigResponse::DbFileName(self.dbfilename.into()));
                         }
                         ConfigResource::FilePath => {
-                            let _ =
-                                query.respond_with(ConfigResponse::FilePath(self.get_filepath()));
+                            query.respond_with(ConfigResponse::FilePath(self.get_filepath()));
                         }
                         ConfigResource::ReplicationInfo => {
-                            let _ = query.respond_with(ConfigResponse::ReplicationInfo(
+                            query.respond_with(ConfigResponse::ReplicationInfo(
                                 self.replication.clone(),
                             ));
                         }
diff --git a/src/services/config/replication.rs b/src/services/config/replication.rs
index 0c65707f..f65d27d9 100644
--- a/src/services/config/replication.rs
+++ b/src/services/config/replication.rs
@@ -42,11 +42,11 @@ impl Replication {
         ]
     }
     fn role(&self) -> String {
-        self.master_host
-            .is_some()
-            .then(|| "role:slave")
-            .unwrap_or("role:master")
-            .to_string()
+        if self.master_host.is_some() {
+            "role:slave".into()
+        } else {
+            "role:master".into()
+        }
     }
 
     pub fn master_cluster_bind_addr(&self) -> String {
diff --git a/src/services/error.rs b/src/services/error.rs
index c3458ef9..d37379b4 100644
--- a/src/services/error.rs
+++ b/src/services/error.rs
@@ -11,15 +11,15 @@ pub enum IoError {
 
 impl IoError {
     pub fn should_break(self) -> bool {
-        match self {
+        matches!(
+            self,
             IoError::ConnectionRefused
-            | IoError::ConnectionReset
-            | IoError::ConnectionAborted
-            | IoError::NotConnected
-            | IoError::BrokenPipe
-            | IoError::TimedOut => true,
-            _ => false,
-        }
+                | IoError::ConnectionReset
+                | IoError::ConnectionAborted
+                | IoError::NotConnected
+                | IoError::BrokenPipe
+                | IoError::TimedOut
+        )
     }
 }
 
diff --git a/src/services/statefuls/persist/actor.rs b/src/services/statefuls/persist/actor.rs
index 40eb67db..97e2736f 100644
--- a/src/services/statefuls/persist/actor.rs
+++ b/src/services/statefuls/persist/actor.rs
@@ -23,16 +23,13 @@ impl PersistActor<Load> {
     }
 }
 
-impl<T> PersistActor<SavingProcessor<T>>
-where
-    T: TWriterFactory,
-{
+impl PersistActor<SavingProcessor> {
     pub async fn run(
         filepath: String,
         num_of_cache_actors: usize,
     ) -> anyhow::Result<Sender<SaveCommand>> {
         // * Propagate error to caller before sending it to the background
-        let file = T::create_writer(filepath).await?;
+        let file = tokio::fs::File::create_writer(filepath).await?;
 
         let processor = SavingProcessor::new(file, SaveMeta::new(num_of_cache_actors));
 
diff --git a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
index 4c0777a8..7b95c61a 100644
--- a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
+++ b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
@@ -10,13 +10,13 @@ use super::byte_encoder::{
     encode_metadata,
 };
 
-pub struct SavingProcessor<T: TWrite> {
-    pub(super) writer: T,
+pub struct SavingProcessor {
+    pub(super) writer: tokio::fs::File,
     pub(super) meta: SaveMeta,
 }
 
-impl<T: TWrite> SavingProcessor<T> {
-    pub fn new(file: T, meta: SaveMeta) -> Self {
+impl SavingProcessor {
+    pub fn new(file: tokio::fs::File, meta: SaveMeta) -> Self {
         Self { writer: file, meta }
     }
 


Test patch:
diff --git a/tests/common.rs b/tests/common.rs
index 58e9f9f4..c76dfe06 100644
--- a/tests/common.rs
+++ b/tests/common.rs
@@ -1,8 +1,8 @@
 use redis_starter_rust::services::cluster::actor::{ClusterActor, PeerAddr};
 use redis_starter_rust::services::config::actor::ConfigActor;
 use redis_starter_rust::services::config::manager::ConfigManager;
-use redis_starter_rust::services::query_io::QueryIO;
 use redis_starter_rust::services::interface::{TCancellationTokenFactory, TStream};
+use redis_starter_rust::services::query_io::QueryIO;
 use redis_starter_rust::{make_smart_pointer, StartUpFacade, TNotifyStartUp};
 use std::process::{Child, Command, Stdio};
 use std::sync::Arc;
@@ -245,7 +245,7 @@ pub async fn threeway_handshake_helper(
 }
 
 pub fn create_cluster_actor_with_peers(peers: Vec<String>) -> ClusterActor {
-    let mut cluster_actor = ClusterActor::new();
+    let mut cluster_actor = ClusterActor::default();
     for peer in peers {
         cluster_actor.peers.insert(PeerAddr(peer));
     }
diff --git a/tests/test_cancellation_token.rs b/tests/test_cancellation_token.rs
index 6ab566a9..03f3de48 100644
--- a/tests/test_cancellation_token.rs
+++ b/tests/test_cancellation_token.rs
@@ -14,7 +14,7 @@ use tokio::net::TcpStream;
 async fn test_cancellation_token() {
     // GIVEN
     let config = init_config_manager_with_free_port().await;
-    let cluster_actor = ClusterActor::new();
+    let cluster_actor = ClusterActor::default();
     let _ = start_test_server(TestCancellationTokenFactory, config.clone(), cluster_actor).await;
 
     let mut client_stream = TcpStream::connect(config.bind_addr()).await.unwrap();
diff --git a/tests/test_config_get_dir.rs b/tests/test_config_get_dir.rs
index 3fdd8e27..9f5366e0 100644
--- a/tests/test_config_get_dir.rs
+++ b/tests/test_config_get_dir.rs
@@ -17,7 +17,7 @@ async fn test_config_get_dir() {
     // GIVEN
     //TODO test config should be dynamically configured
     let config = init_config_manager_with_free_port().await;
-    let cluster_actor = ClusterActor::new();
+    let cluster_actor = ClusterActor::default();
     let _ = start_test_server(CancellationTokenFactory, config.clone(), cluster_actor).await;
 
     let mut client_stream = TcpStream::connect(config.bind_addr()).await.unwrap();
diff --git a/tests/test_handshake.rs b/tests/test_handshake.rs
index a865f820..2916025c 100644
--- a/tests/test_handshake.rs
+++ b/tests/test_handshake.rs
@@ -30,7 +30,7 @@ async fn test_master_threeway_handshake() {
     // ! `peer_bind_addr` is bind_addr dedicated for peer connections
     manager.port = find_free_port_in_range(6000, 6553).await.unwrap();
     let master_cluster_bind_addr = manager.peer_bind_addr();
-    let cluster_actor: ClusterActor = ClusterActor::new();
+    let cluster_actor: ClusterActor = ClusterActor::default();
 
     let _ = start_test_server(CancellationTokenFactory, manager, cluster_actor).await;
     let mut client_stream = TcpStream::connect(master_cluster_bind_addr).await.unwrap();
diff --git a/tests/test_heartbeat.rs b/tests/test_heartbeat.rs
index d4386da9..fcb52aec 100644
--- a/tests/test_heartbeat.rs
+++ b/tests/test_heartbeat.rs
@@ -49,7 +49,7 @@ async fn test_heartbeat() {
     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();
-    let _ = start_test_server(CancellationTokenFactory, manager, ClusterActor::new()).await;
+    let _ = start_test_server(CancellationTokenFactory, manager, ClusterActor::default()).await;
 
     // run the client bind stream on a random port so it can later get connection request from server
     let handler = receive_server_ping_from_replica_stream(master_cluster_bind_addr.clone()).await;
@@ -69,7 +69,7 @@ async fn test_heartbeat_sent_to_multiple_replicas() {
     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();
-    let _ = start_test_server(CancellationTokenFactory, manager, ClusterActor::new()).await;
+    let _ = start_test_server(CancellationTokenFactory, manager, ClusterActor::default()).await;
 
     // run the client bind stream on a random port so it can later get connection request from server
     let repl1_handler =
diff --git a/tests/test_heartbeat_broadcasting.rs b/tests/test_heartbeat_broadcasting.rs
index 53725b9c..159f8774 100644
--- a/tests/test_heartbeat_broadcasting.rs
+++ b/tests/test_heartbeat_broadcasting.rs
@@ -87,7 +87,7 @@ async fn test_heartbeat_sent_to_multiple_replicas() {
         config.replication.master_host = Some("localhost".to_string());
         let mut manager = ConfigManager::new(config);
         manager.port = connecting_replica_port;
-        let _ = start_test_server(CancellationTokenFactory, manager, ClusterActor::new()).await;
+        let _ = start_test_server(CancellationTokenFactory, manager, ClusterActor::default()).await;
     }
 
     // THEN
diff --git a/tests/test_keys.rs b/tests/test_keys.rs
index 6d12eed7..a1cb3a55 100644
--- a/tests/test_keys.rs
+++ b/tests/test_keys.rs
@@ -15,7 +15,7 @@ async fn test_keys() {
     let _ = start_test_server(
         CancellationTokenFactory,
         config.clone(),
-        ClusterActor::new(),
+        ClusterActor::default(),
     )
     .await;
 
diff --git a/tests/test_replication_info.rs b/tests/test_replication_info.rs
index 5fbe0763..a460e4cb 100644
--- a/tests/test_replication_info.rs
+++ b/tests/test_replication_info.rs
@@ -21,7 +21,7 @@ async fn test_replication_info() {
     start_test_server(
         CancellationTokenFactory,
         config.clone(),
-        ClusterActor::new(),
+        ClusterActor::default(),
     )
     .await;
 
diff --git a/tests/test_save_read_dump.rs b/tests/test_save_read_dump.rs
index 626999a0..1d6f10bc 100644
--- a/tests/test_save_read_dump.rs
+++ b/tests/test_save_read_dump.rs
@@ -42,7 +42,7 @@ async fn test_save_read_dump() {
     let _ = start_test_server(
         CancellationTokenFactory,
         config.clone(),
-        ClusterActor::new(),
+        ClusterActor::default(),
     )
     .await;
 
@@ -73,7 +73,7 @@ async fn test_save_read_dump() {
     let _ = start_test_server(
         CancellationTokenFactory,
         config.clone(),
-        ClusterActor::new(),
+        ClusterActor::default(),
     )
     .await;
 
diff --git a/tests/test_set_get.rs b/tests/test_set_get.rs
index 20717453..3fcd6e7e 100644
--- a/tests/test_set_get.rs
+++ b/tests/test_set_get.rs
@@ -21,7 +21,7 @@ async fn test_set_get() {
     let _ = start_test_server(
         CancellationTokenFactory,
         config.clone(),
-        ClusterActor::new(),
+        ClusterActor::default(),
     )
     .await;