Batch subscriber

chitchat-test/src/main.rs

@@ -37,8 +37,7 @@ impl Api {
     #[oai(path = "/set_kv/", method = "get")]
     async fn set_kv(&self, key: Query<String>, value: Query<String>) -> Json<serde_json::Value> {
         let mut chitchat_guard = self.chitchat.lock().await;
-
-        let cc_state = chitchat_guard.self_node_state();
+        let mut cc_state = chitchat_guard.self_node_state();
         cc_state.set(key.as_str(), value.as_str());
 
         Json(serde_json::to_value(&SetKeyValueResponse { status: true }).unwrap())
@@ -48,8 +47,7 @@ impl Api {
     #[oai(path = "/mark_for_deletion/", method = "get")]
     async fn mark_for_deletion(&self, key: Query<String>) -> Json<serde_json::Value> {
         let mut chitchat_guard = self.chitchat.lock().await;
-
-        let cc_state = chitchat_guard.self_node_state();
+        let mut cc_state = chitchat_guard.self_node_state();
         cc_state.delete(key.as_str());
         Json(serde_json::to_value(&SetKeyValueResponse { status: true }).unwrap())
     }

chitchat/src/delta.rs

@@ -240,7 +240,7 @@ impl Delta {
     pub(crate) fn num_tuples(&self) -> usize {
         self.node_deltas
             .iter()
-            .map(|node_delta| node_delta.num_tuples())
+            .map(|node_delta| node_delta.key_values.len())
             .sum()
     }
 
@@ -323,10 +323,12 @@ pub(crate) struct NodeDelta {
     pub max_version: Option<Version>,
 }
 
-#[cfg(test)]
-impl NodeDelta {
-    pub fn num_tuples(&self) -> usize {
-        self.key_values.len()
+impl IntoIterator for NodeDelta {
+    type Item = KeyValueMutation;
+    type IntoIter = std::vec::IntoIter<Self::Item>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.key_values.into_iter()
     }
 }