watch_pipe

Author: taslimmuhammed

common/src/watcher.rs

@@ -14,7 +14,7 @@ use tokio::{
     task::JoinHandle,
     time::{self, sleep},
 };
-use tracing::warn;
+use tracing::{error, warn};
 
 /// Creates a new watcher that auto initializes it with initial_value
 /// and updates it given an interval
@@ -106,7 +106,7 @@ where
                     function(value).await;
                 }
                 Err(err) => {
-                    warn!("{err}");
+                    error!("There was an error piping the watcher results: {err}");
                     break;
                 }
             };

tap-agent/src/agent/sender_account.rs

@@ -498,10 +498,9 @@ impl Actor for SenderAccount {
     ) -> std::result::Result<Self::State, ActorProcessingErr> {
         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();
             // Update the allocation_ids
-            myself
+            myself_clone
                 .cast(SenderAccountMessage::UpdateAllocationIds(allocation_ids))
                 .unwrap_or_else(|e| {
                     error!("Error while updating allocation_ids: {:?}", e);

tap-agent/src/agent/sender_accounts_manager.rs

@@ -106,7 +106,6 @@ impl Actor for SenderAccountsManager {
     ) -> std::result::Result<Self::State, ActorProcessingErr> {
         let (allocations_tx, allocations_rx) = watch::channel(HashSet::<Address>::new());
         watch_pipe(indexer_allocations.clone(), move |allocation_id| {
-            let allocations_tx = allocations_tx.clone();
             let allocation_set = allocation_id.keys().cloned().collect::<HashSet<Address>>();
             allocations_tx
                 .send(allocation_set)
@@ -125,9 +124,8 @@ impl Actor for SenderAccountsManager {
         let accounts_clone = escrow_accounts.clone();
         let _eligible_allocations_senders_handle =
             watch_pipe(accounts_clone, move |escrow_accounts| {
-                let myself = myself_clone.clone();
                 let senders = escrow_accounts.get_senders();
-                myself
+                myself_clone
                     .cast(SenderAccountsManagerMessage::UpdateSenderAccounts(senders))
                     .unwrap_or_else(|e| {
                         error!("Error while updating sender_accounts: {:?}", e);