Fn(T)->Fut

Author: taslimmuhammed

common/src/watcher.rs

@@ -91,14 +91,25 @@ where
 pub fn watch_pipe<T, F, Fut>(rx: watch::Receiver<T>, function: F) -> JoinHandle<()>
 where
     T: Clone + Send + Sync + 'static,
-    F: Fn(watch::Receiver<T>) -> Fut + Send + Sync + 'static,
+    F: Fn(T) -> Fut + Send + Sync + 'static,
     Fut: Future<Output = ()> + Send + 'static,
 {
     tokio::spawn(async move {
         let mut rx = rx;
-        function(rx.clone()).await;
-        while rx.changed().await.is_ok() {
-            function(rx.clone()).await;
+        let value = rx.borrow().clone();
+        function(value).await;
+        loop {
+            let res = rx.changed().await;
+            match res {
+                Ok(_) => {
+                    let value = rx.borrow().clone();
+                    function(value).await;
+                }
+                Err(err) => {
+                    warn!("{err}");
+                    break;
+                }
+            };
         }
     })
 }

tap-agent/src/agent/sender_account.rs

@@ -497,10 +497,9 @@ impl Actor for SenderAccount {
         }: Self::Arguments,
     ) -> std::result::Result<Self::State, ActorProcessingErr> {
         let myself_clone = myself.clone();
-        let _indexer_allocations_handle = watch_pipe(indexer_allocations, move |rx| {
+        let _indexer_allocations_handle = watch_pipe(indexer_allocations, move |allocation_ids| {
             let myself = myself_clone.clone();
             async move {
-                let allocation_ids = rx.borrow().clone();
                 // Update the allocation_ids
                 myself
                     .cast(SenderAccountMessage::UpdateAllocationIds(allocation_ids))
@@ -513,8 +512,7 @@ impl Actor for SenderAccount {
         let myself_clone = myself.clone();
         let pgpool_clone = pgpool.clone();
         let accounts_clone = escrow_accounts.clone();
-        let _escrow_account_monitor = watch_pipe(accounts_clone, move |rx| {
-            let escrow_account = rx.borrow().clone();
+        let _escrow_account_monitor = watch_pipe(accounts_clone, move |escrow_account| {
             let myself = myself_clone.clone();
             let pgpool = pgpool_clone.clone();
             async move {

tap-agent/src/agent/sender_accounts_manager.rs

@@ -105,11 +105,12 @@ impl Actor for SenderAccountsManager {
         }: Self::Arguments,
     ) -> std::result::Result<Self::State, ActorProcessingErr> {
         let (allocations_tx, allocations_rx) = watch::channel(HashSet::<Address>::new());
-        watch_pipe(indexer_allocations.clone(), move |rx| {
+        watch_pipe(indexer_allocations.clone(), move |allocation_id| {
             let allocations_tx = allocations_tx.clone();
+            let allocation_set = allocation_id.keys().cloned().collect::<HashSet<Address>>();
             async move {
                 allocations_tx
-                    .send(rx.borrow().keys().cloned().collect::<HashSet<Address>>())
+                    .send(allocation_set)
                     .expect("Failed to update indexer_allocations_set channel");
             }
         });
@@ -123,18 +124,18 @@ impl Actor for SenderAccountsManager {
             );
         let myself_clone = myself.clone();
         let accounts_clone = escrow_accounts.clone();
-        let _eligible_allocations_senders_handle = watch_pipe(accounts_clone, move |rx| {
-            let myself = myself_clone.clone();
-            async move {
-                myself
-                    .cast(SenderAccountsManagerMessage::UpdateSenderAccounts(
-                        rx.borrow().get_senders(),
-                    ))
-                    .unwrap_or_else(|e| {
-                        error!("Error while updating sender_accounts: {:?}", e);
-                    });
-            }
-        });
+        let _eligible_allocations_senders_handle =
+            watch_pipe(accounts_clone, move |escrow_accounts| {
+                let myself = myself_clone.clone();
+                let senders = escrow_accounts.get_senders();
+                async move {
+                    myself
+                        .cast(SenderAccountsManagerMessage::UpdateSenderAccounts(senders))
+                        .unwrap_or_else(|e| {
+                            error!("Error while updating sender_accounts: {:?}", e);
+                        });
+                }
+            });
 
         let mut state = State {
             config,