rovenue/
observer.rs

1use crate::logging::Logger;
2use std::sync::{Arc, Mutex};
3
4/// What changed in the SDK's internal state.
5///
6/// Façades translate these into platform-native streams (AsyncStream / Flow / JS bus).
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum ChangeEvent {
9    EntitlementsChanged,
10    IdentityChanged,
11    VirtualCurrenciesChanged,
12    RemoteConfigChanged,
13}
14
15/// Implemented by façades to receive cache-state notifications from the core.
16pub trait Observer: Send + Sync {
17    fn on_change(&self, event: ChangeEvent);
18}
19
20/// Holds strong `Arc` references so FFI-registered observers stay alive for the
21/// bus's lifetime. The FFI boundary passes a `Box<dyn Observer>` once and never
22/// holds another strong reference on the caller side, so the bus must own it.
23///
24/// Rust-side callers that want lifecycle control can drop the entire bus or
25/// call `clear()` to release all observers at once.
26#[derive(Default)]
27pub struct ObserverBus {
28    subs: Mutex<Vec<Arc<dyn Observer>>>,
29    logger: Mutex<Option<Arc<Logger>>>,
30}
31
32impl ObserverBus {
33    pub fn with_logger(self, logger: Arc<Logger>) -> Self {
34        *self.logger.lock().unwrap_or_else(|e| e.into_inner()) = Some(logger);
35        self
36    }
37
38    pub fn register(&self, obs: Arc<dyn Observer>) {
39        let mut guard = self.subs.lock().unwrap_or_else(|e| e.into_inner());
40        guard.push(obs);
41    }
42
43    pub fn emit(&self, event: ChangeEvent) {
44        let guard = self.subs.lock().unwrap_or_else(|e| e.into_inner());
45        for s in guard.iter() {
46            let s = s.clone();
47            let ev = event;
48            if std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || s.on_change(ev)))
49                .is_err()
50            {
51                if let Some(l) = self
52                    .logger
53                    .lock()
54                    .unwrap_or_else(|e| e.into_inner())
55                    .as_ref()
56                {
57                    l.warn("observer.on_change panicked; skipping");
58                }
59            }
60        }
61    }
62
63    /// Releases all registered observers. Primarily intended for tests and
64    /// shutdown paths.
65    pub fn clear(&self) {
66        let mut guard = self.subs.lock().unwrap_or_else(|e| e.into_inner());
67        guard.clear();
68    }
69
70    pub fn live_count(&self) -> usize {
71        let guard = self.subs.lock().unwrap_or_else(|e| e.into_inner());
72        guard.len()
73    }
74}
75
76#[cfg(test)]
77mod panic_tests {
78    use super::*;
79    use crate::logging::{LogLevel, LogRecord, LogSink, Logger};
80    use std::sync::{
81        atomic::{AtomicU32, Ordering},
82        Arc,
83    };
84
85    struct Panicky;
86    impl Observer for Panicky {
87        fn on_change(&self, _e: ChangeEvent) {
88            panic!("boom");
89        }
90    }
91
92    struct Counter(Arc<AtomicU32>);
93    impl Observer for Counter {
94        fn on_change(&self, _e: ChangeEvent) {
95            self.0.fetch_add(1, Ordering::SeqCst);
96        }
97    }
98
99    struct Collector(Arc<std::sync::Mutex<Vec<LogRecord>>>);
100    impl LogSink for Collector {
101        fn on_log(&self, r: LogRecord) {
102            self.0.lock().unwrap().push(r);
103        }
104    }
105
106    #[test]
107    fn a_panicking_observer_does_not_abort_dispatch() {
108        let bus = ObserverBus::default();
109        let hits = Arc::new(AtomicU32::new(0));
110        bus.register(Arc::new(Panicky));
111        bus.register(Arc::new(Counter(hits.clone())));
112        bus.emit(ChangeEvent::EntitlementsChanged); // must not unwind; Counter must still run
113        assert_eq!(hits.load(Ordering::SeqCst), 1);
114    }
115
116    #[test]
117    fn observer_panic_is_logged_as_warn_not_eprintln() {
118        let recs = Arc::new(std::sync::Mutex::new(Vec::new()));
119        let logger = Arc::new(Logger::new(LogLevel::Warn));
120        logger.set_sink(Arc::new(Collector(recs.clone())));
121        let bus = ObserverBus::default().with_logger(logger);
122        bus.register(Arc::new(Panicky));
123        bus.emit(ChangeEvent::EntitlementsChanged);
124        let got = recs.lock().unwrap();
125        assert_eq!(got.len(), 1);
126        assert_eq!(got[0].level, LogLevel::Warn);
127        assert!(got[0].message.contains("observer.on_change panicked"));
128    }
129}
rovenue/observer.rs

rovenue/
observer.rs
