Merge pull request #1806 from arik-so/2022-10-background-processor-deparametrization

[rust-lightning] / lightning / src / chain / chainmonitor.rs

diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs
index 00e25ae9db1dc9a66095b280b67323ff6c4f3de1..17fe69182ee81d060bc5c7c8c092608770104c84 100644 (file)
--- a/lightning/src/chain/chainmonitor.rs
+++ b/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::*;
@@ -395,6 +395,23 @@ where C::Target: chain::Filter,
                self.monitors.read().unwrap().keys().map(|outpoint| *outpoint).collect()
        }
 
+       #[cfg(not(c_bindings))]
+       /// Lists the pending updates for each [`ChannelMonitor`] (by `OutPoint` being monitored).
+       pub fn list_pending_monitor_updates(&self) -> HashMap<OutPoint, Vec<MonitorUpdateId>> {
+               self.monitors.read().unwrap().iter().map(|(outpoint, holder)| {
+                       (*outpoint, holder.pending_monitor_updates.lock().unwrap().clone())
+               }).collect()
+       }
+
+       #[cfg(c_bindings)]
+       /// Lists the pending updates for each [`ChannelMonitor`] (by `OutPoint` being monitored).
+       pub fn list_pending_monitor_updates(&self) -> Vec<(OutPoint, Vec<MonitorUpdateId>)> {
+               self.monitors.read().unwrap().iter().map(|(outpoint, holder)| {
+                       (*outpoint, holder.pending_monitor_updates.lock().unwrap().clone())
+               }).collect()
+       }
+
+
        #[cfg(test)]
        pub fn remove_monitor(&self, funding_txo: &OutPoint) -> ChannelMonitor<ChannelSigner> {
                self.monitors.write().unwrap().remove(funding_txo).unwrap().monitor
@@ -475,10 +492,28 @@ where C::Target: chain::Filter,
        pub fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
                use crate::util::events::EventsProvider;
                let events = core::cell::RefCell::new(Vec::new());
-               let event_handler = |event: &events::Event| events.borrow_mut().push(event.clone());
+               let event_handler = |event: events::Event| events.borrow_mut().push(event);
                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 {
+                       handler(event).await;
+               }
+       }
 }
 
 impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>
@@ -719,8 +754,8 @@ impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> even
                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.handle_event(&event);
+               for event in pending_events {
+                       handler.handle_event(event);
                }
        }
        #[cfg(anchors)]
@@ -742,8 +777,8 @@ impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> even
                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.handle_event(&event);
+               for event in pending_events {
+                       handler.handle_event(event);
                }
        }
 }
@@ -798,7 +833,22 @@ mod tests {
                // Note that updates is a HashMap so the ordering here is actually random. This shouldn't
                // fail either way but if it fails intermittently it's depending on the ordering of updates.
                let mut update_iter = updates.iter();
-               nodes[1].chain_monitor.chain_monitor.channel_monitor_updated(*funding_txo, update_iter.next().unwrap().clone()).unwrap();
+               let next_update = update_iter.next().unwrap().clone();
+               // Should contain next_update when pending updates listed.
+               #[cfg(not(c_bindings))]
+               assert!(nodes[1].chain_monitor.chain_monitor.list_pending_monitor_updates().get(funding_txo)
+                       .unwrap().contains(&next_update));
+               #[cfg(c_bindings)]
+               assert!(nodes[1].chain_monitor.chain_monitor.list_pending_monitor_updates().iter()
+                       .find(|(txo, _)| txo == funding_txo).unwrap().1.contains(&next_update));
+               nodes[1].chain_monitor.chain_monitor.channel_monitor_updated(*funding_txo, next_update.clone()).unwrap();
+               // Should not contain the previously pending next_update when pending updates listed.
+               #[cfg(not(c_bindings))]
+               assert!(!nodes[1].chain_monitor.chain_monitor.list_pending_monitor_updates().get(funding_txo)
+                       .unwrap().contains(&next_update));
+               #[cfg(c_bindings)]
+               assert!(!nodes[1].chain_monitor.chain_monitor.list_pending_monitor_updates().iter()
+                       .find(|(txo, _)| txo == funding_txo).unwrap().1.contains(&next_update));
                assert!(nodes[1].chain_monitor.release_pending_monitor_events().is_empty());
                assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
                nodes[1].chain_monitor.chain_monitor.channel_monitor_updated(*funding_txo, update_iter.next().unwrap().clone()).unwrap();