split function to flatten recursion

Author: Migorithm

duva/src/lib.rs

@@ -112,17 +112,19 @@ impl StartUpFacade {
     // #1 Check if it has persisted cluster node info locally, if it does and finds some other nodes, try connect
     // #2 Otherwise, check if this node is a follower and connect to the leader
     async fn discover_peers(&self) -> anyhow::Result<()> {
-        let connection_manager = self.registry.cluster_connection_manager();
-        let peer_identifier = self
+        let connect_to = self
             .registry
             .cluster_communication_manager()
             .replication_info()
             .await?
             .leader_bind_addr()
             .context("No leader bind address found")?;
-        let (tx, rx) = tokio::sync::oneshot::channel();
-        connection_manager.discover_cluster(self.config_manager.port, peer_identifier, tx).await?;
-        let _ = rx.await;
+
+        self.registry
+            .cluster_connection_manager()
+            .discover_cluster(self.config_manager.port, connect_to)
+            .await?;
+
         Ok(())
     }
 

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

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

duva/src/presentation/clusters/connection_manager.rs

@@ -1,10 +1,11 @@
+use std::collections::VecDeque;
+
 use super::communication_manager::ClusterCommunicationManager;
 use super::outbound::stream::OutboundStream;
 
 use crate::domains::cluster_actors::commands::ClusterCommand;
 use crate::domains::peers::identifier::PeerIdentifier;
 use crate::{InboundStream, make_smart_pointer};
-use tokio::sync::mpsc::Sender;
 
 pub(crate) struct ClusterConnectionManager(pub(crate) ClusterCommunicationManager);
 
@@ -20,7 +21,7 @@ impl ClusterConnectionManager {
         peer_stream.may_try_sync(ccm, &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?;
+        self.send(peer_stream.into_add_peer(self.0.0.clone(), connected_peer_info, tx)?).await?;
 
         Ok(())
     }
@@ -29,45 +30,46 @@ impl ClusterConnectionManager {
         self,
         self_port: u16,
         connect_to: PeerIdentifier,
-        sender: tokio::sync::oneshot::Sender<()>,
     ) -> anyhow::Result<()> {
-        // Base case
+        let mut queue = VecDeque::from(vec![connect_to.clone()]);
+        let mut callbacks = Vec::new();
+        while let Some(connect_to) = queue.pop_front() {
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            queue.extend(
+                ClusterConnectionManager(self.clone())
+                    .connect_to_server(self_port, connect_to, tx)
+                    .await?,
+            );
+            callbacks.push(rx);
+        }
+        for callback in callbacks {
+            let _ = callback.await;
+        }
+        Ok(())
+    }
+
+    async fn connect_to_server(
+        self,
+        self_port: u16,
+        connect_to: PeerIdentifier,
+        sender: tokio::sync::oneshot::Sender<()>,
+    ) -> anyhow::Result<Vec<PeerIdentifier>> {
         let existing_peers = self.get_peers().await?;
         if existing_peers.contains(&connect_to) {
-            return Ok(());
+            return Ok(vec![]);
         }
 
-        // Recursive case
         let replication_info = self.replication_info().await?;
         let (add_peer_cmd, peer_list) = OutboundStream::new(connect_to, replication_info)
             .await?
             .establish_connection(self_port)
             .await?
             .set_replication_info(&self)
             .await?
-            .create_peer_cmd(self.clone(), sender)?;
+            .create_peer_cmd(self.0.0.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, tx),
-            )
-            .await?;
-            callbacks.push(rx);
-        }
-        for callback in callbacks {
-            let _ = callback.await;
-        }
-        Ok(())
-    }
-
-    pub fn clone(&self) -> Sender<ClusterCommand> {
-        self.0.0.clone()
+        Ok(peer_list)
     }
 
     pub async fn send(&self, cmd: ClusterCommand) -> anyhow::Result<()> {