Update BlockNotifier to accept Derefs as registered listeners.

Author: valentinewallace

lightning/src/chain/chaininterface.rs

@@ -14,9 +14,11 @@ use bitcoin::network::constants::Network;
 
 use util::logger::Logger;
 
-use std::sync::{Mutex,Weak,MutexGuard,Arc};
+use std::sync::{Mutex, MutexGuard, Arc};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::collections::HashSet;
+use std::ops::Deref;
+use std::marker::Sized;
 
 /// Used to give chain error details upstream
 pub enum ChainError {
@@ -207,24 +209,23 @@ impl ChainWatchedUtil {
 
 /// Utility for notifying listeners about new blocks, and handling block rescans if new watch
 /// data is registered.
-pub struct BlockNotifier {
-	listeners: Mutex<Vec<Weak<ChainListener>>>, //TODO(vmw): try removing Weak
+pub struct BlockNotifier<CL: Deref<Target = ChainListener> + Sized + Clone> {
+	listeners: Mutex<Vec<CL>>,
 	chain_monitor: Arc<ChainWatchInterface>,
 }
 
-impl BlockNotifier {
+impl<CL: Deref<Target = ChainListener> + Sized + Clone> BlockNotifier<CL> {
 	/// Constructs a new BlockNotifier without any listeners.
-	pub fn new(chain_monitor: Arc<ChainWatchInterface>) -> BlockNotifier {
+	pub fn new(chain_monitor: Arc<ChainWatchInterface>) -> BlockNotifier<CL> {
 		BlockNotifier {
 			listeners: Mutex::new(Vec::new()),
 			chain_monitor,
 		}
 	}
 
-	/// Register the given listener to receive events. Only a weak pointer is provided and
-	/// the registration should be freed once that pointer expires.
+	/// Register the given listener to receive events.
 	// TODO: unregister
-	pub fn register_listener(&self, listener: Weak<ChainListener>) {
+	pub fn register_listener(&self, listener: CL) {
 		let mut vec = self.listeners.lock().unwrap();
 		vec.push(listener);
 	}
@@ -252,10 +253,7 @@ impl BlockNotifier {
 
 		let listeners = self.listeners.lock().unwrap().clone();
 		for listener in listeners.iter() {
-			match listener.upgrade() {
-				Some(arc) => arc.block_connected(header, height, txn_matched, indexes_of_txn_matched),
-				None => ()
-			}
+			listener.block_connected(header, height, txn_matched, indexes_of_txn_matched)
 		}
 		return last_seen != self.chain_monitor.reentered();
 	}
@@ -265,10 +263,7 @@ impl BlockNotifier {
 	pub fn block_disconnected(&self, header: &BlockHeader, disconnected_height: u32) {
 		let listeners = self.listeners.lock().unwrap().clone();
 		for listener in listeners.iter() {
-			match listener.upgrade() {
-				Some(arc) => arc.block_disconnected(&header, disconnected_height),
-				None => ()
-			}
+			listener.block_disconnected(&header, disconnected_height);
 		}
 	}
 

lightning/src/ln/functional_test_utils.rs

@@ -35,10 +35,9 @@ use std::rc::Rc;
 use std::sync::{Arc, Mutex};
 use std::mem;
 use std::ops::Deref;
-use std::marker::Sized;
 
 pub const CHAN_CONFIRM_DEPTH: u32 = 100;
-pub fn confirm_transaction(notifier: &chaininterface::BlockNotifier, chain: &chaininterface::ChainWatchInterfaceUtil, tx: &Transaction, chan_id: u32) {
+pub fn confirm_transaction<CL: Deref<Target = chaininterface::ChainListener> + Clone>(notifier: &chaininterface::BlockNotifier<CL>, chain: &chaininterface::ChainWatchInterfaceUtil, tx: &Transaction, chan_id: u32) {
 	assert!(chain.does_match_tx(tx));
 	let mut header = BlockHeader { version: 0x20000000, prev_blockhash: Default::default(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
 	notifier.block_connected_checked(&header, 1, &[tx; 1], &[chan_id; 1]);
@@ -48,7 +47,7 @@ pub fn confirm_transaction(notifier: &chaininterface::BlockNotifier, chain: &cha
 	}
 }
 
-pub fn connect_blocks(notifier: &chaininterface::BlockNotifier, depth: u32, height: u32, parent: bool, prev_blockhash: Sha256d) -> Sha256d {
+pub fn connect_blocks<CL: Deref<Target = chaininterface::ChainListener> + Clone>(notifier: &chaininterface::BlockNotifier<CL>, depth: u32, height: u32, parent: bool, prev_blockhash: Sha256d) -> Sha256d {
 	let mut header = BlockHeader { version: 0x2000000, prev_blockhash: if parent { prev_blockhash } else { Default::default() }, merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
 	notifier.block_connected_checked(&header, height + 1, &Vec::new(), &Vec::new());
 	for i in 2..depth + 1 {
@@ -59,7 +58,7 @@ pub fn connect_blocks(notifier: &chaininterface::BlockNotifier, depth: u32, heig
 }
 
 pub struct Node {
-	pub block_notifier: Arc<chaininterface::BlockNotifier>,
+	pub block_notifier: Arc<chaininterface::BlockNotifier<Arc<chaininterface::ChainListener>>>,
 	pub chain_monitor: Arc<chaininterface::ChainWatchInterfaceUtil>,
 	pub tx_broadcaster: Arc<test_utils::TestBroadcaster>,
 	pub chan_monitor: Arc<test_utils::TestChannelMonitor>,
@@ -858,14 +857,12 @@ pub fn create_network(node_count: usize, node_config: &[Option<UserConfig>]) ->
 		rng.fill_bytes(&mut seed);
 		let keys_manager = Arc::new(test_utils::TestKeysInterface::new(&seed, Network::Testnet, Arc::clone(&logger)));
 		let chan_monitor = Arc::new(test_utils::TestChannelMonitor::new(chain_monitor.clone(), tx_broadcaster.clone(), logger.clone(), feeest.clone()));
-		let weak_res = Arc::downgrade(&chan_monitor.simple_monitor);
-		block_notifier.register_listener(weak_res);
+		block_notifier.register_listener(chan_monitor.simple_monitor.clone() as Arc<chaininterface::ChainListener>);
 		let mut default_config = UserConfig::default();
 		default_config.channel_options.announced_channel = true;
 		default_config.peer_channel_config_limits.force_announced_channel_preference = false;
 		let node = ChannelManager::new(Network::Testnet, feeest.clone(), chan_monitor.clone() as Arc<ManyChannelMonitor>, tx_broadcaster.clone(), Arc::clone(&logger), keys_manager.clone(), if node_config[i].is_some() { node_config[i].clone().unwrap() } else { default_config }, 0).unwrap();
-		let weak_res = Arc::downgrade(&node);
-		block_notifier.register_listener(weak_res);
+		block_notifier.register_listener(node.clone());
 		let router = Router::new(PublicKey::from_secret_key(&secp_ctx, &keys_manager.get_node_secret()), chain_monitor.clone(), Arc::clone(&logger));
 		nodes.push(Node { chain_monitor, tx_broadcaster, chan_monitor, node, router, keys_manager, node_seed: seed,
 			network_payment_count: payment_count.clone(),

lightning/src/ln/functional_tests.rs

@@ -3,7 +3,7 @@
 //! claim outputs on-chain.
 
 use chain::transaction::OutPoint;
-use chain::chaininterface::{ChainListener, ChainWatchInterfaceUtil};
+use chain::chaininterface::{ChainListener, ChainWatchInterfaceUtil, BlockNotifier};
 use chain::keysinterface::{KeysInterface, SpendableOutputDescriptor};
 use ln::channel::{COMMITMENT_TX_BASE_WEIGHT, COMMITMENT_TX_WEIGHT_PER_HTLC};
 use ln::channelmanager::{ChannelManager,ChannelManagerReadArgs,HTLCForwardInfo,RAACommitmentOrder, PaymentPreimage, PaymentHash, BREAKDOWN_TIMEOUT};
@@ -44,7 +44,6 @@ use std::default::Default;
 use std::sync::{Arc, Mutex};
 use std::sync::atomic::Ordering;
 use std::mem;
-use std::ops::Deref;
 
 use rand::{thread_rng, Rng};
 
@@ -3447,7 +3446,7 @@ fn test_no_txn_manager_serialize_deserialize() {
 	assert!(nodes[0].chan_monitor.add_update_monitor(chan_0_monitor.get_funding_txo().unwrap(), chan_0_monitor).is_ok());
 	nodes[0].node = Arc::new(nodes_0_deserialized);
 	let nodes_0_as_listener: Arc<ChainListener> = nodes[0].node.clone();
-	nodes[0].block_notifier.register_listener(Arc::downgrade(&nodes_0_as_listener));
+	nodes[0].block_notifier.register_listener(nodes_0_as_listener);
 	assert_eq!(nodes[0].node.list_channels().len(), 1);
 	check_added_monitors!(nodes[0], 1);
 
@@ -6147,11 +6146,10 @@ fn test_data_loss_protect() {
 	assert!(monitor.add_update_monitor(OutPoint { txid: chan.3.txid(), index: 0 }, chan_monitor.clone()).is_ok());
 	nodes[0].chan_monitor = monitor;
 	nodes[0].chain_monitor = chain_monitor;
+	nodes[0].block_notifier = Arc::new(BlockNotifier::new(nodes[0].chain_monitor.clone()));
 
-	let weak_res = Arc::downgrade(&nodes[0].chan_monitor.simple_monitor);
-	nodes[0].block_notifier.register_listener(weak_res);
-	let weak_res = Arc::downgrade(&nodes[0].node);
-	nodes[0].block_notifier.register_listener(weak_res);
+	nodes[0].block_notifier.register_listener(nodes[0].chan_monitor.simple_monitor.clone());
+	nodes[0].block_notifier.register_listener(nodes[0].node.clone());
 
 	check_added_monitors!(nodes[0], 1);