Impl event queue persistence

Author: tnull

src/event.rs

@@ -1,5 +1,5 @@
 use crate::{
-	ChannelManager, EventSender, LdkLiteChainAccess, LdkLiteConfig, NetworkGraph,
+	ChannelManager, Error, FilesystemPersister, LdkLiteChainAccess, LdkLiteConfig, NetworkGraph,
 	PaymentInfoStorage,
 };
 
@@ -12,48 +12,153 @@ use crate::logger::{
 use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
 use lightning::chain::keysinterface::KeysManager;
 use lightning::util::events as ldk_events;
+use lightning::util::persist::KVStorePersister;
+use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 
 use bdk::database::MemoryDatabase;
 
 use bitcoin::secp256k1::Secp256k1;
-//use std::collections::{hash_map, VecDeque};
-//use std::iter::Iterator;
-use std::sync::Arc;
+use std::collections::VecDeque;
+use std::sync::{Arc, Condvar, Mutex};
+
+/// The event queue will be persisted under this key.
+pub(crate) const EVENTS_PERSISTENCE_KEY: &str = "events";
 
 /// An LdkLiteEvent that should be handled by the user.
+#[derive(Debug, Clone)]
 pub enum LdkLiteEvent {
 	/// asdf
 	Test,
 }
 
+impl Readable for LdkLiteEvent {
+	fn read<R: lightning::io::Read>(
+		reader: &mut R,
+	) -> Result<Self, lightning::ln::msgs::DecodeError> {
+		match Readable::read(reader)? {
+			// TODO
+			0u8 => Ok(LdkLiteEvent::Test),
+			_ => Ok(LdkLiteEvent::Test),
+		}
+	}
+}
+
+impl Writeable for LdkLiteEvent {
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), lightning::io::Error> {
+		match self {
+			Test => {
+				// TODO
+				0u8.write(writer)?;
+				Ok(())
+			}
+		}
+	}
+}
+
+pub(crate) struct LdkLiteEventQueue<K: KVStorePersister> {
+	queue: Mutex<EventQueueSerWrapper>,
+	notifier: Condvar,
+	persister: Arc<K>,
+}
+
+impl<K: KVStorePersister> LdkLiteEventQueue<K> {
+	pub(crate) fn new(persister: Arc<K>) -> Self {
+		let queue: Mutex<EventQueueSerWrapper> = Mutex::new(EventQueueSerWrapper(VecDeque::new()));
+		let notifier = Condvar::new();
+		Self { queue, notifier, persister }
+	}
+	pub(crate) fn add_event(&self, event: LdkLiteEvent) -> Result<(), Error> {
+		let mut locked_queue = self.queue.lock().unwrap();
+		locked_queue.0.push_back(event);
+
+		self.persister.persist(EVENTS_PERSISTENCE_KEY, &*locked_queue)?;
+
+		self.notifier.notify_one();
+		Ok(())
+	}
+
+	pub(crate) fn next_event(&self) -> LdkLiteEvent {
+		let locked_queue = self
+			.notifier
+			.wait_while(self.queue.lock().unwrap(), |queue| queue.0.is_empty())
+			.unwrap();
+		locked_queue.0.front().unwrap().clone()
+	}
+
+	pub(crate) fn event_handled(&self) -> Result<(), Error> {
+		let mut locked_queue = self.queue.lock().unwrap();
+		locked_queue.0.pop_front();
+		self.persister.persist(EVENTS_PERSISTENCE_KEY, &*locked_queue)?;
+		self.notifier.notify_one();
+		Ok(())
+	}
+}
+
+impl<K: KVStorePersister> ReadableArgs<Arc<K>> for LdkLiteEventQueue<K> {
+	#[inline]
+	fn read<R: lightning::io::Read>(
+		reader: &mut R, persister: Arc<K>,
+	) -> Result<Self, lightning::ln::msgs::DecodeError> {
+		let queue: Mutex<EventQueueSerWrapper> = Mutex::new(Readable::read(reader)?);
+		let notifier = Condvar::new();
+		Ok(Self { queue, notifier, persister })
+	}
+}
+
+struct EventQueueSerWrapper(VecDeque<LdkLiteEvent>);
+
+impl Readable for EventQueueSerWrapper {
+	fn read<R: lightning::io::Read>(
+		reader: &mut R,
+	) -> Result<Self, lightning::ln::msgs::DecodeError> {
+		let len: u16 = Readable::read(reader)?;
+		let mut queue = VecDeque::with_capacity(len as usize);
+		for _ in 0..len {
+			queue.push_back(Readable::read(reader)?);
+		}
+		Ok(EventQueueSerWrapper(queue))
+	}
+}
+
+impl Writeable for EventQueueSerWrapper {
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), lightning::io::Error> {
+		(self.0.len() as u16).write(writer)?;
+		for e in self.0.iter() {
+			e.write(writer)?;
+		}
+		Ok(())
+	}
+}
+
 pub(crate) struct LdkLiteEventHandler {
 	chain_access: Arc<LdkLiteChainAccess<MemoryDatabase>>,
+	event_queue: Arc<LdkLiteEventQueue<FilesystemPersister>>,
 	channel_manager: Arc<ChannelManager>,
 	_network_graph: Arc<NetworkGraph>,
 	keys_manager: Arc<KeysManager>,
 	_inbound_payments: Arc<PaymentInfoStorage>,
 	_outbound_payments: Arc<PaymentInfoStorage>,
-	_event_sender: EventSender,
 	logger: Arc<FilesystemLogger>,
 	_config: Arc<LdkLiteConfig>,
 }
 
 impl LdkLiteEventHandler {
 	pub fn new(
 		chain_access: Arc<LdkLiteChainAccess<MemoryDatabase>>,
+		event_queue: Arc<LdkLiteEventQueue<FilesystemPersister>>,
 		channel_manager: Arc<ChannelManager>, _network_graph: Arc<NetworkGraph>,
 		keys_manager: Arc<KeysManager>, _inbound_payments: Arc<PaymentInfoStorage>,
-		_outbound_payments: Arc<PaymentInfoStorage>, _event_sender: EventSender,
-		logger: Arc<FilesystemLogger>, _config: Arc<LdkLiteConfig>,
+		_outbound_payments: Arc<PaymentInfoStorage>, logger: Arc<FilesystemLogger>,
+		_config: Arc<LdkLiteConfig>,
 	) -> Self {
 		Self {
+			event_queue,
 			chain_access,
 			channel_manager,
 			_network_graph,
 			keys_manager,
 			_inbound_payments,
 			_outbound_payments,
-			_event_sender,
 			logger,
 			_config,
 		}