f Use tokio::sync::broadcast channel

Author: tnull

src/wallet.rs

@@ -33,14 +33,13 @@ use bitcoin::secp256k1::ecdsa::{RecoverableSignature, Signature};
 use bitcoin::secp256k1::{PublicKey, Scalar, Secp256k1, SecretKey, Signing};
 use bitcoin::{ScriptBuf, Transaction, TxOut, Txid};
 
-use std::mem;
 use std::ops::{Deref, DerefMut};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::Duration;
 
 enum WalletSyncStatus {
 	Completed,
-	InProgress { subscribers: Vec<tokio::sync::oneshot::Sender<Result<(), Error>>> },
+	InProgress { subscribers: tokio::sync::broadcast::Sender<Result<(), Error>> },
 }
 
 pub struct Wallet<D, B: Deref, E: Deref, L: Deref>
@@ -89,9 +88,9 @@ where
 	}
 
 	pub(crate) async fn sync(&self) -> Result<(), Error> {
-		if let Some(sync_receiver) = self.register_or_subscribe_pending_sync() {
+		if let Some(mut sync_receiver) = self.register_or_subscribe_pending_sync() {
 			log_info!(self.logger, "Sync in progress, skipping.");
-			return sync_receiver.await.map_err(|e| {
+			return sync_receiver.recv().await.map_err(|e| {
 				debug_assert!(false, "Failed to receive wallet sync result: {:?}", e);
 				log_error!(self.logger, "Failed to receive wallet sync result: {:?}", e);
 				Error::WalletOperationFailed
@@ -310,26 +309,25 @@ where
 
 	fn register_or_subscribe_pending_sync(
 		&self,
-	) -> Option<tokio::sync::oneshot::Receiver<Result<(), Error>>> {
+	) -> Option<tokio::sync::broadcast::Receiver<Result<(), Error>>> {
 		let mut sync_status_lock = self.sync_status.lock().unwrap();
 		match sync_status_lock.deref_mut() {
 			WalletSyncStatus::Completed => {
 				// We're first to register for a sync.
-				*sync_status_lock = WalletSyncStatus::InProgress { subscribers: Vec::new() };
+				let (tx, _) = tokio::sync::broadcast::channel(1);
+				*sync_status_lock = WalletSyncStatus::InProgress { subscribers: tx };
 				None
 			},
 			WalletSyncStatus::InProgress { subscribers } => {
 				// A sync is in-progress, we subscribe.
-				let (tx, rx) = tokio::sync::oneshot::channel();
-				subscribers.push(tx);
+				let rx = subscribers.subscribe();
 				Some(rx)
 			},
 		}
 	}
 
 	fn propagate_result_to_subscribers(&self, res: Result<(), Error>) {
 		// Send the notification to any other tasks that might be waiting on it by now.
-		let mut waiting_subscribers = Vec::new();
 		{
 			let mut sync_status_lock = self.sync_status.lock().unwrap();
 			match sync_status_lock.deref_mut() {
@@ -339,22 +337,27 @@ where
 				},
 				WalletSyncStatus::InProgress { subscribers } => {
 					// A sync is in-progress, we notify subscribers.
-					mem::swap(&mut waiting_subscribers, subscribers);
+					if subscribers.receiver_count() > 0 {
+						match subscribers.send(res) {
+							Ok(_) => (),
+							Err(e) => {
+								debug_assert!(
+									false,
+									"Failed to send wallet sync result to subscribers: {:?}",
+									e
+								);
+								log_error!(
+									self.logger,
+									"Failed to send wallet sync result to subscribers: {:?}",
+									e
+								);
+							},
+						}
+					}
 					*sync_status_lock = WalletSyncStatus::Completed;
 				},
 			}
 		}
-
-		for sender in waiting_subscribers {
-			sender.send(res).unwrap_or_else(|e| {
-				debug_assert!(false, "Failed to send wallet sync result to subscribers: {:?}", e);
-				log_error!(
-					self.logger,
-					"Failed to send wallet sync result to subscribers: {:?}",
-					e
-				);
-			});
-		}
 	}
 }