Repository: Migorithm/duva
Pull request #133 (2025-01-05 22:29:44)

refactor: service architecture


Patch:
diff --git a/src/adapters/cancellation_token.rs b/src/adapters/cancellation_token.rs
index 318dbb21..6f6cca4e 100644
--- a/src/adapters/cancellation_token.rs
+++ b/src/adapters/cancellation_token.rs
@@ -1,4 +1,4 @@
-use crate::services::stream_manager::interface::{
+use crate::services::interface::{
     TCancellationNotifier, TCancellationTokenFactory, TCancellationWatcher,
 };
 
diff --git a/src/adapters/io/tokio_file.rs b/src/adapters/io/tokio_file.rs
index 5b6a919e..917fda0e 100644
--- a/src/adapters/io/tokio_file.rs
+++ b/src/adapters/io/tokio_file.rs
@@ -1,4 +1,4 @@
-use crate::services::stream_manager::{
+use crate::services::{
     error::IoError,
     interface::{TWrite, TWriterFactory},
 };
diff --git a/src/adapters/io/tokio_stream.rs b/src/adapters/io/tokio_stream.rs
index 91f3d6f0..09796ffd 100644
--- a/src/adapters/io/tokio_stream.rs
+++ b/src/adapters/io/tokio_stream.rs
@@ -1,9 +1,6 @@
-use crate::services::stream_manager::interface::TWrite;
-use crate::services::stream_manager::{
-    error::IoError,
-    interface::{TGetPeerIp, TRead, TStream},
-    query_io::{parse, QueryIO},
-};
+use crate::services::error::IoError;
+use crate::services::interface::{TGetPeerIp, TRead, TStream, TWrite};
+use crate::services::query_io::{parse, QueryIO};
 use bytes::BytesMut;
 use std::io::ErrorKind;
 use tokio::net::tcp::OwnedWriteHalf;
diff --git a/src/lib.rs b/src/lib.rs
index 955e7c77..08060b15 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -4,17 +4,18 @@ pub mod services;
 use crate::adapters::io::tokio_stream::TokioStreamListenerFactory;
 use anyhow::Result;
 use services::cluster::actor::ClusterActor;
-use services::cluster::inbound_mode::InboundStream;
+use services::cluster::inbound_stream::InboundStream;
 use services::cluster::manager::ClusterManager;
-use services::cluster::outbound_mode::OutboundStream;
+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 services::stream_manager::error::IoError;
-use services::stream_manager::interface::TCancellationTokenFactory;
-use services::stream_manager::request_controller::client::ClientRequestController;
 use std::time::Duration;
 use tokio::net::TcpStream;
 
@@ -62,7 +63,6 @@ where
         }
     }
 
-    // TODO: remove input config and use config manager
     pub async fn run(&mut self, startup_notifier: impl TNotifyStartUp) -> Result<()> {
         if let Some(filepath) = self.config_manager.try_filepath().await? {
             let dump = PersistActor::dump(filepath).await?;
diff --git a/src/services/stream_manager/request_controller/client/arguments.rs b/src/services/client_request_controller/arguments.rs
similarity index 97%
rename from src/services/stream_manager/request_controller/client/arguments.rs
rename to src/services/client_request_controller/arguments.rs
index 4509337a..7553af10 100644
--- a/src/services/stream_manager/request_controller/client/arguments.rs
+++ b/src/services/client_request_controller/arguments.rs
@@ -1,6 +1,6 @@
 use crate::make_smart_pointer;
+use crate::services::query_io::QueryIO;
 use crate::services::statefuls::cache::CacheEntry;
-use crate::services::stream_manager::query_io::QueryIO;
 use anyhow::Result;
 
 #[derive(Debug, Clone)]
diff --git a/src/services/stream_manager/request_controller/client/mod.rs b/src/services/client_request_controller/mod.rs
similarity index 93%
rename from src/services/stream_manager/request_controller/client/mod.rs
rename to src/services/client_request_controller/mod.rs
index 7dbc071d..576fa572 100644
--- a/src/services/stream_manager/request_controller/client/mod.rs
+++ b/src/services/client_request_controller/mod.rs
@@ -1,21 +1,23 @@
 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::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 crate::services::stream_manager::interface::{TCancellationTokenFactory, TCancellationWatcher};
-use crate::services::stream_manager::query_io::QueryIO;
-use crate::services::stream_manager::{ClientStream, ClientStreamManager};
+
 use arguments::ClientRequestArguments;
-use client_request::ClientRequest;
+use request::ClientRequest;
+use stream_manager::{ClientStream, ClientStreamManager};
 use tokio::select;
 
-use crate::services::stream_manager::interface::TWriterFactory;
-
 pub mod arguments;
-pub mod client_request;
+pub mod request;
+pub mod stream_manager;
 
 pub(crate) struct ClientRequestController {
     config_manager: ConfigManager,
diff --git a/src/services/stream_manager/request_controller/client/client_request.rs b/src/services/client_request_controller/request.rs
similarity index 100%
rename from src/services/stream_manager/request_controller/client/client_request.rs
rename to src/services/client_request_controller/request.rs
diff --git a/src/services/stream_manager/mod.rs b/src/services/client_request_controller/stream_manager.rs
similarity index 83%
rename from src/services/stream_manager/mod.rs
rename to src/services/client_request_controller/stream_manager.rs
index e054d41c..fe835978 100644
--- a/src/services/stream_manager/mod.rs
+++ b/src/services/client_request_controller/stream_manager.rs
@@ -1,21 +1,19 @@
-pub mod error;
-pub mod interface;
-pub mod query_io;
-pub(crate) mod request_controller;
+use crate::{
+    make_smart_pointer,
+    services::{
+        error::IoError,
+        interface::{TCancellationNotifier, TCancellationTokenFactory, TStream},
+        query_io::QueryIO,
+    },
+};
 use anyhow::Context;
-use error::IoError;
-use interface::TCancellationNotifier;
-use interface::TCancellationTokenFactory;
 
-use interface::TStream;
-use interface::TWriterFactory;
-use query_io::QueryIO;
-use request_controller::client::arguments::ClientRequestArguments;
-use request_controller::client::client_request::ClientRequest;
-use request_controller::client::ClientRequestController;
 use tokio::net::TcpStream;
 
-use crate::make_smart_pointer;
+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> {
diff --git a/src/services/cluster/actor.rs b/src/services/cluster/actor.rs
index b3d3ac67..f28c9f15 100644
--- a/src/services/cluster/actor.rs
+++ b/src/services/cluster/actor.rs
@@ -1,7 +1,8 @@
 use crate::make_smart_pointer;
 use crate::services::cluster::command::ClusterCommand;
-use crate::services::stream_manager::interface::TWrite;
-use crate::services::stream_manager::query_io::QueryIO;
+
+use crate::services::interface::TWrite;
+use crate::services::query_io::QueryIO;
 use std::collections::{BTreeMap, HashSet};
 use std::time::Duration;
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
diff --git a/src/services/cluster/establishment/arguments.rs b/src/services/cluster/establishment/arguments.rs
new file mode 100644
index 00000000..5c0a64de
--- /dev/null
+++ b/src/services/cluster/establishment/arguments.rs
@@ -0,0 +1,13 @@
+use crate::make_smart_pointer;
+use crate::services::query_io::QueryIO;
+
+#[derive(Debug, Clone)]
+pub struct QueryArguments(pub Vec<QueryIO>);
+
+make_smart_pointer!(QueryArguments, Vec<QueryIO>);
+
+impl QueryArguments {
+    pub fn new(values: Vec<QueryIO>) -> Self {
+        Self(values)
+    }
+}
diff --git a/src/services/cluster/establishment/inbound.rs b/src/services/cluster/establishment/inbound.rs
new file mode 100644
index 00000000..c6c1eb8d
--- /dev/null
+++ b/src/services/cluster/establishment/inbound.rs
@@ -0,0 +1,115 @@
+use super::arguments::QueryArguments;
+use crate::services::query_io::QueryIO;
+use anyhow::Context;
+
+pub(crate) struct HandShakeRequest {
+    pub(crate) command: HandShakeRequestEnum,
+    pub(crate) args: QueryArguments,
+}
+
+impl HandShakeRequest {
+    pub(crate) fn new(query_io: QueryIO) -> anyhow::Result<Self> {
+        match query_io {
+            QueryIO::Array(value_array) => Ok(Self {
+                command: value_array
+                    .first()
+                    .context("request not given")?
+                    .clone()
+                    .unpack_bulk_str()?
+                    .try_into()?,
+                args: QueryArguments::new(value_array.into_iter().skip(1).collect()),
+            }),
+            _ => Err(anyhow::anyhow!("Unexpected command format")),
+        }
+    }
+
+    pub(crate) fn match_query(&self, request: HandShakeRequestEnum) -> anyhow::Result<()> {
+        if self.command == request {
+            Ok(())
+        } else {
+            Err(anyhow::anyhow!("{:?} not given during handshake", request))
+        }
+    }
+
+    pub(crate) fn extract_listening_port(&mut self) -> anyhow::Result<u16> {
+        self.match_query(HandShakeRequestEnum::ReplConf)?;
+
+        if self.args.len() < 2 {
+            return Err(anyhow::anyhow!("Not enough arguments"));
+        }
+
+        match self.args.as_mut_slice() {
+            [QueryIO::BulkString(key), QueryIO::BulkString(port), ..]
+                if key == "listening-port" =>
+            {
+                Ok(port.parse::<u16>()?)
+            }
+            _ => Err(anyhow::anyhow!("Invalid listening-port arguments")),
+        }
+    }
+
+    pub(crate) fn extract_capa(&mut self) -> anyhow::Result<Vec<(String, String)>> {
+        self.match_query(HandShakeRequestEnum::ReplConf)?;
+        if self.args.is_empty() || self.args.len() % 2 != 0 {
+            return Err(anyhow::anyhow!("Invalid number of arguments"));
+        }
+
+        // Process pairs directly using chunks_exact
+        let capabilities: Vec<(String, String)> = self
+            .args
+            .chunks_exact(2)
+            .filter_map(|chunk| match (&chunk[0], &chunk[1]) {
+                (QueryIO::BulkString(capa), QueryIO::BulkString(value)) if capa == "capa" => {
+                    Some((capa.clone(), value.clone()))
+                }
+                _ => None,
+            })
+            .collect();
+
+        // Validate last capability is psync2
+        if capabilities.last().context("No capabilities given")?.1 != "psync2" {
+            return Err(anyhow::anyhow!(
+                "psync2 must be given as the last capability"
+            ));
+        }
+        Ok(capabilities)
+    }
+    pub(crate) fn extract_psync(&mut self) -> anyhow::Result<(String, i64)> {
+        self.match_query(HandShakeRequestEnum::Psync)?;
+
+        let replica_id = self
+            .args
+            .get(0)
+            .map(|v| v.clone().unpack_bulk_str())
+            .ok_or(anyhow::anyhow!("No replica id"))??;
+        let offset = self
+            .args
+            .get(1)
+            .map(|v| v.clone().unpack_bulk_str().map(|s| s.parse::<i64>()))
+            .ok_or(anyhow::anyhow!("No offset"))???;
+        Ok((replica_id, offset))
+    }
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum HandShakeRequestEnum {
+    Ping,
+    ReplConf,
+    Psync,
+}
+
+impl TryFrom<String> for HandShakeRequestEnum {
+    type Error = anyhow::Error;
+    fn try_from(value: String) -> Result<Self, Self::Error> {
+        match value.to_lowercase().as_str() {
+            "ping" => Ok(HandShakeRequestEnum::Ping),
+            "replconf" => Ok(HandShakeRequestEnum::ReplConf),
+            "psync" => Ok(HandShakeRequestEnum::Psync),
+
+            invalid_value => {
+                eprintln!("Invalid command,{}", invalid_value);
+                Err(anyhow::anyhow!("Invalid command"))
+            }
+        }
+    }
+}
diff --git a/src/services/cluster/establishment/mod.rs b/src/services/cluster/establishment/mod.rs
new file mode 100644
index 00000000..aab2f6f8
--- /dev/null
+++ b/src/services/cluster/establishment/mod.rs
@@ -0,0 +1,4 @@
+pub mod outbound;
+
+pub mod arguments;
+pub mod inbound;
diff --git a/src/services/stream_manager/request_controller/replica/replication_request.rs b/src/services/cluster/establishment/outbound.rs
similarity index 55%
rename from src/services/stream_manager/request_controller/replica/replication_request.rs
rename to src/services/cluster/establishment/outbound.rs
index 609d0fd3..cf06c45f 100644
--- a/src/services/stream_manager/request_controller/replica/replication_request.rs
+++ b/src/services/cluster/establishment/outbound.rs
@@ -1,38 +1,5 @@
 use anyhow::Context;
 
-pub enum ReplicationRequest {}
-
-pub enum HandShakeRequest {
-    Ping,
-    ReplConf,
-    Psync,
-}
-
-impl TryFrom<String> for ReplicationRequest {
-    type Error = anyhow::Error;
-    fn try_from(value: String) -> Result<Self, Self::Error> {
-        match value.to_lowercase().as_str() {
-            _ => Err(anyhow::anyhow!("Invalid command")),
-        }
-    }
-}
-
-impl TryFrom<String> for HandShakeRequest {
-    type Error = anyhow::Error;
-    fn try_from(value: String) -> Result<Self, Self::Error> {
-        match value.to_lowercase().as_str() {
-            "ping" => Ok(HandShakeRequest::Ping),
-            "replconf" => Ok(HandShakeRequest::ReplConf),
-            "psync" => Ok(HandShakeRequest::Psync),
-
-            invalid_value => {
-                eprintln!("Invalid command,{}", invalid_value);
-                Err(anyhow::anyhow!("Invalid command"))
-            }
-        }
-    }
-}
-
 pub enum HandShakeResponse {
     PONG,
     OK,
diff --git a/src/services/cluster/inbound_mode.rs b/src/services/cluster/inbound_mode.rs
deleted file mode 100644
index 9561e4b4..00000000
--- a/src/services/cluster/inbound_mode.rs
+++ /dev/null
@@ -1,102 +0,0 @@
-use super::actor::PeerAddr;
-use crate::make_smart_pointer;
-use crate::services::stream_manager::interface::{TGetPeerIp, TStream};
-use crate::services::stream_manager::query_io::QueryIO;
-use crate::services::stream_manager::request_controller::replica::arguments::QueryArguments;
-use crate::services::stream_manager::request_controller::replica::replication_request::HandShakeRequest;
-use anyhow::Context;
-use tokio::net::TcpStream;
-use tokio::task::yield_now;
-
-// The following is used only when the node is in master mode
-pub(crate) struct InboundStream(pub TcpStream);
-
-make_smart_pointer!(InboundStream, TcpStream);
-
-impl InboundStream {
-    pub async fn recv_threeway_handshake(&mut self) -> anyhow::Result<(PeerAddr, bool)> {
-        self.recv_ping().await?;
-
-        let port = self.recv_replconf_listening_port().await?;
-
-        // TODO find use of capa?
-        let _capa_val_vec = self.recv_replconf_capa().await?;
-
-        // TODO check repl_id is '?' or of mine. If not, consider incoming as peer
-        let (_repl_id, _offset) = self.recv_psync().await?;
-
-        // ! TODO: STRANGE BEHAVIOUR
-        // if not for the following, message is sent with the previosly sent message
-        // even with this, it shows flaking behaviour
-        yield_now().await;
-
-        Ok((
-            PeerAddr(format!("{}:{}", self.get_peer_ip()?, port)),
-            _repl_id == "?", // if repl_id is '?' or of mine, it's slave, otherwise it's a peer.
-        ))
-    }
-
-    async fn recv_ping(&mut self) -> anyhow::Result<()> {
-        let (HandShakeRequest::Ping, _) = self.extract_query().await? else {
-            return Err(anyhow::anyhow!("Ping not given during handshake"));
-        };
-        self.write(QueryIO::SimpleString("PONG".to_string()))
-            .await?;
-        Ok(())
-    }
-
-    async fn recv_replconf_listening_port(&mut self) -> anyhow::Result<u16> {
-        let (HandShakeRequest::ReplConf, query_args) = self.extract_query().await? else {
-            return Err(anyhow::anyhow!("ReplConf not given during handshake"));
-        };
-        let port = if query_args.first() == Some(&QueryIO::BulkString("listening-port".to_string()))
-        {
-            query_args.take_replica_port()?
-        } else {
-            return Err(anyhow::anyhow!("Invalid listening-port given"));
-        };
-        self.write(QueryIO::SimpleString("OK".to_string())).await?;
-
-        Ok(port.parse::<u16>()?)
-    }
-
-    async fn recv_replconf_capa(&mut self) -> anyhow::Result<Vec<(String, String)>> {
-        let (HandShakeRequest::ReplConf, query_args) = self.extract_query().await? else {
-            return Err(anyhow::anyhow!("ReplConf not given during handshake"));
-        };
-        let capa_val_vec = query_args.take_capabilities()?;
-        self.write(QueryIO::SimpleString("OK".to_string())).await?;
-
-        Ok(capa_val_vec)
-    }
-    async fn recv_psync(&mut self) -> anyhow::Result<(String, i64)> {
-        let (HandShakeRequest::Psync, query_args) = self.extract_query().await? else {
-            return Err(anyhow::anyhow!("Psync not given during handshake"));
-        };
-        let (repl_id, offset) = query_args.take_psync()?;
-
-        self.write(QueryIO::SimpleString(
-            "FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0".to_string(),
-        ))
-        .await?;
-
-        Ok((repl_id, offset))
-    }
-
-    async fn extract_query(&mut self) -> anyhow::Result<(HandShakeRequest, QueryArguments)> {
-        let query_io = self.read_value().await?;
-        match query_io {
-            // TODO refactor
-            QueryIO::Array(value_array) => Ok((
-                value_array
-                    .first()
-                    .context("request not given")?
-                    .clone()
-                    .unpack_bulk_str()?
-                    .try_into()?,
-                QueryArguments::new(value_array.into_iter().skip(1).collect()),
-            )),
-            _ => Err(anyhow::anyhow!("Unexpected command format")),
-        }
-    }
-}
diff --git a/src/services/cluster/inbound_stream.rs b/src/services/cluster/inbound_stream.rs
new file mode 100644
index 00000000..603ed44e
--- /dev/null
+++ b/src/services/cluster/inbound_stream.rs
@@ -0,0 +1,70 @@
+use super::actor::PeerAddr;
+use super::establishment::inbound::{HandShakeRequest, HandShakeRequestEnum};
+use crate::make_smart_pointer;
+
+use crate::services::query_io::QueryIO;
+use crate::services::interface::{TGetPeerIp, TStream};
+use tokio::net::TcpStream;
+
+// The following is used only when the node is in master mode
+pub(crate) struct InboundStream(pub TcpStream);
+
+make_smart_pointer!(InboundStream, TcpStream);
+
+impl InboundStream {
+    pub async fn recv_threeway_handshake(&mut self) -> anyhow::Result<(PeerAddr, bool)> {
+        self.recv_ping().await?;
+
+        let port = self.recv_replconf_listening_port().await?;
+
+        // TODO find use of capa?
+        let _capa_val_vec = self.recv_replconf_capa().await?;
+
+        // TODO check repl_id is '?' or of mine. If not, consider incoming as peer
+        let (_repl_id, _offset) = self.recv_psync().await?;
+
+        Ok((
+            PeerAddr(format!("{}:{}", self.get_peer_ip()?, port)),
+            _repl_id == "?", // if repl_id is '?' or of mine, it's slave, otherwise it's a peer.
+        ))
+    }
+
+    async fn recv_ping(&mut self) -> anyhow::Result<()> {
+        let cmd = self.extract_cmd().await?;
+        cmd.match_query(HandShakeRequestEnum::Ping)?;
+
+        self.write(QueryIO::SimpleString("PONG".to_string()))
+            .await?;
+        Ok(())
+    }
+
+    async fn recv_replconf_listening_port(&mut self) -> anyhow::Result<u16> {
+        let mut cmd = self.extract_cmd().await?;
+        let port = cmd.extract_listening_port()?;
+        self.write(QueryIO::SimpleString("OK".to_string())).await?;
+        Ok(port)
+    }
+
+    async fn recv_replconf_capa(&mut self) -> anyhow::Result<Vec<(String, String)>> {
+        let mut cmd = self.extract_cmd().await?;
+        let capa_val_vec = cmd.extract_capa()?;
+        self.write(QueryIO::SimpleString("OK".to_string())).await?;
+        Ok(capa_val_vec)
+    }
+    async fn recv_psync(&mut self) -> anyhow::Result<(String, i64)> {
+        let mut cmd = self.extract_cmd().await?;
+        let (repl_id, offset) = cmd.extract_psync()?;
+
+        self.write(QueryIO::SimpleString(
+            "FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0".to_string(),
+        ))
+        .await?;
+
+        Ok((repl_id, offset))
+    }
+
+    async fn extract_cmd(&mut self) -> anyhow::Result<HandShakeRequest> {
+        let query_io = self.read_value().await?;
+        HandShakeRequest::new(query_io)
+    }
+}
diff --git a/src/services/cluster/manager.rs b/src/services/cluster/manager.rs
index e2a0f211..2c7d1fd6 100644
--- a/src/services/cluster/manager.rs
+++ b/src/services/cluster/manager.rs
@@ -2,13 +2,13 @@ use crate::make_smart_pointer;
 use crate::services::cluster::actor::{ClusterActor, PeerAddr};
 use crate::services::cluster::command::ClusterCommand;
 use crate::services::config::replication::Replication;
-use crate::services::stream_manager::interface::TStream;
-use crate::services::stream_manager::query_io::QueryIO;
+use crate::services::interface::TStream;
+use crate::services::query_io::QueryIO;
 use tokio::net::TcpStream;
 use tokio::sync::mpsc::Sender;
 
-use super::inbound_mode::InboundStream;
-use super::outbound_mode::OutboundStream;
+use super::inbound_stream::InboundStream;
+use super::outbound_stream::OutboundStream;
 
 #[derive(Clone)]
 pub struct ClusterManager(Sender<ClusterCommand>);
@@ -35,7 +35,6 @@ impl ClusterManager {
         self.disseminate_peers(&mut peer_stream).await.unwrap();
 
         // TODO At this point again, slave tries to connect to other nodes as peer in the cluster
-
         self.send(ClusterCommand::AddPeer {
             peer_addr,
             stream: peer_stream.0,
@@ -43,14 +42,6 @@ impl ClusterManager {
         })
         .await
         .unwrap();
-
-        // TODO Bidirectional communication should be established
-        // self.schedule_heartbeat(peer_addr).await.unwrap();
-
-        // TODO Following should be moved to actor and infinite loop will not be needed for replication
-        // self.handle_replication_request(&mut peer_stream)
-        //     .await
-        //     .unwrap();
     }
 
     async fn disseminate_peers(&self, stream: &mut TcpStream) -> anyhow::Result<()> {
diff --git a/src/services/cluster/mod.rs b/src/services/cluster/mod.rs
index d3dcf858..f11a0d94 100644
--- a/src/services/cluster/mod.rs
+++ b/src/services/cluster/mod.rs
@@ -1,5 +1,6 @@
 pub mod actor;
 mod command;
-pub mod inbound_mode;
+pub mod establishment;
+pub mod inbound_stream;
 pub mod manager;
-pub mod outbound_mode;
+pub mod outbound_stream;
diff --git a/src/services/cluster/outbound_mode.rs b/src/services/cluster/outbound_stream.rs
similarity index 92%
rename from src/services/cluster/outbound_mode.rs
rename to src/services/cluster/outbound_stream.rs
index fefea42a..8c4d8b61 100644
--- a/src/services/cluster/outbound_mode.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::stream_manager::interface::TStream;
-use crate::services::stream_manager::query_io::QueryIO;
-use crate::services::stream_manager::request_controller::replica::replication_request::HandShakeResponse;
+use crate::services::query_io::QueryIO;
+use crate::services::interface::TStream;
 use crate::{make_smart_pointer, write_array};
 use tokio::net::TcpStream;
 
diff --git a/src/services/config/manager.rs b/src/services/config/manager.rs
index 17cd506f..700ec1ec 100644
--- a/src/services/config/manager.rs
+++ b/src/services/config/manager.rs
@@ -76,11 +76,11 @@ impl ConfigManager {
             } // Not given a dbfilename
         }
     }
-    //TODO this is immutable! Should be moved to manager
+
     pub fn bind_addr(&self) -> String {
         format!("{}:{}", self.host, self.port)
     }
-    //TODO this is immutable! Should be moved to manager
+
     pub fn peer_bind_addr(&self) -> String {
         format!("{}:{}", self.host, self.port + 10000)
     }
diff --git a/src/services/stream_manager/error.rs b/src/services/error.rs
similarity index 100%
rename from src/services/stream_manager/error.rs
rename to src/services/error.rs
diff --git a/src/services/stream_manager/interface.rs b/src/services/interface.rs
similarity index 95%
rename from src/services/stream_manager/interface.rs
rename to src/services/interface.rs
index 96663afc..d3e39715 100644
--- a/src/services/stream_manager/interface.rs
+++ b/src/services/interface.rs
@@ -1,6 +1,9 @@
-use super::{error::IoError, query_io::QueryIO};
+use crate::services::error::IoError;
+
 use bytes::BytesMut;
 
+use super::query_io::QueryIO;
+
 pub trait TStream: TGetPeerIp + Send + Sync + 'static {
     // TODO deprecated
     fn read_value(&mut self) -> impl std::future::Future<Output = anyhow::Result<QueryIO>> + Send;
diff --git a/src/services/interfaces/mod.rs b/src/services/interfaces/mod.rs
deleted file mode 100644
index 0f4a360f..00000000
--- a/src/services/interfaces/mod.rs
+++ /dev/null
@@ -1,3 +0,0 @@
-// marker trait for types that are thread-safe and cloneable
-pub trait ThreadSafeCloneable: 'static + Send + Sync + Clone {}
-impl<T: 'static + Send + Sync + Clone> ThreadSafeCloneable for T {}
diff --git a/src/services/mod.rs b/src/services/mod.rs
index cf4b4508..efa35c59 100644
--- a/src/services/mod.rs
+++ b/src/services/mod.rs
@@ -1,5 +1,7 @@
+pub mod client_request_controller;
 pub mod cluster;
 pub mod config;
-pub mod interfaces;
+pub mod error;
+pub mod interface;
+pub mod query_io;
 pub mod statefuls;
-pub mod stream_manager;
diff --git a/src/services/stream_manager/query_io.rs b/src/services/query_io.rs
similarity index 99%
rename from src/services/stream_manager/query_io.rs
rename to src/services/query_io.rs
index b9bb1283..684fe666 100644
--- a/src/services/stream_manager/query_io.rs
+++ b/src/services/query_io.rs
@@ -70,7 +70,7 @@ impl From<String> for QueryIO {
         QueryIO::BulkString(v)
     }
 }
-
+//
 pub fn parse(buffer: BytesMut) -> Result<(QueryIO, usize)> {
     match buffer[0] as char {
         '+' => parse_simple_string(buffer),
diff --git a/src/services/statefuls/cache/actor.rs b/src/services/statefuls/cache/actor.rs
index 5da1291d..6520b16e 100644
--- a/src/services/statefuls/cache/actor.rs
+++ b/src/services/statefuls/cache/actor.rs
@@ -1,9 +1,9 @@
-use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use super::CacheEntry;
 use super::CacheValue;
 use crate::make_smart_pointer;
+use crate::services::query_io::QueryIO;
+use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 use crate::services::statefuls::persist::save_command::SaveCommand;
-use crate::services::stream_manager::query_io::QueryIO;
 use anyhow::Context;
 use anyhow::Result;
 use std::collections::HashMap;
diff --git a/src/services/statefuls/cache/manager.rs b/src/services/statefuls/cache/manager.rs
index 20c6ca6b..0719f1af 100644
--- a/src/services/statefuls/cache/manager.rs
+++ b/src/services/statefuls/cache/manager.rs
@@ -1,8 +1,9 @@
 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::query_io::QueryIO;
+use crate::services::statefuls::cache::ttl::manager::TtlSchedulerInbox;
 
+use crate::services::statefuls::cache::ttl::actor::TtlActor;
 use crate::services::statefuls::persist::save_command::SaveCommand;
 use crate::services::statefuls::persist::DumpFile;
 use anyhow::Result;
@@ -10,7 +11,6 @@ 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/persist/actor.rs b/src/services/statefuls/persist/actor.rs
index f2c9cde1..40eb67db 100644
--- a/src/services/statefuls/persist/actor.rs
+++ b/src/services/statefuls/persist/actor.rs
@@ -3,8 +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::interface::TWriterFactory;
 use crate::services::statefuls::persist::save_command::SaveCommand;
-use crate::services::stream_manager::interface::TWriterFactory;
 use tokio::sync::mpsc::{Receiver, Sender};
 
 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 6408eac5..4c0777a8 100644
--- a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
+++ b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
@@ -1,5 +1,4 @@
-use crate::services::statefuls::cache::CacheEntry;
-use crate::services::stream_manager::interface::TWrite;
+use crate::services::{interface::TWrite, statefuls::cache::CacheEntry};
 
 use crate::services::statefuls::persist::save_command::SaveCommand;
 use anyhow::Result;
diff --git a/src/services/statefuls/persist/mod.rs b/src/services/statefuls/persist/mod.rs
index 6fa73ba5..243836eb 100644
--- a/src/services/statefuls/persist/mod.rs
+++ b/src/services/statefuls/persist/mod.rs
@@ -1,10 +1,8 @@
-pub mod endec;
 pub mod actor;
+pub mod endec;
 pub mod save_command;
-
-use std::collections::HashMap;
-
 use super::cache::CacheEntry;
+use std::collections::HashMap;
 
 #[derive(Debug)]
 pub struct DumpFile {
diff --git a/src/services/stream_manager/request_controller/mod.rs b/src/services/stream_manager/request_controller/mod.rs
deleted file mode 100644
index 842e582b..00000000
--- a/src/services/stream_manager/request_controller/mod.rs
+++ /dev/null
@@ -1,2 +0,0 @@
-pub mod client;
-pub mod replica;
\ No newline at end of file
diff --git a/src/services/stream_manager/request_controller/replica/arguments.rs b/src/services/stream_manager/request_controller/replica/arguments.rs
deleted file mode 100644
index c1513e8e..00000000
--- a/src/services/stream_manager/request_controller/replica/arguments.rs
+++ /dev/null
@@ -1,58 +0,0 @@
-use crate::make_smart_pointer;
-use crate::services::stream_manager::query_io::QueryIO;
-
-#[derive(Debug, Clone)]
-pub struct QueryArguments(pub Vec<QueryIO>);
-
-make_smart_pointer!(QueryArguments, Vec<QueryIO>);
-
-impl QueryArguments {
-    pub fn new(values: Vec<QueryIO>) -> Self {
-        Self(values)
-    }
-
-    // take replica port info
-    pub(crate) fn take_replica_port(self) -> anyhow::Result<String> {
-        let port = self.get(1).ok_or(anyhow::anyhow!("No value"))?;
-
-        let QueryIO::BulkString(port) = port else {
-            return Err(anyhow::anyhow!("Invalid arguments"));
-        };
-        Ok(port.to_string())
-    }
-    pub(crate) fn take_capabilities(self) -> anyhow::Result<Vec<(String, String)>> {
-        let mut capabilities = Vec::new();
-        for i in 0..self.len() {
-            if i % 2 == 1 {
-                continue;
-            }
-            let capa = self.get(i).ok_or(anyhow::anyhow!("No value"))?;
-            let QueryIO::BulkString(capa) = capa else {
-                return Err(anyhow::anyhow!("Invalid arguments"));
-            };
-            if capa != "capa" {
-                return Err(anyhow::anyhow!("Invalid arguments"));
-            }
-            let Some(QueryIO::BulkString(value)) = self.get(i + 1) else {
-                return Err(anyhow::anyhow!("Invalid arguments"));
-            };
-            // last value must be psync2
-            if i == self.len() - 2 && value != "psync2" {
-                return Err(anyhow::anyhow!("Invalid arguments"));
-            }
-            capabilities.push((capa.to_string(), value.to_string()));
-        }
-        Ok(capabilities)
-    }
-    pub(crate) fn take_psync(self) -> anyhow::Result<(String, i64)> {
-        let replica_id = self
-            .get(0)
-            .map(|v| v.clone().unpack_bulk_str())
-            .ok_or(anyhow::anyhow!("No replica id"))??;
-        let offset = self
-            .get(1)
-            .map(|v| v.clone().unpack_bulk_str().map(|s| s.parse::<i64>()))
-            .ok_or(anyhow::anyhow!("No offset"))???;
-        Ok((replica_id, offset))
-    }
-}
diff --git a/src/services/stream_manager/request_controller/replica/mod.rs b/src/services/stream_manager/request_controller/replica/mod.rs
deleted file mode 100644
index 5da12b8c..00000000
--- a/src/services/stream_manager/request_controller/replica/mod.rs
+++ /dev/null
@@ -1,2 +0,0 @@
-pub mod arguments;
-pub mod replication_request;


Test patch:
diff --git a/tests/common.rs b/tests/common.rs
index 646ecee8..58e9f9f4 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::stream_manager::interface::{TCancellationTokenFactory, TStream};
-use redis_starter_rust::services::stream_manager::query_io::QueryIO;
+use redis_starter_rust::services::query_io::QueryIO;
+use redis_starter_rust::services::interface::{TCancellationTokenFactory, TStream};
 use redis_starter_rust::{make_smart_pointer, StartUpFacade, TNotifyStartUp};
 use std::process::{Child, Command, Stdio};
 use std::sync::Arc;
diff --git a/tests/test_cancellation_token.rs b/tests/test_cancellation_token.rs
index 43c4cbcc..6ab566a9 100644
--- a/tests/test_cancellation_token.rs
+++ b/tests/test_cancellation_token.rs
@@ -6,9 +6,7 @@ mod common;
 use common::{init_config_manager_with_free_port, start_test_server, TestStreamHandler};
 use redis_starter_rust::services::{
     cluster::actor::ClusterActor,
-    stream_manager::interface::{
-        TCancellationNotifier, TCancellationTokenFactory, TCancellationWatcher,
-    },
+    interface::{TCancellationNotifier, TCancellationTokenFactory, TCancellationWatcher},
 };
 use tokio::net::TcpStream;
 
diff --git a/tests/test_disseminate_peers.rs b/tests/test_disseminate_peers.rs
index c864979b..40d27dc3 100644
--- a/tests/test_disseminate_peers.rs
+++ b/tests/test_disseminate_peers.rs
@@ -8,7 +8,7 @@ use redis_starter_rust::{
     adapters::cancellation_token::CancellationTokenFactory,
     services::{
         config::{actor::ConfigActor, manager::ConfigManager},
-        stream_manager::interface::TStream,
+        interface::TStream,
     },
 };
 use tokio::net::TcpStream;
diff --git a/tests/test_handshake.rs b/tests/test_handshake.rs
index cf8195e4..a865f820 100644
--- a/tests/test_handshake.rs
+++ b/tests/test_handshake.rs
@@ -1,9 +1,4 @@
-use std::{
-    io::{BufRead, BufReader},
-    process::Stdio,
-    thread::sleep,
-    time::Duration,
-};
+use std::io::{BufRead, BufReader};
 
 /// Three-way handshake test
 /// 1. Client sends PING command
@@ -23,7 +18,7 @@ use redis_starter_rust::{
         config::{actor::ConfigActor, manager::ConfigManager},
     },
 };
-use tokio::{io::AsyncBufReadExt, net::TcpStream};
+use tokio::net::TcpStream;
 mod common;
 
 #[tokio::test]
diff --git a/tests/test_heartbeat.rs b/tests/test_heartbeat.rs
index 6a3a4390..d4386da9 100644
--- a/tests/test_heartbeat.rs
+++ b/tests/test_heartbeat.rs
@@ -14,7 +14,7 @@ use redis_starter_rust::{
     services::{
         cluster::actor::ClusterActor,
         config::{actor::ConfigActor, manager::ConfigManager},
-        stream_manager::interface::TStream,
+        interface::TStream,
     },
 };
 use tokio::{net::TcpStream, task::JoinHandle, time::timeout};
diff --git a/tests/test_heartbeat_broadcasting.rs b/tests/test_heartbeat_broadcasting.rs
index f2fb2ea0..53725b9c 100644
--- a/tests/test_heartbeat_broadcasting.rs
+++ b/tests/test_heartbeat_broadcasting.rs
@@ -8,7 +8,7 @@ use redis_starter_rust::{
     services::{
         cluster::actor::ClusterActor,
         config::{actor::ConfigActor, manager::ConfigManager},
-        stream_manager::interface::TStream,
+        interface::TStream,
     },
 };
 use std::collections::HashMap;