Implement async versions of process_pending_events

Author: wpaulino

lightning/src/chain/chainmonitor.rs

@@ -36,7 +36,7 @@ use crate::util::atomic_counter::AtomicCounter;
 use crate::util::logger::Logger;
 use crate::util::errors::APIError;
 use crate::util::events;
-use crate::util::events::EventHandler;
+use crate::util::events::{Event, EventHandler};
 use crate::ln::channelmanager::ChannelDetails;
 
 use crate::prelude::*;
@@ -479,6 +479,22 @@ where C::Target: chain::Filter,
 		self.process_pending_events(&event_handler);
 		events.into_inner()
 	}
+
+	/// Processes any events asynchronously in the order they were generated since the last call
+	/// using the given event handler.
+	///
+	/// See the trait-level documentation of [`EventsProvider`] for requirements.
+	///
+	/// [`EventsProvider`]: crate::util::events::EventsProvider
+	pub async fn process_pending_events_async<Future: core::future::Future, H: Fn(Event) -> Future>(&self, handler: H) {
+		let mut pending_events = Vec::new();
+		for monitor_state in self.monitors.read().unwrap().values() {
+			pending_events.append(&mut monitor_state.monitor.get_and_clear_pending_events());
+		}
+		for event in pending_events.drain(..) {
+			handler(event).await;
+		}
+	}
 }
 
 impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>

lightning/src/ln/channelmanager.rs

@@ -53,7 +53,7 @@ use crate::ln::msgs::{ChannelMessageHandler, DecodeError, LightningError, MAX_VA
 use crate::ln::wire::Encode;
 use crate::chain::keysinterface::{Sign, KeysInterface, KeysManager, Recipient};
 use crate::util::config::{UserConfig, ChannelConfig};
-use crate::util::events::{EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider, ClosureReason, HTLCDestination};
+use crate::util::events::{Event, EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider, ClosureReason, HTLCDestination};
 use crate::util::{byte_utils, events};
 use crate::util::wakers::{Future, Notifier};
 use crate::util::scid_utils::fake_scid;
@@ -5588,6 +5588,37 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 	pub fn clear_pending_payments(&self) {
 		self.pending_outbound_payments.lock().unwrap().clear()
 	}
+
+	/// Processes any events asynchronously in the order they were generated since the last call
+	/// using the given event handler.
+	///
+	/// See the trait-level documentation of [`EventsProvider`] for requirements.
+	pub async fn process_pending_events_async<Future: core::future::Future, H: Fn(Event) -> Future>(&self, handler: H) {
+		// We'll acquire our total consistency lock until the returned future completes so that
+		// we can be sure no other persists happen while processing events.
+		let _read_guard = self.total_consistency_lock.read().unwrap();
+
+		let mut result = NotifyOption::SkipPersist;
+
+		// TODO: This behavior should be documented. It's unintuitive that we query
+		// ChannelMonitors when clearing other events.
+		if self.process_pending_monitor_events() {
+			result = NotifyOption::DoPersist;
+		}
+
+		let mut pending_events = mem::replace(&mut *self.pending_events.lock().unwrap(), vec![]);
+		if !pending_events.is_empty() {
+			result = NotifyOption::DoPersist;
+		}
+
+		for event in pending_events.drain(..) {
+			handler(event).await;
+		}
+
+		if result == NotifyOption::DoPersist {
+			self.persistence_notifier.notify();
+		}
+	}
 }
 
 impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> MessageSendEventsProvider for ChannelManager<M, T, K, F, L>