Implement async versions of process_pending_events

[rust-lightning] / lightning / src / ln / channelmanager.rs

diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 7f8c82b41bdb81e846e9f22c3ab871ace41b0260..06ef2aa0dc3422fc457ee05504c395e96f00eedc 100644 (file)
--- a/lightning/src/ln/channelmanager.rs
+++ b/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;
@@ -5728,6 +5728,39 @@ 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 pending_events = mem::replace(&mut *self.pending_events.lock().unwrap(), vec![]);
+               if !pending_events.is_empty() {
+                       result = NotifyOption::DoPersist;
+               }
+
+               for event in pending_events {
+                       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>