Repository: Migorithm/duva
Pull request #107 (2024-12-12 20:20:29)

feat: handshake enum added, temporal coupling specified
#95 #7 #5 


Patch:
diff --git a/src/adapters/cancellation_token.rs b/src/adapters/cancellation_token.rs
index e1cf2e76..318dbb21 100644
--- a/src/adapters/cancellation_token.rs
+++ b/src/adapters/cancellation_token.rs
@@ -1,4 +1,4 @@
-use crate::services::query_manager::interface::{
+use crate::services::stream_manager::interface::{
     TCancellationNotifier, TCancellationTokenFactory, TCancellationWatcher,
 };
 
diff --git a/src/adapters/io/tokio_file.rs b/src/adapters/io/tokio_file.rs
index fe09f088..811a8dbc 100644
--- a/src/adapters/io/tokio_file.rs
+++ b/src/adapters/io/tokio_file.rs
@@ -1,4 +1,4 @@
-use crate::services::query_manager::{
+use crate::services::stream_manager::{
     error::IoError,
     interface::{TWrite, TWriterFactory},
 };
diff --git a/src/adapters/io/tokio_stream.rs b/src/adapters/io/tokio_stream.rs
index 56cb498e..863a0ea6 100644
--- a/src/adapters/io/tokio_stream.rs
+++ b/src/adapters/io/tokio_stream.rs
@@ -1,7 +1,7 @@
 use std::io::ErrorKind;
 
 use crate::{
-    services::query_manager::{
+    services::stream_manager::{
         error::IoError,
         interface::{TConnectStreamFactory, TGetPeerIp, TRead, TStream, TWrite},
         PeerAddr,
diff --git a/src/lib.rs b/src/lib.rs
index 011bb006..421ee8cf 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,19 +1,19 @@
 pub mod adapters;
 pub mod macros;
 pub mod services;
-use crate::services::query_manager::client_request_controllers::ClientRequestController;
+use crate::services::stream_manager::client_request_controllers::ClientRequestController;
 use anyhow::Result;
 use services::config::config_actor::Config;
 use services::config::config_manager::ConfigManager;
 
-use services::query_manager::interface::{
-    TCancellationTokenFactory, TConnectStreamFactory, TStreamListener, TStreamListenerFactory,
-};
-use services::query_manager::replication_request_controllers::ReplicationRequestController;
-use services::query_manager::{self, QueryManager};
 use services::statefuls::cache::cache_manager::CacheManager;
 use services::statefuls::cache::ttl_manager::TtlSchedulerInbox;
 use services::statefuls::persist::persist_actor::PersistActor;
+use services::stream_manager::interface::{
+    TCancellationTokenFactory, TConnectStreamFactory, TStreamListener, TStreamListenerFactory,
+};
+use services::stream_manager::replication_request_controllers::ReplicationRequestController;
+use services::stream_manager::StreamManager;
 
 // * StartUp Facade that manages invokes subsystems
 pub struct StartUpFacade<T, U, V> {
@@ -92,7 +92,7 @@ where
                 match replication_listener.listen().await {
                     Ok((peer_stream, _)) => {
                         let query_manager =
-                            QueryManager::new(peer_stream, replication_request_controller);
+                            StreamManager::new(peer_stream, replication_request_controller);
 
                         tokio::spawn(query_manager.handle_peer_stream(connect_stream_factory));
                     }
@@ -121,7 +121,7 @@ where
                 Ok((stream, _)) =>
                 // Spawn a new task to handle the connection without blocking the main thread.
                 {
-                    let query_manager = QueryManager::new(stream, self.client_request_controller);
+                    let query_manager = StreamManager::new(stream, self.client_request_controller);
                     tokio::spawn(
                         query_manager.handle_single_client_stream::<tokio::fs::File>(
                             self.cancellation_factory,
diff --git a/src/services/config/config_actor.rs b/src/services/config/config_actor.rs
index 70cfa61f..f3a47c5c 100644
--- a/src/services/config/config_actor.rs
+++ b/src/services/config/config_actor.rs
@@ -88,10 +88,6 @@ impl Config {
     pub fn get_filepath(&self) -> String {
         format!("{}/{}", self.dir, self.dbfilename)
     }
-
-    pub(crate) fn is_replica(&self) -> bool {
-        self.replication.role() == "slave"
-    }
 }
 
 macro_rules! env_var {
diff --git a/src/services/mod.rs b/src/services/mod.rs
index f3026f6a..8e643427 100644
--- a/src/services/mod.rs
+++ b/src/services/mod.rs
@@ -1,4 +1,4 @@
 pub mod config;
 pub mod interfaces;
-pub mod query_manager;
 pub mod statefuls;
+pub mod stream_manager;
diff --git a/src/services/query_manager/replication_request_controllers/replication_request.rs b/src/services/query_manager/replication_request_controllers/replication_request.rs
deleted file mode 100644
index 86f45208..00000000
--- a/src/services/query_manager/replication_request_controllers/replication_request.rs
+++ /dev/null
@@ -1,18 +0,0 @@
-pub enum ReplicationRequest {
-    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() {
-            "ping" => Ok(ReplicationRequest::Ping),
-            "replconf" => Ok(ReplicationRequest::ReplConf),
-            "psync" => Ok(ReplicationRequest::Psync),
-
-            _ => Err(anyhow::anyhow!("Invalid command")),
-        }
-    }
-}
diff --git a/src/services/statefuls/cache/cache_actor.rs b/src/services/statefuls/cache/cache_actor.rs
index ed2124af..6641f04d 100644
--- a/src/services/statefuls/cache/cache_actor.rs
+++ b/src/services/statefuls/cache/cache_actor.rs
@@ -2,8 +2,8 @@ use super::ttl_manager::TtlSchedulerInbox;
 use super::CacheEntry;
 use super::CacheValue;
 use crate::make_smart_pointer;
-use crate::services::query_manager::query_io::QueryIO;
 use crate::services::statefuls::persist::persist_actor::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/cache_manager.rs b/src/services/statefuls/cache/cache_manager.rs
index 2688dec5..07b3f82c 100644
--- a/src/services/statefuls/cache/cache_manager.rs
+++ b/src/services/statefuls/cache/cache_manager.rs
@@ -1,7 +1,7 @@
 use super::cache_actor::{CacheActor, CacheCommand, CacheCommandSender};
 use super::ttl_manager::{TtlActor, TtlSchedulerInbox};
 use super::CacheEntry;
-use crate::services::query_manager::query_io::QueryIO;
+use crate::services::stream_manager::query_io::QueryIO;
 
 use crate::services::statefuls::persist::persist_actor::SaveCommand;
 use crate::services::statefuls::persist::DumpFile;
diff --git a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
index 3da37989..3b8ace14 100644
--- a/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
+++ b/src/services/statefuls/persist/endec/encoder/encoding_processor.rs
@@ -1,5 +1,5 @@
-use crate::services::query_manager::interface::TWrite;
 use crate::services::statefuls::cache::CacheEntry;
+use crate::services::stream_manager::interface::TWrite;
 
 use crate::services::statefuls::persist::persist_actor::SaveCommand;
 use anyhow::Result;
diff --git a/src/services/statefuls/persist/persist_actor.rs b/src/services/statefuls/persist/persist_actor.rs
index 00bb8908..83c6fb7d 100644
--- a/src/services/statefuls/persist/persist_actor.rs
+++ b/src/services/statefuls/persist/persist_actor.rs
@@ -3,8 +3,8 @@ 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::query_manager::interface::TWriterFactory;
 use crate::services::statefuls::cache::CacheEntry;
+use crate::services::stream_manager::interface::TWriterFactory;
 use tokio::sync::mpsc::{Receiver, Sender};
 
 pub enum SaveCommand {
diff --git a/src/services/query_manager/client_request_controllers/arguments.rs b/src/services/stream_manager/client_request_controllers/arguments.rs
similarity index 97%
rename from src/services/query_manager/client_request_controllers/arguments.rs
rename to src/services/stream_manager/client_request_controllers/arguments.rs
index 24b2a303..4509337a 100644
--- a/src/services/query_manager/client_request_controllers/arguments.rs
+++ b/src/services/stream_manager/client_request_controllers/arguments.rs
@@ -1,6 +1,6 @@
 use crate::make_smart_pointer;
-use crate::services::query_manager::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/query_manager/client_request_controllers/client_request.rs b/src/services/stream_manager/client_request_controllers/client_request.rs
similarity index 100%
rename from src/services/query_manager/client_request_controllers/client_request.rs
rename to src/services/stream_manager/client_request_controllers/client_request.rs
diff --git a/src/services/query_manager/client_request_controllers/mod.rs b/src/services/stream_manager/client_request_controllers/mod.rs
similarity index 96%
rename from src/services/query_manager/client_request_controllers/mod.rs
rename to src/services/stream_manager/client_request_controllers/mod.rs
index bccc8f1d..d1419f2f 100644
--- a/src/services/query_manager/client_request_controllers/mod.rs
+++ b/src/services/stream_manager/client_request_controllers/mod.rs
@@ -1,11 +1,11 @@
 use crate::services::config::config_manager::ConfigManager;
 use crate::services::config::ConfigResponse;
-use crate::services::query_manager::interface::TCancellationWatcher;
-use crate::services::query_manager::query_io::QueryIO;
 use crate::services::statefuls::cache::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::stream_manager::interface::TCancellationWatcher;
+use crate::services::stream_manager::query_io::QueryIO;
 use arguments::ClientRequestArguments;
 use client_request::ClientRequest;
 
diff --git a/src/services/query_manager/error.rs b/src/services/stream_manager/error.rs
similarity index 100%
rename from src/services/query_manager/error.rs
rename to src/services/stream_manager/error.rs
diff --git a/src/services/query_manager/interface.rs b/src/services/stream_manager/interface.rs
similarity index 100%
rename from src/services/query_manager/interface.rs
rename to src/services/stream_manager/interface.rs
diff --git a/src/services/query_manager/mod.rs b/src/services/stream_manager/mod.rs
similarity index 73%
rename from src/services/query_manager/mod.rs
rename to src/services/stream_manager/mod.rs
index 7a78d778..54b95ce1 100644
--- a/src/services/query_manager/mod.rs
+++ b/src/services/stream_manager/mod.rs
@@ -3,8 +3,8 @@ pub mod error;
 pub mod interface;
 pub mod query_io;
 pub mod replication_request_controllers;
-use crate::services::query_manager::client_request_controllers::client_request::ClientRequest;
-use crate::services::query_manager::client_request_controllers::ClientRequestController;
+use crate::services::stream_manager::client_request_controllers::client_request::ClientRequest;
+use crate::services::stream_manager::client_request_controllers::ClientRequestController;
 
 use anyhow::Context;
 use bytes::BytesMut;
@@ -17,12 +17,12 @@ use interface::{
 
 use query_io::QueryIO;
 use replication_request_controllers::{
-    arguments::ReplicationRequestArguments, replication_request::ReplicationRequest,
+    arguments::PeerRequestArguments, replication_request::HandShakeRequest,
     ReplicationRequestController,
 };
 
 /// Controller is a struct that will be used to read and write values to the client.
-pub struct QueryManager<T, U>
+pub struct StreamManager<T, U>
 where
     T: TStream,
 {
@@ -30,12 +30,12 @@ where
     pub(crate) controller: U,
 }
 
-impl<T, U> QueryManager<T, U>
+impl<T, U> StreamManager<T, U>
 where
     T: TStream,
 {
     pub(crate) fn new(stream: T, controller: U) -> Self {
-        QueryManager { stream, controller }
+        StreamManager { stream, controller }
     }
 
     // crlf
@@ -53,7 +53,7 @@ where
     }
 }
 
-impl<T> QueryManager<T, &'static ClientRequestController>
+impl<T> StreamManager<T, &'static ClientRequestController>
 where
     T: TStream,
 {
@@ -112,13 +112,15 @@ where
 }
 
 pub struct PeerAddr(pub String);
-impl<T> QueryManager<T, &'static ReplicationRequestController>
+impl<T> StreamManager<T, &'static ReplicationRequestController>
 where
     T: TStream,
 {
-    async fn extract_query(
-        &mut self,
-    ) -> anyhow::Result<(ReplicationRequest, ReplicationRequestArguments)> {
+    async fn extract_query<R>(&mut self) -> anyhow::Result<(R, PeerRequestArguments)>
+    where
+        R: TryFrom<String>,
+        anyhow::Error: From<R::Error>,
+    {
         let query_io = self.read_value().await?;
         match query_io {
             // TODO refactor
@@ -129,7 +131,7 @@ where
                     .clone()
                     .unpack_bulk_str()?
                     .try_into()?,
-                ReplicationRequestArguments::new(value_array.into_iter().skip(1).collect()),
+                PeerRequestArguments::new(value_array.into_iter().skip(1).collect()),
             )),
             _ => Err(anyhow::anyhow!("Unexpected command format")),
         }
@@ -139,16 +141,31 @@ where
             .await
     }
 
+    // Temporal coupling: each handling functions inside the following method must be in order
     async fn establish_threeway_handshake(&mut self) -> anyhow::Result<PeerAddr> {
-        let (ReplicationRequest::Ping, _) = self.extract_query().await? else {
+        self.handle_ping().await?;
+        let port = self.handle_replconf_listening_port().await?;
+
+        // TODO find use of capa?
+        let _capa_val_vec = self.handle_replconf_capa().await?;
+
+        // TODO find use of psync info?
+        let (_repl_id, _offset) = self.handle_psync().await?;
+
+        Ok(PeerAddr(format!("{}:{}", self.stream.get_peer_ip()?, port)))
+    }
+
+    async fn handle_ping(&mut self) -> anyhow::Result<()> {
+        let (HandShakeRequest::Ping, _) = self.extract_query().await? else {
             return Err(anyhow::anyhow!("Ping not given"));
         };
         self.send_simple_string("PONG").await?;
-
-        let (ReplicationRequest::ReplConf, query_args) = self.extract_query().await? else {
+        Ok(())
+    }
+    async fn handle_replconf_listening_port(&mut self) -> anyhow::Result<i16> {
+        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()?
@@ -157,37 +174,36 @@ where
         };
         self.send_simple_string("OK").await?;
 
-        let (ReplicationRequest::ReplConf, query_args) = self.extract_query().await? else {
+        Ok(port.parse::<i16>()?)
+    }
+
+    async fn handle_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"));
         };
-        // TODO find use of capa?
-        let _capa_val_vec = query_args.take_capabilities()?;
+        let capa_val_vec = query_args.take_capabilities()?;
         self.send_simple_string("OK").await?;
 
-        let (ReplicationRequest::Psync, query_args) = self.extract_query().await? else {
+        Ok(capa_val_vec)
+    }
+    async fn handle_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"));
         };
-
-        // TODO find use of psync info?
-        let (_repl_id, _offset) = query_args.take_psync()?;
+        let (repl_id, offset) = query_args.take_psync()?;
         self.send_simple_string("FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0")
             .await?;
 
-        self.get_peer_addr(port.parse::<i16>()?)
-    }
-
-    fn get_peer_addr(&self, port: i16) -> anyhow::Result<PeerAddr> {
-        Ok(PeerAddr(format!("{}:{}", self.stream.get_peer_ip()?, port)))
+        Ok((repl_id, offset))
     }
 
     pub(crate) async fn handle_peer_stream(
         mut self,
         connect_stream_factory: impl TConnectStreamFactory,
     ) -> anyhow::Result<()> {
-        // TODO Three way handshake
         let peer_addr = self.establish_threeway_handshake().await?;
 
-        // TODO Stream factory DI - p3
+        // TODO the following may be skipped if connection has already been made
         let out_stream = connect_stream_factory.connect(peer_addr).await?;
 
         // Following infinite loop may need to be changed once replica information is given
diff --git a/src/services/query_manager/query_io.rs b/src/services/stream_manager/query_io.rs
similarity index 100%
rename from src/services/query_manager/query_io.rs
rename to src/services/stream_manager/query_io.rs
diff --git a/src/services/query_manager/replication_request_controllers/arguments.rs b/src/services/stream_manager/replication_request_controllers/arguments.rs
similarity index 87%
rename from src/services/query_manager/replication_request_controllers/arguments.rs
rename to src/services/stream_manager/replication_request_controllers/arguments.rs
index ab59c2b6..be6a3f83 100644
--- a/src/services/query_manager/replication_request_controllers/arguments.rs
+++ b/src/services/stream_manager/replication_request_controllers/arguments.rs
@@ -1,12 +1,12 @@
 use crate::make_smart_pointer;
-use crate::services::query_manager::query_io::QueryIO;
+use crate::services::stream_manager::query_io::QueryIO;
 
 #[derive(Debug, Clone)]
-pub struct ReplicationRequestArguments(Vec<QueryIO>);
+pub struct PeerRequestArguments(Vec<QueryIO>);
 
-make_smart_pointer!(ReplicationRequestArguments, Vec<QueryIO>);
+make_smart_pointer!(PeerRequestArguments, Vec<QueryIO>);
 
-impl ReplicationRequestArguments {
+impl PeerRequestArguments {
     pub fn new(values: Vec<QueryIO>) -> Self {
         Self(values)
     }
@@ -15,7 +15,7 @@ impl ReplicationRequestArguments {
     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 {
+        let QueryIO::BulkString(port) = port else {
             return Err(anyhow::anyhow!("Invalid arguments"));
         };
         Ok(port.to_string())
diff --git a/src/services/query_manager/replication_request_controllers/mod.rs b/src/services/stream_manager/replication_request_controllers/mod.rs
similarity index 66%
rename from src/services/query_manager/replication_request_controllers/mod.rs
rename to src/services/stream_manager/replication_request_controllers/mod.rs
index 00c09050..c56b7283 100644
--- a/src/services/query_manager/replication_request_controllers/mod.rs
+++ b/src/services/stream_manager/replication_request_controllers/mod.rs
@@ -1,4 +1,4 @@
-use arguments::ReplicationRequestArguments;
+use arguments::PeerRequestArguments;
 use replication_request::ReplicationRequest;
 
 use crate::services::config::{config_manager::ConfigManager, ConfigCommand};
@@ -19,7 +19,7 @@ impl ReplicationRequestController {
     pub async fn handle(
         &self,
         request: ReplicationRequest,
-        args: ReplicationRequestArguments,
+        _args: PeerRequestArguments,
     ) -> anyhow::Result<QueryIO> {
         // handle replication request
         let response = match request {
@@ -30,16 +30,6 @@ impl ReplicationRequestController {
                     .await?;
                 QueryIO::SimpleString("PONG".to_string())
             }
-            ReplicationRequest::ReplConf => {
-                // args will be:
-                // "listening-port" "port" OR
-                // "capa" "eof" "capa" "psync2"
-                QueryIO::BulkString("OK".to_string())
-            }
-
-            ReplicationRequest::Psync => QueryIO::BulkString(
-                "FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0".to_string(),
-            ),
         };
         Ok(response)
     }
diff --git a/src/services/stream_manager/replication_request_controllers/replication_request.rs b/src/services/stream_manager/replication_request_controllers/replication_request.rs
new file mode 100644
index 00000000..1ea93f95
--- /dev/null
+++ b/src/services/stream_manager/replication_request_controllers/replication_request.rs
@@ -0,0 +1,32 @@
+pub enum ReplicationRequest {
+    Ping,
+}
+
+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() {
+            "ping" => Ok(ReplicationRequest::Ping),
+            _ => 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),
+
+            _ => Err(anyhow::anyhow!("Invalid command")),
+        }
+    }
+}


Test patch:
diff --git a/tests/common.rs b/tests/common.rs
index f32131bb..a8e91915 100644
--- a/tests/common.rs
+++ b/tests/common.rs
@@ -1,6 +1,6 @@
 use redis_starter_rust::adapters::io::tokio_stream::TokioConnectStreamFactory;
-use redis_starter_rust::services::query_manager::interface::TCancellationTokenFactory;
-use redis_starter_rust::services::query_manager::query_io::QueryIO;
+use redis_starter_rust::services::stream_manager::interface::TCancellationTokenFactory;
+use redis_starter_rust::services::stream_manager::query_io::QueryIO;
 use redis_starter_rust::{
     adapters::io::tokio_stream::TokioStreamListenerFactory, services::config::config_actor::Config,
 };
diff --git a/tests/test_cancellation_token.rs b/tests/test_cancellation_token.rs
index e478c1ec..aad54040 100644
--- a/tests/test_cancellation_token.rs
+++ b/tests/test_cancellation_token.rs
@@ -4,7 +4,7 @@
 /// After 300ms, we get the key again and check if the value is not returned (-1)
 mod common;
 use common::{init_config_with_free_port, start_test_server, TestStreamHandler};
-use redis_starter_rust::services::query_manager::interface::{
+use redis_starter_rust::services::stream_manager::interface::{
     TCancellationNotifier, TCancellationTokenFactory, TCancellationWatcher,
 };
 use tokio::net::TcpStream;