wait on drop

Author: Migorithm

duva/src/domains/caches/cache_manager.rs

@@ -225,6 +225,13 @@ impl CacheManager {
     }
 
     pub(crate) async fn drop_cache(&self) {
-        join_all(self.inboxes.iter().map(|shard| shard.send(CacheCommand::Drop))).await;
+        let (txs, rxs) = self.oneshot_channels();
+        join_all(
+            self.chain(txs)
+                .map(|(shard, sender)| shard.send(CacheCommand::Drop { callback: sender })),
+        )
+        .await;
+
+        join_all(rxs.into_iter().map(|rx| rx)).await;
     }
 }

duva/src/domains/caches/command.rs

@@ -13,6 +13,6 @@ pub enum CacheCommand {
     IndexGet { key: String, read_idx: u64, callback: oneshot::Sender<QueryIO> },
     Ping,
     StopSentinel,
-    Drop,
+    Drop { callback: oneshot::Sender<()> },
     Exists { key: String, callback: oneshot::Sender<bool> },
 }

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

@@ -94,7 +94,6 @@ impl ClientController<Handler> {
                 }
             },
             ClientAction::ReplicaOf(peer_identifier) => {
-                // TODO should check if the peer is in the cluster?
                 self.cluster_communication_manager.replicaof(peer_identifier.clone()).await;
 
                 let (tx, rx) = tokio::sync::oneshot::channel();

duva/src/presentation/clusters/connection_manager.rs

@@ -29,7 +29,7 @@ impl ClusterConnectionManager {
         self,
         self_port: u16,
         connect_to: PeerIdentifier,
-        Sender: tokio::sync::oneshot::Sender<()>,
+        sender: tokio::sync::oneshot::Sender<()>,
     ) -> anyhow::Result<()> {
         // Base case
         let existing_peers = self.get_peers().await?;
@@ -45,7 +45,7 @@ impl ClusterConnectionManager {
             .await?
             .set_replication_info(&self)
             .await?
-            .create_peer_cmd(self.clone(), Sender)?;
+            .create_peer_cmd(self.clone(), sender)?;
         self.send(add_peer_cmd).await?;
 
         // Discover additional peers concurrently

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>,
-        Sender: 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 }, Sender), peer_list))
+        Ok((ClusterCommand::AddPeer(AddPeer { peer_id: self.connect_to, peer }, sender), peer_list))
     }
 }

duva/src/services/handlers/cache.rs

@@ -61,8 +61,9 @@ impl CacheActor {
                         }
                     };
                 },
-                CacheCommand::Drop => {
+                CacheCommand::Drop { callback } => {
                     self.cache.clear();
+                    let _ = callback.send(());
                 },
             }
         }
@@ -104,7 +105,9 @@ mod test {
             self.0.send(CacheCommand::Ping).await.unwrap();
         }
         async fn drop(&self) {
-            self.0.send(CacheCommand::Drop).await.unwrap();
+            let (tx, rx) = oneshot::channel();
+            self.0.send(CacheCommand::Drop { callback: tx }).await.unwrap();
+            let _ = rx.await;
         }
     }