mirrors/rust-lightning
1
0
Fork 0
mirror of https://github.com/lightningdevkit/rust-lightning.git synced 2025-03-03 10:46:48 +01:00
Code Issues Projects Releases Packages Wiki Activity

Implement async versions of process_pending_events

Browse source
This commit is contained in:
Wilmer Paulino 2022-11-02 12:39:07 -07:00
parent 05cb467234
commit 55b714c01d
No known key found for this signature in database
GPG key ID: 6DF57B9F9514972F
2 changed files with 53 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
lightning/src/chain/chainmonitor.rs
View file

@ -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::*;
@ -496,6 +496,24 @@ 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 {
handler(event).await;
}
}
}
impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>

35
lightning/src/ln/channelmanager.rs
View file

@ -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>