refactor: merge join and map watcher functions

Signed-off-by: Gustavo Inacio <[email protected]>

Author: gusinacio

common/src/attestations/signers.rs

@@ -10,7 +10,7 @@ use tracing::warn;
 
 use crate::{
     prelude::{Allocation, AttestationSigner},
-    watcher::{join_watcher, map_watcher},
+    watcher::join_and_map_watcher,
 };
 
 /// An always up-to-date list of attestation signers, one for each of the indexer's allocations.
@@ -24,18 +24,20 @@ pub fn attestation_signers(
         Box::leak(Box::new(Mutex::new(HashMap::new())));
     let indexer_mnemonic = Arc::new(indexer_mnemonic.to_string());
 
-    let joined_watcher = join_watcher(indexer_allocations_rx, dispute_manager_rx);
-
-    map_watcher(joined_watcher, move |(allocation, dispute)| {
-        let indexer_mnemonic = indexer_mnemonic.clone();
-        modify_sigers(
-            &indexer_mnemonic,
-            chain_id,
-            attestation_signers_map,
-            &allocation,
-            &dispute,
-        )
-    })
+    join_and_map_watcher(
+        indexer_allocations_rx,
+        dispute_manager_rx,
+        move |(allocation, dispute)| {
+            let indexer_mnemonic = indexer_mnemonic.clone();
+            modify_sigers(
+                &indexer_mnemonic,
+                chain_id,
+                attestation_signers_map,
+                &allocation,
+                &dispute,
+            )
+        },
+    )
 }
 fn modify_sigers(
     indexer_mnemonic: &str,

common/src/watcher.rs

@@ -1,7 +1,6 @@
 // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
 // SPDX-License-Identifier: Apache-2.0
 
-
 //! This is a module that reimplements eventuals using
 //! tokio::watch module and fixing some problems that eventuals
 //! usually carry like initializing things without initializing
@@ -32,14 +31,13 @@ where
     let (tx, rx) = watch::channel(initial_value);
 
     tokio::spawn(async move {
-        // Refresh indexer allocations every now and then
         let mut time_interval = time::interval(interval);
         time_interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
         loop {
             time_interval.tick().await;
             let result = function().await;
             match result {
-                Ok(allocations) => tx.send(allocations).expect("Failed to update channel"),
+                Ok(value) => tx.send(value).expect("Failed to update channel"),
                 Err(err) => {
                     // TODO mark it as delayed
                     warn!(error = %err, "There was an error while updating watcher");
@@ -52,17 +50,20 @@ where
     Ok(rx)
 }
 
-
 /// Join two watch::Receiver
-pub fn join_watcher<T1, T2>(
+pub fn join_and_map_watcher<T1, T2, T3, F>(
     mut receiver_1: watch::Receiver<T1>,
     mut receiver_2: watch::Receiver<T2>,
-) -> watch::Receiver<(T1, T2)>
+    map_function: F,
+) -> watch::Receiver<T3>
 where
     T1: Clone + Send + Sync + 'static,
     T2: Clone + Send + Sync + 'static,
+    T3: Send + Sync + 'static,
+    F: Fn((T1, T2)) -> T3 + Send + 'static,
 {
-    let (tx, rx) = watch::channel((receiver_1.borrow().clone(), receiver_2.borrow().clone()));
+    let initial_value = map_function((receiver_1.borrow().clone(), receiver_2.borrow().clone()));
+    let (tx, rx) = watch::channel(initial_value);
 
     tokio::spawn(async move {
         loop {
@@ -74,34 +75,10 @@ where
                     panic!("receiver_1 or receiver_2 was dropped");
                 }
             }
-            tx.send((receiver_1.borrow().clone(), receiver_2.borrow().clone()))
-                .expect("Failed to update signers channel");
-        }
-    });
-    rx
-}
 
-/// Maps a watch::Receiver into another type
-pub fn map_watcher<T1, T2, F>(
-    mut receiver: watch::Receiver<T1>,
-    map_function: F,
-) -> watch::Receiver<T2>
-where
-    F: Fn(T1) -> T2 + Send + 'static,
-    T1: Default + Clone + Sync + Send + 'static,
-    T2: Sync + Send + 'static,
-{
-    let initial_value = map_function(receiver.borrow().clone());
-    let (tx, rx) = watch::channel(initial_value);
-
-    tokio::spawn(async move {
-        loop {
-            match receiver.changed().await {
-                Ok(_) => {}
-                Err(_) => panic!("reciever was dropped"),
-            }
-            let current_val = receiver.borrow().clone();
-            let mapped_value = map_function(current_val);
+            let current_val_1 = receiver_1.borrow().clone();
+            let current_val_2 = receiver_2.borrow().clone();
+            let mapped_value = map_function((current_val_1, current_val_2));
             tx.send(mapped_value).expect("Failed to update channel");
         }
     });