fedimint_client/sm/
notifier.rs

1use fedimint_client_module::module::FinalClientIface;
2use fedimint_client_module::sm::{DynState, ModuleNotifier};
3use fedimint_core::core::ModuleInstanceId;
4use fedimint_core::util::FmtCompact;
5use tracing::{debug, trace};
6
7/// State transition notifier owned by the modularized client used to inform
8/// modules of state transitions.
9///
10/// To not lose any state transitions that happen before a module subscribes to
11/// the operation the notifier loads all belonging past state transitions from
12/// the DB. State transitions may be reported multiple times and out of order.
13#[derive(Clone)]
14pub struct Notifier {
15    /// Broadcast channel used to send state transitions to all subscribers
16    broadcast: tokio::sync::broadcast::Sender<DynState>,
17}
18
19impl Notifier {
20    #[allow(clippy::new_without_default)]
21    pub fn new() -> Self {
22        let (sender, _receiver) = tokio::sync::broadcast::channel(10_000);
23        Self { broadcast: sender }
24    }
25
26    /// Notify all subscribers of a state transition
27    pub fn notify(&self, state: DynState) {
28        let queue_len = self.broadcast.len();
29        trace!(?state, %queue_len, "Sending notification about state transition");
30        // FIXME: use more robust notification mechanism
31        if let Err(err) = self.broadcast.send(state) {
32            debug!(
33                err = %err.fmt_compact(),
34                %queue_len,
35                receivers=self.broadcast.receiver_count(),
36                "Could not send state transition notification, no active receivers"
37            );
38        }
39    }
40
41    /// Subscribe to the state transitions of every module and operation.
42    ///
43    /// Unlike a [`ModuleNotifier`] subscription this replays nothing from the
44    /// database: the receiver only sees transitions notified after this call.
45    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<DynState> {
46        self.broadcast.subscribe()
47    }
48
49    /// Create a new notifier for a specific module instance that can only
50    /// subscribe to the instance's state transitions
51    pub fn module_notifier<S>(
52        &self,
53        module_instance: ModuleInstanceId,
54        client: FinalClientIface,
55    ) -> ModuleNotifier<S>
56    where
57        S: fedimint_client_module::sm::State,
58    {
59        ModuleNotifier::new(self.broadcast.clone(), module_instance, client)
60    }
61
62    /// Create a [`NotifierSender`] handle that lets the owner trigger
63    /// notifications without having to hold a full `Notifier`.
64    pub fn sender(&self) -> NotifierSender {
65        NotifierSender {
66            sender: self.broadcast.clone(),
67        }
68    }
69}
70
71/// Notifier send handle that can be shared to places where we don't need an
72/// entire [`Notifier`] but still need to trigger notifications. The main use
73/// case is triggering notifications when a DB transaction was committed
74/// successfully.
75pub struct NotifierSender {
76    sender: tokio::sync::broadcast::Sender<DynState>,
77}
78
79impl NotifierSender {
80    /// Notify all subscribers of a state transition
81    pub fn notify(&self, state: DynState) {
82        let _res = self.sender.send(state);
83    }
84}
fedimint_client/sm/notifier.rs

fedimint_client/sm/
notifier.rs
