peer list added auth response

Author: Migorithm

duva-client/src/controller.rs

@@ -3,7 +3,7 @@ use duva::{
     clients::authentications::{AuthRequest, AuthResponse},
     domains::query_parsers::query_io::{QueryIO, deserialize},
     prelude::{
-        BytesMut,
+        BytesMut, PeerIdentifier,
         tokio::{
             self,
             io::{AsyncReadExt, AsyncWriteExt},
@@ -19,27 +19,31 @@ pub struct ClientController<T> {
     client_id: Uuid,
     request_id: u64,
     latest_known_index: u64,
+    cluster_nodes: Vec<PeerIdentifier>,
     pub target: T,
 }
 
 impl<T> ClientController<T> {
     pub async fn new(editor: T, server_addr: &str) -> Self {
-        let (stream, client_id, request_id) =
-            ClientController::<T>::authenticate(server_addr).await;
-        Self { stream, client_id, target: editor, latest_known_index: 0, request_id }
+        let (stream, mut auth_response) = ClientController::<T>::authenticate(server_addr).await;
+        auth_response.cluster_nodes.push(server_addr.to_string().into());
+        Self {
+            stream,
+            client_id: Uuid::parse_str(&auth_response.client_id).unwrap(),
+            target: editor,
+            latest_known_index: 0,
+            request_id: auth_response.request_id,
+            cluster_nodes: auth_response.cluster_nodes,
+        }
     }
 
-    async fn authenticate(server_addr: &str) -> (TcpStream, Uuid, u64) {
+    async fn authenticate(server_addr: &str) -> (TcpStream, AuthResponse) {
         let mut stream = TcpStream::connect(server_addr).await.unwrap();
         stream.serialized_write(AuthRequest::default()).await.unwrap(); // client_id not exist
 
-        let AuthResponse { client_id, request_id } = stream.deserialized_read().await.unwrap();
-
-        let client_id = Uuid::parse_str(&client_id).unwrap();
-        println!("Client ID: {}", client_id);
-        println!("Connected to Redis at {}", server_addr);
+        let auth_response: AuthResponse = stream.deserialized_read().await.unwrap();
 
-        (stream, client_id, request_id)
+        (stream, auth_response)
     }
 
     pub async fn send_command(

duva/src/clients/authentications.rs

@@ -1,3 +1,5 @@
+use crate::domains::peers::identifier::PeerIdentifier;
+
 #[derive(Debug, Clone, PartialEq, Eq, Default, bincode::Decode, bincode::Encode)]
 pub struct AuthRequest {
     pub client_id: Option<String>,
@@ -7,4 +9,5 @@ pub struct AuthRequest {
 pub struct AuthResponse {
     pub client_id: String,
     pub request_id: u64,
+    pub cluster_nodes: Vec<PeerIdentifier>,
 }

duva/src/clients/utils.rs

@@ -34,7 +34,8 @@ impl ClientStreamHandler {
 
         stream.serialized_write(AuthRequest::default()).await.unwrap(); // client_id not exist
 
-        let AuthResponse { client_id, request_id } = stream.deserialized_read().await.unwrap();
+        let AuthResponse { client_id, request_id, cluster_nodes } =
+            stream.deserialized_read().await.unwrap();
         let client_id = Uuid::parse_str(&client_id).unwrap();
 
         let (read_half, write_half) = stream.into_split();

duva/src/domains/peers/identifier.rs

@@ -3,7 +3,7 @@ use crate::{from_to, make_smart_pointer};
 #[derive(
     Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Hash, Default, bincode::Encode, bincode::Decode,
 )]
-pub(crate) struct PeerIdentifier(pub String);
+pub struct PeerIdentifier(pub String);
 impl PeerIdentifier {
     pub(crate) fn new(host: &str, port: u16) -> Self {
         Self(format!("{}:{}", host, port))

duva/src/domains/peers/mod.rs

@@ -1,5 +1,5 @@
 pub mod connected_peer_info;
 pub(crate) mod connected_types;
-pub(crate) mod identifier;
+pub mod identifier;
 
 pub(crate) mod peer;

duva/src/lib.rs

@@ -23,6 +23,7 @@ use services::interface::TSerdeReadWrite;
 use tokio::net::TcpListener;
 
 pub mod prelude {
+    pub use crate::domains::peers::identifier::PeerIdentifier;
     pub use bytes;
     pub use bytes::BytesMut;
     pub use tokio;
@@ -128,8 +129,10 @@ impl StartUpFacade {
         let client_stream_listener = TcpListener::bind(&self.config_manager.bind_addr()).await?;
         println!("start listening on {}", self.config_manager.bind_addr());
         let mut conn_handlers: Vec<tokio::task::JoinHandle<()>> = Vec::with_capacity(100);
+
         while let Ok((stream, _)) = client_stream_listener.accept().await {
-            let Ok(client_stream) = ClientStream::authenticate(stream).await else {
+            let peers = self.registry.cluster_communication_manager().get_peers().await?;
+            let Ok(client_stream) = ClientStream::authenticate(stream, peers).await else {
                 eprintln!("[ERROR] Failed to authenticate client stream");
                 continue;
             };

duva/src/presentation/clients/stream.rs

@@ -2,7 +2,10 @@ use super::{parser::parse_query, request::ClientRequest};
 use crate::{
     TSerdeReadWrite,
     clients::authentications::{AuthRequest, AuthResponse},
-    domains::{IoError, cluster_actors::session::SessionRequest, query_parsers::QueryIO},
+    domains::{
+        IoError, cluster_actors::session::SessionRequest, peers::identifier::PeerIdentifier,
+        query_parsers::QueryIO,
+    },
     make_smart_pointer,
     services::interface::TRead,
 };
@@ -18,7 +21,10 @@ pub struct ClientStream {
 make_smart_pointer!(ClientStream, TcpStream=>stream);
 
 impl ClientStream {
-    pub(crate) async fn authenticate(mut stream: TcpStream) -> Result<Self, IoError> {
+    pub(crate) async fn authenticate(
+        mut stream: TcpStream,
+        peers: Vec<PeerIdentifier>,
+    ) -> Result<Self, IoError> {
         let auth_req: AuthRequest = stream.deserialized_read().await?;
 
         let client_id = match auth_req.client_id {
@@ -30,7 +36,11 @@ impl ClientStream {
         };
 
         stream
-            .serialized_write(AuthResponse { client_id: client_id.to_string(), request_id: 0 })
+            .serialized_write(AuthResponse {
+                client_id: client_id.to_string(),
+                request_id: 0,
+                cluster_nodes: peers,
+            })
             .await?;
 
         Ok(Self { stream, client_id })