lumen_core/
plugin_events.rs

1//! Cross-thread event bus for plugins, modules and add-ons.
2//!
3//! Something outside a script call pushes events at the engine from whichever
4//! thread it likes: a worker delivering what it watched for, a call body
5//! handing work past its own return, a promise settling in a page. This bus is
6//! where those events wait for the tick that drains them.
7//!
8//! An event arrives in one of two forms. A portable plugin lives across a
9//! serialized boundary, so its event is encoded bytes. Code that shares the
10//! engine's address space (a runtime module, a browser add-on) hands over the
11//! value itself, and nothing is encoded or decoded on its way through.
12//!
13//! The core knows neither shape: the event is built from script-surface types
14//! this crate does not know, and the script layer that does know them
15//! registers the per-tick drain (`lumen-script`'s `collect_plugin_events`).
16//! What lives here is the bus itself, mirroring the external typed-property
17//! bus in [`crate::property_store`], so [`crate::tick::work_pending`] can count
18//! undrained events as pending work and a parked app wakes to take them.
19
20use std::any::Any;
21use std::sync::{Mutex, OnceLock};
22
23use crossbeam_channel::{Receiver, Sender, unbounded};
24
25/// One event waiting on the bus, in the form it was pushed in.
26pub enum QueuedEvent {
27    /// Encoded by a plugin on the other side of a serialized boundary.
28    Bytes(Vec<u8>),
29    /// Handed over by code in the engine's own address space. The script
30    /// layer that drains the bus knows the type to take it back as.
31    Value(Box<dyn Any + Send>),
32}
33
34impl QueuedEvent {
35    /// The encoded bytes, for an event pushed in that form.
36    pub fn bytes(&self) -> Option<&[u8]> {
37        match self {
38            QueuedEvent::Bytes(bytes) => Some(bytes),
39            QueuedEvent::Value(_) => None,
40        }
41    }
42}
43
44impl std::fmt::Debug for QueuedEvent {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        match self {
47            QueuedEvent::Bytes(bytes) => f.debug_tuple("Bytes").field(bytes).finish(),
48            QueuedEvent::Value(_) => f.write_str("Value(..)"),
49        }
50    }
51}
52
53static PLUGIN_EVENT_TX: OnceLock<Sender<QueuedEvent>> = OnceLock::new();
54static PLUGIN_EVENT_RX: OnceLock<Mutex<Receiver<QueuedEvent>>> = OnceLock::new();
55static PLUGIN_EVENT_WAKER: Mutex<Option<crate::app::EventLoopWaker>> = Mutex::new(None);
56
57fn init_plugin_event_channel() -> &'static Sender<QueuedEvent> {
58    PLUGIN_EVENT_TX.get_or_init(|| {
59        let (tx, rx) = unbounded();
60        let _ = PLUGIN_EVENT_RX.set(Mutex::new(rx));
61        tx
62    })
63}
64
65/// Idempotently initialises the plugin-event channel. Safe to call multiple
66/// times.
67pub fn init_plugin_events() {
68    let _ = init_plugin_event_channel();
69}
70
71/// Install the waker a push nudges, so an event arriving while the app sits
72/// parked in its event loop's `Wait` state triggers the tick that drains it
73/// instead of waiting for an unrelated wake. The backends install it beside
74/// the [`crate::app::EventLoopWaker`] resource; a later install replaces the
75/// earlier one, because a process can run apps in sequence and the bus
76/// belongs to whichever loop is live.
77pub fn set_plugin_event_waker(waker: crate::app::EventLoopWaker) {
78    if let Ok(mut slot) = PLUGIN_EVENT_WAKER.lock() {
79        *slot = Some(waker);
80    }
81}
82
83/// Queue one encoded plugin event from any thread. Picked up on the next
84/// tick by the script layer's drain; a parked event loop is woken to run
85/// that tick when a waker is installed ([`set_plugin_event_waker`]).
86///
87/// Returns `false` when the channel has disconnected.
88pub fn push_plugin_event(bytes: Vec<u8>) -> bool {
89    push(QueuedEvent::Bytes(bytes))
90}
91
92/// Queue one event as the value itself, from any thread, for code that shares
93/// the engine's address space. Delivered and woken for exactly like
94/// [`push_plugin_event`]; the drain takes it back as the type it was pushed
95/// as.
96///
97/// Returns `false` when the channel has disconnected.
98pub fn push_plugin_value(value: Box<dyn Any + Send>) -> bool {
99    push(QueuedEvent::Value(value))
100}
101
102fn push(event: QueuedEvent) -> bool {
103    let sent = init_plugin_event_channel().send(event).is_ok();
104    if sent
105        && let Ok(slot) = PLUGIN_EVENT_WAKER.lock()
106        && let Some(waker) = slot.as_ref()
107    {
108        waker.wake();
109    }
110    sent
111}
112
113/// Take every queued event, in arrival order. Returns the empty vector when
114/// the channel was never initialised, is empty, or its lock is poisoned.
115pub fn drain_plugin_events() -> Vec<QueuedEvent> {
116    let Some(rx_lock) = PLUGIN_EVENT_RX.get() else {
117        return Vec::new();
118    };
119    let Ok(rx) = rx_lock.lock() else {
120        return Vec::new();
121    };
122    let mut events = Vec::new();
123    while let Ok(bytes) = rx.try_recv() {
124        events.push(bytes);
125    }
126    events
127}
128
129/// Whether the bus currently holds undrained events.
130///
131/// Non-destructive; one of [`crate::tick::work_pending`]'s sources, so a
132/// driver that only wakes on events schedules another tick while an event a
133/// worker thread pushed is still sitting here.
134pub fn plugin_events_pending() -> bool {
135    PLUGIN_EVENT_RX
136        .get()
137        .and_then(|rx_lock| rx_lock.lock().ok().map(|rx| !rx.is_empty()))
138        .unwrap_or(false)
139}
140
141/// Empties the bus, throwing away whatever it holds.
142///
143/// One channel per process, so a caller that runs several apps in sequence
144/// calls this between them, beside
145/// [`crate::property_store::discard_external_properties`].
146pub fn discard_plugin_events() {
147    let Some(rx_lock) = PLUGIN_EVENT_RX.get() else {
148        return;
149    };
150    let Ok(rx) = rx_lock.lock() else {
151        return;
152    };
153    while rx.try_recv().is_ok() {}
154}
155
156#[cfg(test)]
157mod tests {
158    use super::*;
159
160    /// The bus is one process-global channel, so its tests hold this to keep
161    /// from draining each other's pushes.
162    static SERIAL: Mutex<()> = Mutex::new(());
163
164    /// The encoded events of a drain, in order.
165    fn bytes_of(events: Vec<QueuedEvent>) -> Vec<Vec<u8>> {
166        events
167            .iter()
168            .filter_map(|event| event.bytes().map(<[u8]>::to_vec))
169            .collect()
170    }
171
172    #[test]
173    fn an_event_says_which_form_it_travels_in() {
174        let bytes = QueuedEvent::Bytes(vec![7, 8]);
175        let value = QueuedEvent::Value(Box::new(3_u32));
176        assert_eq!(bytes.bytes(), Some(&[7, 8][..]));
177        assert_eq!(value.bytes(), None);
178        assert_eq!(format!("{bytes:?}"), "Bytes([7, 8])");
179        assert_eq!(format!("{value:?}"), "Value(..)");
180    }
181
182    #[test]
183    fn a_value_travels_as_itself_beside_the_bytes() {
184        let _guard = SERIAL.lock().unwrap_or_else(|e| e.into_inner());
185        discard_plugin_events();
186        assert!(push_plugin_event(vec![1]));
187        assert!(push_plugin_value(Box::new(String::from("typed"))));
188        assert!(plugin_events_pending());
189        let mut drained = drain_plugin_events().into_iter();
190        assert_eq!(
191            drained.next().and_then(|e| e.bytes().map(<[u8]>::to_vec)),
192            Some(vec![1])
193        );
194        match drained.next() {
195            Some(QueuedEvent::Value(value)) => {
196                assert_eq!(
197                    value
198                        .downcast::<String>()
199                        .ok()
200                        .as_deref()
201                        .map(String::as_str),
202                    Some("typed")
203                );
204            }
205            other => panic!("expected the value back, got {other:?}"),
206        }
207        assert!(drained.next().is_none());
208    }
209
210    #[test]
211    fn events_queue_report_pending_and_drain_in_order() {
212        let _guard = SERIAL.lock().unwrap_or_else(|e| e.into_inner());
213        discard_plugin_events();
214        assert!(!plugin_events_pending());
215        assert!(push_plugin_event(vec![1]));
216        assert!(push_plugin_event(vec![2, 3]));
217        assert!(plugin_events_pending());
218        assert_eq!(bytes_of(drain_plugin_events()), vec![vec![1], vec![2, 3]]);
219        assert!(!plugin_events_pending());
220        assert!(drain_plugin_events().is_empty());
221    }
222
223    #[test]
224    fn discard_empties_the_bus() {
225        let _guard = SERIAL.lock().unwrap_or_else(|e| e.into_inner());
226        discard_plugin_events();
227        assert!(push_plugin_event(vec![9]));
228        discard_plugin_events();
229        assert!(!plugin_events_pending());
230        assert!(drain_plugin_events().is_empty());
231    }
232
233    #[test]
234    fn a_push_wakes_a_parked_loop() {
235        let _guard = SERIAL.lock().unwrap_or_else(|e| e.into_inner());
236        discard_plugin_events();
237        // Stand-in for a backend's parked `Wait` state: a thread blocks on a
238        // condvar until the waker fires, then drains the bus - the same
239        // wake-then-tick sequence the real loops run.
240        let parked = std::sync::Arc::new((Mutex::new(false), std::sync::Condvar::new()));
241        let waker_side = std::sync::Arc::clone(&parked);
242        set_plugin_event_waker(crate::app::EventLoopWaker(std::sync::Arc::new(move || {
243            let (flag, cv) = &*waker_side;
244            *flag.lock().unwrap_or_else(|e| e.into_inner()) = true;
245            cv.notify_all();
246        })));
247
248        let loop_side = std::sync::Arc::clone(&parked);
249        let parked_loop = std::thread::spawn(move || {
250            let (flag, cv) = &*loop_side;
251            let mut woken = flag.lock().unwrap_or_else(|e| e.into_inner());
252            while !*woken {
253                let (g, timed_out) = cv
254                    .wait_timeout(woken, std::time::Duration::from_secs(10))
255                    .unwrap_or_else(|e| e.into_inner());
256                woken = g;
257                assert!(!timed_out.timed_out(), "the push never woke the loop");
258            }
259            drain_plugin_events()
260        });
261
262        assert!(push_plugin_event(vec![5]));
263        let delivered = parked_loop.join().expect("parked loop");
264        assert_eq!(bytes_of(delivered), vec![vec![5]]);
265
266        // Put the slot back so later tests see the uninstalled state.
267        if let Ok(mut slot) = PLUGIN_EVENT_WAKER.lock() {
268            *slot = None;
269        }
270    }
271
272    #[test]
273    fn an_undrained_event_counts_as_pending_work() {
274        let _guard = SERIAL.lock().unwrap_or_else(|e| e.into_inner());
275        discard_plugin_events();
276        // A bare world carries no animation or frame-dirty resources, so with
277        // an event queued the bus is what `work_pending` reports. Only the
278        // monotonic direction is asserted against `work_pending` itself: the
279        // sibling external-property bus is process-global too, and a parallel
280        // test may hold a write in it while this one runs.
281        let world = bevy_ecs::world::World::new();
282        assert!(push_plugin_event(vec![7]));
283        assert!(
284            crate::tick::work_pending(&world),
285            "a queued plugin event must report pending so a parked driver schedules a tick"
286        );
287        discard_plugin_events();
288        assert!(!plugin_events_pending());
289    }
290}
lumen_core/plugin_events.rs

lumen_core/
plugin_events.rs
