Add PeerManager::process_pending_events_async

Author: jkczyz

lightning/src/events/mod.rs

@@ -1885,9 +1885,11 @@ pub trait MessageSendEventsProvider {
 ///
 /// Events are processed by passing an [`EventHandler`] to [`process_pending_events`].
 ///
-/// Implementations of this trait may also feature an async version of event handling, as shown with
-/// [`ChannelManager::process_pending_events_async`] and
-/// [`ChainMonitor::process_pending_events_async`].
+/// Implementations of this trait may also feature an async version of event handling, as shown
+/// with:
+/// - [`ChannelManager::process_pending_events_async`]
+/// - [`ChainMonitor::process_pending_events_async`]
+/// - [`PeerManager::process_pending_events_async`]
 ///
 /// # Requirements
 ///
@@ -1917,6 +1919,7 @@ pub trait MessageSendEventsProvider {
 /// [`ChainMonitor::process_pending_events`]: crate::chain::chainmonitor::ChainMonitor#method.process_pending_events
 /// [`ChannelManager::process_pending_events_async`]: crate::ln::channelmanager::ChannelManager::process_pending_events_async
 /// [`ChainMonitor::process_pending_events_async`]: crate::chain::chainmonitor::ChainMonitor::process_pending_events_async
+/// [`PeerManager::process_pending_events_async`]: crate::ln::peer_handler::PeerManager::process_pending_events_async
 pub trait EventsProvider {
 	/// Processes any events generated since the last call using the given event handler.
 	///

lightning/src/ln/peer_handler.rs

@@ -2478,6 +2478,19 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 		let _ = self.message_handler.route_handler.handle_node_announcement(&msg);
 		self.forward_broadcast_msg(&*self.peers.read().unwrap(), &wire::Message::NodeAnnouncement(msg), None);
 	}
+
+	/// 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
+	) {
+		let pending_events = core::mem::take(&mut *self.pending_events.lock().unwrap());
+		for event in pending_events {
+			handler(event).await;
+		}
+	}
 }
 
 fn is_gossip_msg(type_id: u16) -> bool {