Fn(Ref<'_,T>)->Fut

Author: taslimmuhammed

common/src/watcher.rs

@@ -10,7 +10,7 @@ use std::{future::Future, time::Duration};
 
 use tokio::{
     select,
-    sync::watch,
+    sync::watch::{self, Ref},
     task::JoinHandle,
     time::{self, sleep},
 };
@@ -91,18 +91,18 @@ where
 pub fn watch_pipe<T, F, Fut>(rx: watch::Receiver<T>, function: F) -> JoinHandle<()>
 where
     T: Clone + Send + Sync + 'static,
-    F: Fn(T) -> Fut + Send + Sync + 'static,
+    F: Fn(Ref<'_, T>) -> Fut + Send + Sync + 'static,
     Fut: Future<Output = ()> + Send + 'static,
 {
     tokio::spawn(async move {
         let mut rx = rx;
-        let value = rx.borrow().clone();
+        let value = rx.borrow();
         function(value).await;
         loop {
             let res = rx.changed().await;
             match res {
                 Ok(_) => {
-                    let value = rx.borrow().clone();
+                    let value = rx.borrow();
                     function(value).await;
                 }
                 Err(err) => {

tap-agent/src/agent/sender_account.rs

@@ -499,6 +499,7 @@ impl Actor for SenderAccount {
         let myself_clone = myself.clone();
         let _indexer_allocations_handle = watch_pipe(indexer_allocations, move |allocation_ids| {
             let myself = myself_clone.clone();
+            let allocation_ids = allocation_ids.clone();
             async move {
                 // Update the allocation_ids
                 myself
@@ -515,6 +516,7 @@ impl Actor for SenderAccount {
         let _escrow_account_monitor = watch_pipe(accounts_clone, move |escrow_account| {
             let myself = myself_clone.clone();
             let pgpool = pgpool_clone.clone();
+            let escrow_account = escrow_account.clone();
             async move {
                 // Get balance or default value for sender
                 // this balance already takes into account thawing