Add ElectrumSyncActor struct

We add an `ElectrumSyncActor` that follows the actor pattern (hence the
name) to isolate the blocking/threaded API of `rust-electrum-client`
from our async internals.

Author: tnull

src/chain/electrum.rs

@@ -0,0 +1,97 @@
+// This file is Copyright its original authors, visible in version control history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license <LICENSE-MIT or
+// 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::error::Error;
+use crate::logger::{log_error, LdkLogger, Logger};
+
+use bitcoin::{FeeRate, Transaction};
+
+use tokio::sync::oneshot;
+
+use std::collections::VecDeque;
+use std::sync::{Arc, Condvar, Mutex};
+use std::thread;
+
+type SyncCommandResultSender<T> = oneshot::Sender<Result<T, Error>>;
+type SyncCommandResultReceiver<T> = oneshot::Receiver<Result<T, Error>>;
+type SyncCommandQueue = Mutex<VecDeque<SyncCommand>>;
+
+const NUM_WORKER_THREADS: usize = 2;
+
+enum SyncCommand {
+	Stop,
+}
+
+struct ElectrumSyncActor {
+	command_queue: Arc<SyncCommandQueue>,
+	command_queue_condvar: Arc<Condvar>,
+	worker_threads: Mutex<Vec<thread::JoinHandle<()>>>,
+	logger: Arc<Logger>,
+}
+
+impl ElectrumSyncActor {
+	pub(crate) fn new(logger: Arc<Logger>) -> Self {
+		let command_queue = Arc::new(Mutex::new(VecDeque::new()));
+		let command_queue_condvar = Arc::new(Condvar::new());
+		let worker_threads = Mutex::new(Vec::with_capacity(NUM_WORKER_THREADS));
+		Self { command_queue, command_queue_condvar, worker_threads, logger }
+	}
+
+	pub(crate) fn start(&self) -> Result<(), Error> {
+		let mut worker_threads = self.worker_threads.lock().unwrap();
+		for _ in 0..NUM_WORKER_THREADS {
+			let command_queue = Arc::clone(&self.command_queue);
+			let command_queue_condvar = Arc::clone(&self.command_queue_condvar);
+
+			let handle = std::thread::spawn(move || {
+				queue_processing_loop(command_queue, command_queue_condvar);
+			});
+			worker_threads.push(handle);
+		}
+
+		Ok(())
+	}
+
+	pub(crate) async fn stop(&self) -> Result<(), Error> {
+		{
+			let mut cmd_queue = self.command_queue.lock().unwrap();
+			let cmd = SyncCommand::Stop;
+			cmd_queue.push_back(cmd);
+		}
+		self.command_queue_condvar.notify_one();
+
+		{
+			let mut worker_threads = self.worker_threads.lock().unwrap();
+			for handle in worker_threads.drain(..) {
+				handle.join().expect("TODO");
+			}
+		}
+		Ok(())
+	}
+}
+
+fn queue_processing_loop(
+	command_queue: Arc<SyncCommandQueue>, command_queue_condvar: Arc<Condvar>,
+) {
+	loop {
+		let mut queue_lock = command_queue_condvar
+			.wait_while(command_queue.lock().unwrap(), |queue| queue.is_empty())
+			.unwrap();
+		let next_cmd = match queue_lock.pop_front() {
+			Some(cmd) => cmd,
+			None => {
+				debug_assert!(false, "We just checked the queue is never empty, so we should always have an entry to process.");
+				continue;
+			},
+		};
+
+		match next_cmd {
+			SyncCommand::Stop => break,
+			_ => {},
+		}
+	}
+}

src/chain/mod.rs

@@ -6,6 +6,7 @@
 // accordance with one or both of these licenses.
 
 mod bitcoind_rpc;
+mod electrum;
 
 use crate::chain::bitcoind_rpc::{
 	BitcoindRpcClient, BoundedHeaderCache, ChainListener, FeeRateEstimationMode,