pebble/ecs/
events.rs

1use crate::ecs::resources::Resources;
2use crate::ecs::system::{RequiredResource, Res, SystemParam};
3use crate::threading::{BackgroundTasks, SpawnableFuture};
4
5/// A double-buffered queue of `T` events, stored as a singleton resource.
6///
7/// Write with an [`EventWriter<T>`] system param, read with an
8/// [`EventReader<T>`] system param. An event sent during tick `N` stays
9/// visible to readers for the rest of tick `N` and all of tick `N + 1`, then
10/// is dropped — so a reader running anywhere in either tick sees it exactly
11/// once, regardless of whether it runs before or after the writer that sent
12/// it.
13///
14/// Not usable until registered with
15/// [`App::add_event`](crate::app::App::add_event), which inserts this
16/// resource and schedules the per-tick aging that gives the two-tick
17/// guarantee above.
18pub struct Events<T> {
19    current: Vec<(usize, T)>,
20    previous: Vec<(usize, T)>,
21    next_id: usize,
22}
23
24impl<T> Default for Events<T> {
25    fn default() -> Self {
26        Self {
27            current: Vec::new(),
28            previous: Vec::new(),
29            next_id: 0,
30        }
31    }
32}
33
34impl<T> Events<T> {
35    /// Push a new event, tagged with the next monotonically increasing id.
36    pub fn send(&mut self, event: T) {
37        self.current.push((self.next_id, event));
38        self.next_id += 1;
39    }
40
41    /// Age the buffers: last tick's `current` becomes `previous` (still
42    /// visible to readers that haven't caught up), and a fresh `current`
43    /// starts collecting this tick's sends.
44    ///
45    /// Called once per tick, before any user systems run, by the closure
46    /// [`App::add_event`](crate::app::App::add_event) registers — not meant
47    /// to be called directly.
48    pub(crate) fn update(&mut self) {
49        self.previous = std::mem::take(&mut self.current);
50    }
51}
52
53/// Read-only handle to an [`Events<T>`] resource, obtained as a system
54/// parameter.
55///
56/// Each system with an `EventReader<T>` parameter has its own private
57/// read cursor (persisted like [`Local`](crate::ecs::system::Local)), so
58/// multiple readers of the same event type don't interfere with each other.
59pub struct EventReader<'a, T: 'static> {
60    events: hecs::Ref<'a, Events<T>>,
61    last_seen: &'a mut usize,
62}
63
64impl<'a, T: 'static> EventReader<'a, T> {
65    /// Iterate events this reader hasn't seen yet, oldest first, then mark
66    /// them as read.
67    pub fn iter(&mut self) -> impl Iterator<Item = &T> + '_ {
68        let seen = *self.last_seen;
69        let unread: Vec<&T> = self
70            .events
71            .previous
72            .iter()
73            .chain(self.events.current.iter())
74            .filter(|(id, _)| *id >= seen)
75            .map(|(_, event)| event)
76            .collect();
77        *self.last_seen = self.events.next_id;
78        unread.into_iter()
79    }
80
81    /// `true` if there are no unread events for this reader.
82    pub fn is_empty(&self) -> bool {
83        let seen = *self.last_seen;
84        !self
85            .events
86            .previous
87            .iter()
88            .chain(self.events.current.iter())
89            .any(|(id, _)| *id >= seen)
90    }
91}
92
93impl<T> SystemParam for EventReader<'static, T>
94where
95    T: Send + Sync + 'static,
96{
97    type Item<'a> = EventReader<'a, T>;
98    type State = usize;
99
100    fn fetch<'a>(
101        state: &'a mut Self::State,
102        world: &'a hecs::World,
103        resources: &'a Resources,
104    ) -> Self::Item<'a> {
105        EventReader {
106            events: resources.get_resource(world),
107            last_seen: state,
108        }
109    }
110
111    fn requires() -> Vec<RequiredResource> {
112        vec![RequiredResource {
113            name: std::any::type_name::<Events<T>>(),
114            type_id: std::any::TypeId::of::<Events<T>>(),
115            present: |world, resources| resources.has_resource::<Events<T>>(world),
116            hint: Some(
117                "Register this event type with `app.add_event::<T>()` (or \
118                 `app.add_async_event::<T>()` if it's delivered by a background task) \
119                 before this system runs.",
120            ),
121        }]
122    }
123}
124
125/// Soft-access variant of [`EventReader<T>`]: `None` instead of panicking
126/// while `T` isn't registered (via [`App::add_event`](crate::app::App::add_event)),
127/// so a system that should just skip when an event type it optionally
128/// consumes hasn't been set up yet doesn't have to hard-depend on it. The
129/// read cursor still persists correctly across ticks once the event type
130/// does appear later — it lives in the same per-system `State` slot either way.
131impl<T> SystemParam for Option<EventReader<'static, T>>
132where
133    T: Send + Sync + 'static,
134{
135    type Item<'a> = Option<EventReader<'a, T>>;
136    type State = usize;
137
138    fn fetch<'a>(
139        state: &'a mut Self::State,
140        world: &'a hecs::World,
141        resources: &'a Resources,
142    ) -> Self::Item<'a> {
143        if resources.has_resource::<Events<T>>(world) {
144            return Some(EventReader {
145                events: resources.get_resource(world),
146                last_seen: state,
147            });
148        }
149        None
150    }
151}
152
153/// Write handle to an [`Events<T>`] resource, obtained as a system
154/// parameter.
155pub struct EventWriter<'a, T: 'static> {
156    events: hecs::RefMut<'a, Events<T>>,
157}
158
159impl<'a, T: 'static> EventWriter<'a, T> {
160    /// Queue `event` for delivery to every [`EventReader<T>`] for the rest
161    /// of this tick and all of the next.
162    pub fn send(&mut self, event: T) {
163        self.events.send(event);
164    }
165}
166
167impl<T> SystemParam for EventWriter<'static, T>
168where
169    T: Send + Sync + 'static,
170{
171    type Item<'a> = EventWriter<'a, T>;
172    type State = ();
173
174    fn fetch<'a>(
175        _state: &'a mut Self::State,
176        world: &'a hecs::World,
177        resources: &'a Resources,
178    ) -> Self::Item<'a> {
179        EventWriter {
180            events: resources.get_resource_mut(world),
181        }
182    }
183
184    fn requires() -> Vec<RequiredResource> {
185        vec![RequiredResource {
186            name: std::any::type_name::<Events<T>>(),
187            type_id: std::any::TypeId::of::<Events<T>>(),
188            present: |world, resources| resources.has_resource::<Events<T>>(world),
189            hint: Some(
190                "Register this event type with `app.add_event::<T>()` (or \
191                 `app.add_async_event::<T>()` if it's delivered by a background task) \
192                 before this system runs.",
193            ),
194        }]
195    }
196}
197
198/// Soft-access variant of [`EventWriter<T>`]: `None` instead of panicking
199/// while `T` isn't registered (via [`App::add_event`](crate::app::App::add_event)).
200impl<T> SystemParam for Option<EventWriter<'static, T>>
201where
202    T: Send + Sync + 'static,
203{
204    type Item<'a> = Option<EventWriter<'a, T>>;
205    type State = ();
206
207    fn fetch<'a>(
208        _state: &'a mut Self::State,
209        world: &'a hecs::World,
210        resources: &'a Resources,
211    ) -> Self::Item<'a> {
212        if resources.has_resource::<Events<T>>(world) {
213            return Some(EventWriter {
214                events: resources.get_resource_mut(world),
215            });
216        }
217        None
218    }
219}
220
221/// The channel [`AsyncEventWriter::spawn`] sends into and the drain system
222/// (registered by [`App::add_async_event`](crate::app::App::add_async_event))
223/// reads out of, turning finished background tasks into ordinary `T`
224/// events. A resource, but not one to reach for directly — go through
225/// [`AsyncEventWriter<T>`].
226pub(crate) struct AsyncEventChannel<T> {
227    tx: crossbeam_channel::Sender<T>,
228    rx: crossbeam_channel::Receiver<T>,
229}
230
231impl<T> AsyncEventChannel<T> {
232    pub(crate) fn new() -> Self {
233        let (tx, rx) = crossbeam_channel::unbounded();
234        Self { tx, rx }
235    }
236}
237
238/// `PreUpdate` system registered by [`App::add_async_event`](crate::app::App::add_async_event):
239/// drains every task that finished since last checked and re-delivers each
240/// result as a `T` event. Two ticks of latency at most between a task
241/// finishing and a reader seeing it — one to be drained here, one more it
242/// stays visible for like any [`Events<T>`] send.
243pub(crate) fn drain_async_events<T: Send + Sync + 'static>(
244    channel: Res<AsyncEventChannel<T>>,
245    mut writer: EventWriter<T>,
246) {
247    while let Ok(event) = channel.rx.try_recv() {
248        writer.send(event);
249    }
250}
251
252/// System parameter for spawning async work whose result is delivered as a
253/// `T` event once it resolves. The friendly front door for combining
254/// [`BackgroundTasks::spawn_async`] with the [`Events`] system, instead of
255/// hand-rolling a pending-task resource and poll system per async data
256/// source the way [`GraphicsPlugin`](crate::rendering::graphics_plugin::GraphicsPlugin)
257/// does for the GPU backend.
258///
259/// Register the event type with [`App::add_async_event::<T>()`](crate::app::App::add_async_event)
260/// (instead of [`App::add_event`](crate::app::App::add_event)) to make this usable as a system
261/// parameter. Reading the result is completely ordinary — just an
262/// [`EventReader<T>`], same as any other event; nothing about consuming it
263/// differs from a synchronously-sent one. Requires
264/// [`BackgroundTasksPlugin`](crate::threading::BackgroundTasksPlugin) to be
265/// registered, since [`spawn`](Self::spawn) drives its future through
266/// [`BackgroundTasks::spawn_async`].
267///
268/// ```ignore
269/// app.add_async_event::<ReadbackDone>();
270///
271/// fn start_readback(events: AsyncEventWriter<ReadbackDone>, backend: Res<WGPUBackend>) {
272///     let future = backend.readback_buffer(&buf);
273///     events.spawn(async move { ReadbackDone(future.await) });
274/// }
275///
276/// fn on_readback(mut reader: EventReader<ReadbackDone>) {
277///     for event in reader.iter() {
278///         // event.0
279///     }
280/// }
281/// ```
282pub struct AsyncEventWriter<'a, T: Send + 'static> {
283    tasks: hecs::Ref<'a, BackgroundTasks>,
284    channel: hecs::Ref<'a, AsyncEventChannel<T>>,
285}
286
287impl<'a, T: Send + 'static> AsyncEventWriter<'a, T> {
288    /// Spawn `future`. Once it resolves, its output is delivered to every
289    /// [`EventReader<T>`] — no matter how many ticks the future takes, and
290    /// with no need to hold onto a handle or poll anything yourself.
291    pub fn spawn(&self, future: impl SpawnableFuture<T>) {
292        let tx = self.channel.tx.clone();
293        let _ = self.tasks.spawn_async(async move {
294            // Ignore send failure: it only happens if the receiving end
295            // (owned by the App) was dropped, meaning the app has shut down
296            // and there's nothing left to deliver this to.
297            let _ = tx.send(future.await);
298        });
299    }
300}
301
302impl<T> SystemParam for AsyncEventWriter<'static, T>
303where
304    T: Send + Sync + 'static,
305{
306    type Item<'a> = AsyncEventWriter<'a, T>;
307    type State = ();
308
309    fn fetch<'a>(
310        _state: &'a mut Self::State,
311        world: &'a hecs::World,
312        resources: &'a Resources,
313    ) -> Self::Item<'a> {
314        AsyncEventWriter {
315            tasks: resources.get_resource(world),
316            channel: resources.get_resource(world),
317        }
318    }
319
320    fn requires() -> Vec<RequiredResource> {
321        vec![
322            RequiredResource {
323                name: std::any::type_name::<BackgroundTasks>(),
324                type_id: std::any::TypeId::of::<BackgroundTasks>(),
325                present: |world, resources| resources.has_resource::<BackgroundTasks>(world),
326                hint: Some(
327                    "`AsyncEventWriter` drives its future through `BackgroundTasks` — register \
328                     `app.add_plugin(BackgroundTasksPlugin::new(worker_count))` before this system runs.",
329                ),
330            },
331            RequiredResource {
332                name: std::any::type_name::<AsyncEventChannel<T>>(),
333                type_id: std::any::TypeId::of::<AsyncEventChannel<T>>(),
334                present: |world, resources| resources.has_resource::<AsyncEventChannel<T>>(world),
335                hint: Some(
336                    "Register this event type with `app.add_async_event::<T>()` (not \
337                     `app.add_event`) before this system runs.",
338                ),
339            },
340        ]
341    }
342}
343
344/// Soft-access variant of [`AsyncEventWriter<T>`]: `None` instead of panicking
345/// while `T` isn't registered via [`add_async_event`](crate::app::App::add_async_event)
346/// (or [`BackgroundTasksPlugin`](crate::threading::BackgroundTasksPlugin)
347/// isn't installed yet). Useful for systems that only opportunistically
348/// spawn background work and shouldn't hard-fail while the app is still
349/// assembling its plugins.
350impl<T> SystemParam for Option<AsyncEventWriter<'static, T>>
351where
352    T: Send + Sync + 'static,
353{
354    type Item<'a> = Option<AsyncEventWriter<'a, T>>;
355    type State = ();
356
357    fn fetch<'a>(
358        _state: &'a mut Self::State,
359        world: &'a hecs::World,
360        resources: &'a Resources,
361    ) -> Self::Item<'a> {
362        if resources.has_resource::<BackgroundTasks>(world)
363            && resources.has_resource::<AsyncEventChannel<T>>(world)
364        {
365            return Some(AsyncEventWriter {
366                tasks: resources.get_resource(world),
367                channel: resources.get_resource(world),
368            });
369        }
370        None
371    }
372}
373
374#[cfg(test)]
375mod tests {
376    use super::*;
377    use crate::app::{App, SystemStage};
378    use crate::ecs::system::{Local, ResMut};
379
380    struct Damage(u32);
381
382    struct Seen(Vec<u32>);
383
384    fn send_once(mut writer: EventWriter<Damage>, mut sent: Local<bool>) {
385        if !*sent {
386            writer.send(Damage(5));
387            *sent = true;
388        }
389    }
390
391    fn record_damage(mut reader: EventReader<Damage>, mut seen: ResMut<Seen>) {
392        for event in reader.iter() {
393            seen.0.push(event.0);
394        }
395    }
396
397    #[test]
398    fn reader_in_an_earlier_stage_still_catches_a_same_tick_send_one_tick_later() {
399        let mut app = App::new();
400        app.add_event::<Damage>();
401        app.add_resource(Seen(Vec::new()));
402
403        // The reader runs in PreUpdate, before the writer's Update stage —
404        // so within the tick the event is sent, this reader has already run
405        // and can't see it yet. It should only pick it up on the *next*
406        // tick, via the aged `previous` buffer, and exactly once.
407        app.add_system(SystemStage::PreUpdate, record_damage);
408        app.add_system(SystemStage::Update, send_once);
409        app.build();
410
411        app.update(); // tick 1: reader runs first (nothing sent yet), then writer sends
412        assert_eq!(app.get_resource::<Seen>().0, Vec::<u32>::new());
413
414        app.update(); // tick 2: reader now sees last tick's send
415        assert_eq!(app.get_resource::<Seen>().0, vec![5]);
416
417        app.update(); // tick 3: event has aged out, nothing new
418        assert_eq!(app.get_resource::<Seen>().0, vec![5]);
419    }
420
421    struct Loaded(u32);
422
423    fn kick_off(events: AsyncEventWriter<Loaded>, mut sent: Local<bool>) {
424        if !*sent {
425            events.spawn(async { Loaded(42) });
426            *sent = true;
427        }
428    }
429
430    fn record_loaded(mut reader: EventReader<Loaded>, mut seen: ResMut<Seen>) {
431        for event in reader.iter() {
432            seen.0.push(event.0);
433        }
434    }
435
436    #[test]
437    fn async_events_deliver_a_background_tasks_result_as_an_event() {
438        use crate::ecs::plugin::Plugin;
439        use crate::threading::BackgroundTasksPlugin;
440
441        let mut app = App::new();
442        BackgroundTasksPlugin::new(1).build(&mut app);
443        app.add_async_event::<Loaded>();
444        app.add_resource(Seen(Vec::new()));
445        app.add_system(SystemStage::Update, kick_off);
446        app.add_system(SystemStage::PostRender, record_loaded);
447        app.build();
448
449        // The task resolves on a worker thread, off the tick loop — poll a
450        // bounded number of ticks (with a short sleep so the worker gets a
451        // chance to run) instead of assuming it lands on a specific tick.
452        for _ in 0..200 {
453            app.update();
454            if !app.get_resource::<Seen>().0.is_empty() {
455                break;
456            }
457            std::thread::sleep(std::time::Duration::from_millis(5));
458        }
459
460        assert_eq!(app.get_resource::<Seen>().0, vec![42]);
461    }
462}
pebble/ecs/events.rs

pebble/ecs/
events.rs
