feat: remove option callback for AddPeer Command

Signed-off-by: Eun-chan Cho <eunchan.cho@talabat.com>

Author: echo724

duva/src/domains/cluster_actors/commands/mod.rs

@@ -14,7 +14,7 @@ use crate::domains::{append_only_files::WriteRequest, peers::identifier::PeerIde
 
 #[derive(Debug)]
 pub(crate) enum ClusterCommand {
-    AddPeer(AddPeer, Option<tokio::sync::oneshot::Sender<()>>),
+    AddPeer(AddPeer, tokio::sync::oneshot::Sender<()>),
     GetPeers(tokio::sync::oneshot::Sender<Vec<PeerIdentifier>>),
     ReplicationInfo(tokio::sync::oneshot::Sender<ReplicationState>),
 

duva/src/lib.rs

@@ -120,9 +120,9 @@ impl StartUpFacade {
             .await?
             .leader_bind_addr()
             .context("No leader bind address found")?;
-        connection_manager
-            .discover_cluster(self.config_manager.port, peer_identifier, None)
-            .await?;
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        connection_manager.discover_cluster(self.config_manager.port, peer_identifier, tx).await?;
+        let _ = rx.await;
         Ok(())
     }
 

duva/src/presentation/clients/controllers/handler.rs

@@ -99,10 +99,9 @@ impl ClientController<Handler> {
 
                 let (tx, rx) = tokio::sync::oneshot::channel();
                 ClusterConnectionManager(self.cluster_communication_manager.clone())
-                    .discover_cluster(self.config_manager.port, peer_identifier, Some(tx))
+                    .discover_cluster(self.config_manager.port, peer_identifier, tx)
                     .await?;
                 let _ = rx.await;
-
                 QueryIO::SimpleString("OK".into())
             },
         };

duva/src/presentation/clusters/connection_manager.rs

@@ -19,7 +19,8 @@ impl ClusterConnectionManager {
         peer_stream.disseminate_peers(self.0.get_peers().await?).await?;
         peer_stream.may_try_sync(ccm, &connected_peer_info).await?;
 
-        self.send(peer_stream.into_add_peer(self.clone(), connected_peer_info)?).await?;
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        self.send(peer_stream.into_add_peer(self.clone(), connected_peer_info, tx)?).await?;
 
         Ok(())
     }
@@ -28,7 +29,7 @@ impl ClusterConnectionManager {
         self,
         self_port: u16,
         connect_to: PeerIdentifier,
-        tx: Option<tokio::sync::oneshot::Sender<()>>,
+        Sender: tokio::sync::oneshot::Sender<()>,
     ) -> anyhow::Result<()> {
         // Base case
         let existing_peers = self.get_peers().await?;
@@ -44,19 +45,24 @@ impl ClusterConnectionManager {
             .await?
             .set_replication_info(&self)
             .await?
-            .create_peer_cmd(self.clone(), tx)?;
+            .create_peer_cmd(self.clone(), Sender)?;
         self.send(add_peer_cmd).await?;
 
         // Discover additional peers concurrently
         // TODO Require investigation. Why does 'list_peer_binding_addrs' have to be called at here?
+        let mut callbacks = Vec::new();
         for peer in peer_list {
             println!("Discovering peer: {}", peer);
+            let (tx, rx) = tokio::sync::oneshot::channel();
             Box::pin(
-                ClusterConnectionManager(self.0.clone()).discover_cluster(self_port, peer, None),
+                ClusterConnectionManager(self.0.clone()).discover_cluster(self_port, peer, tx),
             )
             .await?;
+            callbacks.push(rx);
+        }
+        for callback in callbacks {
+            let _ = callback.await;
         }
-
         Ok(())
     }
 

duva/src/presentation/clusters/inbound/stream.rs

@@ -116,6 +116,7 @@ impl InboundStream {
         self,
         cluster_actor_handler: Sender<ClusterCommand>,
         connected_peer_info: ConnectedPeerInfo,
+        Sender: tokio::sync::oneshot::Sender<()>,
     ) -> anyhow::Result<ClusterCommand> {
         let kind = self.decide_peer_kind(&connected_peer_info);
         let peer = create_peer(
@@ -124,7 +125,7 @@ impl InboundStream {
             kind,
             cluster_actor_handler,
         );
-        Ok(ClusterCommand::AddPeer(AddPeer { peer_id: connected_peer_info.id, peer }, None))
+        Ok(ClusterCommand::AddPeer(AddPeer { peer_id: connected_peer_info.id, peer }, Sender))
     }
 
     pub(crate) fn decide_peer_kind(&self, connected_peer_info: &ConnectedPeerInfo) -> PeerState {

duva/src/presentation/clusters/outbound/stream.rs

@@ -114,7 +114,7 @@ impl OutboundStream {
     pub(crate) fn create_peer_cmd(
         self,
         cluster_actor_handler: Sender<ClusterCommand>,
-        tx: Option<tokio::sync::oneshot::Sender<()>>,
+        Sender: tokio::sync::oneshot::Sender<()>,
     ) -> anyhow::Result<(ClusterCommand, Vec<PeerIdentifier>)> {
         let mut connection_info =
             self.connected_node_info.context("Connected node info not found")?;
@@ -127,6 +127,6 @@ impl OutboundStream {
             cluster_actor_handler,
         );
 
-        Ok((ClusterCommand::AddPeer(AddPeer { peer_id: self.connect_to, peer }, tx), peer_list))
+        Ok((ClusterCommand::AddPeer(AddPeer { peer_id: self.connect_to, peer }, Sender), peer_list))
     }
 }

duva/src/services/handlers/cluster.rs

@@ -24,9 +24,7 @@ impl ClusterActor {
                 ClusterCommand::AddPeer(add_peer_cmd, callback) => {
                     self.add_peer(add_peer_cmd).await;
                     self.snapshot_topology().await;
-                    if let Some(callback) = callback {
-                        let _ = callback.send(());
-                    }
+                    let _ = callback.send(());
                 },
                 ClusterCommand::GetPeers(callback) => {
                     let _ = callback.send(self.members.keys().cloned().collect::<Vec<_>>());