Add initial polling logic

We first initialize by synchronizing all `Listen` implementations, and
then head into a loop continuously polling our RPC `BlockSource`.

We also implement a `BoundedHeaderCache` to limit in-memory footprint.

Author: tnull

src/chain/bitcoind_rpc.rs

@@ -5,14 +5,22 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
+use crate::types::{ChainMonitor, ChannelManager, Sweeper, Wallet};
+
+use lightning::chain::Listen;
+
 use lightning_block_sync::http::HttpEndpoint;
+use lightning_block_sync::poll::ValidatedBlockHeader;
 use lightning_block_sync::rpc::RpcClient;
-use lightning_block_sync::{AsyncBlockSourceResult, BlockData, BlockHeaderData, BlockSource};
+use lightning_block_sync::{
+	AsyncBlockSourceResult, BlockData, BlockHeaderData, BlockSource, Cache,
+};
 
 use bitcoin::BlockHash;
 
 use base64::prelude::{Engine, BASE64_STANDARD};
 
+use std::collections::{HashMap, VecDeque};
 use std::sync::Arc;
 
 pub struct BitcoindRpcClient {
@@ -51,3 +59,75 @@ impl BlockSource for BitcoindRpcClient {
 		Box::pin(async move { self.rpc_client.get_best_block().await })
 	}
 }
+
+const MAX_HEADER_CACHE_ENTRIES: usize = 100;
+
+pub(crate) struct BoundedHeaderCache {
+	header_map: HashMap<BlockHash, ValidatedBlockHeader>,
+	recently_seen: VecDeque<BlockHash>,
+}
+
+impl BoundedHeaderCache {
+	pub(crate) fn new() -> Self {
+		let header_map = HashMap::new();
+		let recently_seen = VecDeque::new();
+		Self { header_map, recently_seen }
+	}
+}
+
+impl Cache for BoundedHeaderCache {
+	fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
+		self.header_map.get(block_hash)
+	}
+
+	fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader) {
+		self.recently_seen.push_back(block_hash);
+		self.header_map.insert(block_hash, block_header);
+
+		if self.header_map.len() >= MAX_HEADER_CACHE_ENTRIES {
+			// Keep dropping old entries until we've actually removed a header entry.
+			while let Some(oldest_entry) = self.recently_seen.pop_front() {
+				if self.header_map.remove(&oldest_entry).is_some() {
+					break;
+				}
+			}
+		}
+	}
+
+	fn block_disconnected(&mut self, block_hash: &BlockHash) -> Option<ValidatedBlockHeader> {
+		self.recently_seen.retain(|e| e != block_hash);
+		self.header_map.remove(block_hash)
+	}
+}
+
+pub(crate) struct ChainListener {
+	pub(crate) onchain_wallet: Arc<Wallet>,
+	pub(crate) channel_manager: Arc<ChannelManager>,
+	pub(crate) chain_monitor: Arc<ChainMonitor>,
+	pub(crate) output_sweeper: Arc<Sweeper>,
+}
+
+impl Listen for ChainListener {
+	fn filtered_block_connected(
+		&self, header: &bitcoin::block::Header,
+		txdata: &lightning::chain::transaction::TransactionData, height: u32,
+	) {
+		self.onchain_wallet.filtered_block_connected(header, txdata, height);
+		self.channel_manager.filtered_block_connected(header, txdata, height);
+		self.chain_monitor.filtered_block_connected(header, txdata, height);
+		self.output_sweeper.filtered_block_connected(header, txdata, height);
+	}
+	fn block_connected(&self, block: &bitcoin::Block, height: u32) {
+		self.onchain_wallet.block_connected(block, height);
+		self.channel_manager.block_connected(block, height);
+		self.chain_monitor.block_connected(block, height);
+		self.output_sweeper.block_connected(block, height);
+	}
+
+	fn block_disconnected(&self, header: &bitcoin::block::Header, height: u32) {
+		self.onchain_wallet.block_disconnected(header, height);
+		self.channel_manager.block_disconnected(header, height);
+		self.chain_monitor.block_disconnected(header, height);
+		self.output_sweeper.block_disconnected(header, height);
+	}
+}