f WIP

Author: tnull

Cargo.toml

@@ -77,6 +77,7 @@ rand = "0.8.5"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync", "macros" ] }
 esplora-client = { version = "0.11", default-features = false, features = ["tokio", "async-https-rustls"] }
+electrum-client = { version = "0.22.0", default-features = true }
 libc = "0.2"
 uniffi = { version = "0.27.3", features = ["build"], optional = true }
 serde = { version = "1.0.210", default-features = false, features = ["std", "derive"] }

src/chain/electrum.rs

@@ -6,39 +6,44 @@
 // accordance with one or both of these licenses.
 
 use crate::error::Error;
-use crate::logger::{log_error, LdkLogger, Logger};
+use crate::logger::Logger;
 
-use bitcoin::{FeeRate, Transaction};
+use electrum_client::{Batch, Client, ElectrumApi};
 
 use tokio::sync::oneshot;
 
 use std::collections::VecDeque;
+use std::mem;
 use std::sync::{Arc, Condvar, Mutex};
 use std::thread;
+use std::time::{Duration, SystemTime};
 
 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;
+const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
 
 enum SyncCommand {
 	Stop,
+	EstimateFees { targets: Vec<usize>, result_sender: SyncCommandResultSender<Vec<f64>> },
 }
 
 struct ElectrumSyncActor {
+	server_url: String,
 	command_queue: Arc<SyncCommandQueue>,
 	command_queue_condvar: Arc<Condvar>,
-	worker_threads: Mutex<Vec<thread::JoinHandle<()>>>,
+	worker_threads: Mutex<Vec<thread::JoinHandle<Result<(), Error>>>>,
 	logger: Arc<Logger>,
 }
 
 impl ElectrumSyncActor {
-	pub(crate) fn new(logger: Arc<Logger>) -> Self {
+	pub(crate) fn new(server_url: String, 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 }
+		Self { server_url, command_queue, command_queue_condvar, worker_threads, logger }
 	}
 
 	pub(crate) fn start(&self) -> Result<(), Error> {
@@ -47,8 +52,9 @@ impl ElectrumSyncActor {
 			let command_queue = Arc::clone(&self.command_queue);
 			let command_queue_condvar = Arc::clone(&self.command_queue_condvar);
 
+			let server_url = self.server_url.clone();
 			let handle = std::thread::spawn(move || {
-				queue_processing_loop(command_queue, command_queue_condvar);
+				queue_processing_loop(server_url, command_queue, command_queue_condvar)
 			});
 			worker_threads.push(handle);
 		}
@@ -62,26 +68,48 @@ impl ElectrumSyncActor {
 			let cmd = SyncCommand::Stop;
 			cmd_queue.push_back(cmd);
 		}
-		self.command_queue_condvar.notify_one();
+		self.command_queue_condvar.notify_all();
 
 		{
-			let mut worker_threads = self.worker_threads.lock().unwrap();
-			for handle in worker_threads.drain(..) {
-				handle.join().expect("TODO");
+			let mut worker_threads_lock = self.worker_threads.lock().unwrap();
+			let mut worker_threads = mem::take(&mut *worker_threads_lock);
+			let all_finished = worker_threads.iter().all(|h| h.is_finished());
+
+			if all_finished {
+				for handle in worker_threads.drain(..) {
+					if handle.is_finished() {
+						handle.join().expect("TODO");
+					}
+				}
 			}
 		}
 		Ok(())
 	}
+
+	pub(crate) async fn estimate_fees(&self, targets: Vec<usize>) -> Result<Vec<f64>, Error> {
+		let (result_sender, result_receiver) = oneshot::channel();
+		{
+			let mut cmd_queue = self.command_queue.lock().unwrap();
+			let cmd = SyncCommand::EstimateFees { targets, result_sender };
+			cmd_queue.push_back(cmd);
+		}
+		self.command_queue_condvar.notify_all();
+		// TODO:
+		result_receiver.await.unwrap()
+	}
 }
 
 fn queue_processing_loop(
-	command_queue: Arc<SyncCommandQueue>, command_queue_condvar: Arc<Condvar>,
-) {
-	loop {
+	server_url: String, command_queue: Arc<SyncCommandQueue>, command_queue_condvar: Arc<Condvar>,
+) -> Result<(), Error> {
+	let client = Client::new(&server_url).map_err(|e| Error::TxSyncFailed)?;
+
+	'worker_loop: 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() {
+
+		let next_cmd = match queue_lock.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.");
@@ -90,8 +118,25 @@ fn queue_processing_loop(
 		};
 
 		match next_cmd {
-			SyncCommand::Stop => break,
+			SyncCommand::Stop => {
+				// Early-abort on `Stop` leaving it the front of the queue so all other workers
+				// will also try to handle it.
+				break 'worker_loop;
+			},
+			SyncCommand::EstimateFees { targets, result_sender } => {
+				let mut batch = Batch::default();
+				for t in targets {
+					batch.estimate_fee(*t)
+				}
+				let res = client
+					.batch_call(&batch)
+					.map(|r| r.into_iter().map(|v| v.as_f64()))
+					.map_err(|_| Error::TxSyncFailed);
+			},
 			_ => {},
 		}
+
+		queue_lock.pop_front();
 	}
+	Ok(())
 }