Move continuous tx broadcast processing loop to ChainSource

Rather than looping in the `spawn` method directly, we move the loop to
a refactored `continuously_process_broadcast_queue` method on
`ChainSource`, which also allows us to react on the stop signal if we're
polling `recv`.

Author: tnull

src/chain/mod.rs

@@ -18,7 +18,7 @@ use crate::config::{
 };
 use crate::fee_estimator::OnchainFeeEstimator;
 use crate::io::utils::write_node_metrics;
-use crate::logger::{log_info, log_trace, LdkLogger, Logger};
+use crate::logger::{log_debug, log_info, log_trace, LdkLogger, Logger};
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, NodeMetrics};
 
@@ -425,19 +425,33 @@ impl ChainSource {
 		}
 	}
 
-	pub(crate) async fn process_broadcast_queue(&self) {
+	pub(crate) async fn continuously_process_broadcast_queue(
+		&self, mut stop_tx_bcast_receiver: tokio::sync::watch::Receiver<()>,
+	) {
 		let mut receiver = self.tx_broadcaster.get_broadcast_queue().await;
-		while let Some(next_package) = receiver.recv().await {
-			match &self.kind {
-				ChainSourceKind::Esplora(esplora_chain_source) => {
-					esplora_chain_source.process_broadcast_package(next_package).await
-				},
-				ChainSourceKind::Electrum(electrum_chain_source) => {
-					electrum_chain_source.process_broadcast_package(next_package).await
-				},
-				ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
-					bitcoind_chain_source.process_broadcast_package(next_package).await
-				},
+		loop {
+			let tx_bcast_logger = Arc::clone(&self.logger);
+			tokio::select! {
+				_ = stop_tx_bcast_receiver.changed() => {
+					log_debug!(
+						tx_bcast_logger,
+						"Stopping broadcasting transactions.",
+					);
+					return;
+				}
+				Some(next_package) = receiver.recv() => {
+					match &self.kind {
+						ChainSourceKind::Esplora(esplora_chain_source) => {
+							esplora_chain_source.process_broadcast_package(next_package).await
+						},
+						ChainSourceKind::Electrum(electrum_chain_source) => {
+							electrum_chain_source.process_broadcast_package(next_package).await
+						},
+						ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
+							bitcoind_chain_source.process_broadcast_package(next_package).await
+						},
+					}
+				}
 			}
 		}
 	}

src/lib.rs

@@ -498,27 +498,10 @@ impl Node {
 			});
 		}
 
-		let mut stop_tx_bcast = self.stop_sender.subscribe();
+		let stop_tx_bcast = self.stop_sender.subscribe();
 		let chain_source = Arc::clone(&self.chain_source);
-		let tx_bcast_logger = Arc::clone(&self.logger);
 		runtime.spawn(async move {
-			// Every second we try to clear our broadcasting queue.
-			let mut interval = tokio::time::interval(Duration::from_secs(1));
-			interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
-			loop {
-				tokio::select! {
-						_ = stop_tx_bcast.changed() => {
-							log_debug!(
-								tx_bcast_logger,
-								"Stopping broadcasting transactions.",
-							);
-							return;
-						}
-						_ = interval.tick() => {
-							chain_source.process_broadcast_queue().await;
-						}
-				}
-			}
+			chain_source.continuously_process_broadcast_queue(stop_tx_bcast).await
 		});
 
 		let bump_tx_event_handler = Arc::new(BumpTransactionEventHandler::new(