Restore event queue from disk in setup

Author: tnull

src/lib.rs

@@ -34,7 +34,7 @@ mod logger;
 use access::LdkLiteChainAccess;
 pub use error::LdkLiteError as Error;
 pub use event::LdkLiteEvent;
-use event::LdkLiteEventHandler;
+use event::{LdkLiteEventHandler, LdkLiteEventQueue};
 
 #[allow(unused_imports)]
 use logger::{
@@ -81,16 +81,12 @@ use std::collections::HashMap;
 use std::fs;
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc;
 use std::sync::{Arc, Mutex, RwLock};
 use std::thread;
 use std::time::{Duration, SystemTime};
 
 // TODO: Is MemoryDatabase okay to use?
 
-// The number of messages we buffer in the used channels.
-const CHANNEL_BUF_SIZE: usize = 1000;
-
 // The used 'stop gap' parameter used by BDK's wallet sync. This seems to configure the threshold
 // number of blocks after which BDK stops looking for scripts belonging to the wallet.
 const BDK_CLIENT_STOP_GAP: usize = 20;
@@ -328,17 +324,23 @@ impl LdkLiteBuilder {
 		let inbound_payments = Arc::new(Mutex::new(HashMap::new()));
 		let outbound_payments = Arc::new(Mutex::new(HashMap::new()));
 
-		// Step 14: Handle LDK Events
-		let event_queue = mpsc::sync_channel(CHANNEL_BUF_SIZE);
-		let event_sender = event_queue.0.clone();
+		// Step 14: Restore event handler from disk or create a new one.
+		let event_queue = if let Ok(mut f) =
+			fs::File::open(format!("{}/{}", ldk_data_dir.clone(), event::EVENTS_PERSISTENCE_KEY))
+		{
+			Arc::new(LdkLiteEventQueue::read(&mut f, Arc::clone(&persister))?)
+		} else {
+			Arc::new(LdkLiteEventQueue::new(Arc::clone(&persister)))
+		};
+
 		let event_handler = LdkLiteEventHandler::new(
 			Arc::clone(&chain_access),
+			Arc::clone(&event_queue),
 			Arc::clone(&channel_manager),
 			Arc::clone(&network_graph),
 			Arc::clone(&keys_manager),
 			Arc::clone(&inbound_payments),
 			Arc::clone(&outbound_payments),
-			event_sender,
 			Arc::clone(&logger),
 			Arc::clone(&config),
 		);
@@ -365,6 +367,7 @@ impl LdkLiteBuilder {
 			running,
 			config,
 			chain_access,
+			event_queue,
 			channel_manager,
 			chain_monitor,
 			peer_manager,
@@ -377,7 +380,6 @@ impl LdkLiteBuilder {
 			invoice_payer,
 			inbound_payments,
 			outbound_payments,
-			event_queue,
 		})
 	}
 }
@@ -398,6 +400,7 @@ pub struct LdkLite {
 	running: RwLock<Option<LdkLiteRuntime>>,
 	config: Arc<LdkLiteConfig>,
 	chain_access: Arc<LdkLiteChainAccess<MemoryDatabase>>,
+	event_queue: Arc<LdkLiteEventQueue<FilesystemPersister>>,
 	channel_manager: Arc<ChannelManager>,
 	chain_monitor: Arc<ChainMonitor>,
 	peer_manager: Arc<PeerManager>,
@@ -410,7 +413,6 @@ pub struct LdkLite {
 	invoice_payer: Arc<InvoicePayer<LdkLiteEventHandler>>,
 	inbound_payments: Arc<PaymentInfoStorage>,
 	outbound_payments: Arc<PaymentInfoStorage>,
-	event_queue: (EventSender, EventReceiver),
 }
 
 impl LdkLite {
@@ -579,9 +581,16 @@ impl LdkLite {
 		})
 	}
 
-	/// Returns the next event from the event queue. Blocks until a new event is available.
-	pub fn next_event(&self) -> Result<LdkLiteEvent, Error> {
-		Ok(self.event_queue.1.recv()?)
+	/// Blocks until the next event is available.
+	///
+	/// Note: this will always return the same event until handling is confirmed via [`event_handled`].
+	pub fn next_event(&self) -> LdkLiteEvent {
+		self.event_queue.next_event()
+	}
+
+	/// Confirm the last retrieved event handled.
+	pub fn event_handled(&self) -> Result<(), Error> {
+		self.event_queue.event_handled()
 	}
 
 	/// Returns our own node id
@@ -955,8 +964,5 @@ pub(crate) type NetworkGraph = gossip::NetworkGraph<Arc<FilesystemLogger>>;
 
 pub(crate) type PaymentInfoStorage = Mutex<HashMap<PaymentHash, PaymentInfo>>;
 
-pub(crate) type EventSender = mpsc::SyncSender<LdkLiteEvent>;
-pub(crate) type EventReceiver = mpsc::Receiver<LdkLiteEvent>;
-
 #[cfg(test)]
 mod tests {}