egui_charts/services/
events.rs

1//! A bounded stream of pushes from a service to the terminal.
2
3use crate::data::inbox::Inbox;
4use std::marker::PhantomData;
5use std::sync::Arc;
6use std::sync::atomic::{AtomicBool, Ordering};
7
8/// Updates a service pushes to the terminal, from any thread, runtime or browser task. Cheap to
9/// clone; clones push into the same stream. The first push after a frame wakes the UI, and the
10/// pushes are applied in order on the next frame.
11///
12/// A stream holds at most 10,000 pushes between two frames. A push between frames counts towards
13/// that bound, so a burst larger than the bound always ends the stream. A push beyond the bound is
14/// dropped and ends the stream: the terminal discards what the stream built and calls the service again, which
15/// pushes its current state first.
16pub struct Events<T: Send + 'static> {
17    inbox: Arc<Inbox>,
18    key: u64,
19    active: Arc<AtomicBool>,
20    event: PhantomData<fn(T)>,
21}
22
23impl<T: Send + 'static> Events<T> {
24    /// Delivers `event` to the terminal, or nothing once the stream is inactive.
25    pub fn push(&self, event: T) {
26        self.inbox
27            .push_answer(self.key, Box::new(event), &self.active);
28    }
29
30    /// False once the terminal stopped listening: the symbol or interval changed, the stream
31    /// lagged, or the terminal was dropped. Stop pushing then; later pushes are ignored.
32    pub fn is_active(&self) -> bool {
33        self.active.load(Ordering::Relaxed)
34    }
35}
36
37impl<T: Send + 'static> Clone for Events<T> {
38    fn clone(&self) -> Self {
39        Self {
40            inbox: Arc::clone(&self.inbox),
41            key: self.key,
42            active: Arc::clone(&self.active),
43            event: PhantomData,
44        }
45    }
46}
47
48/// The terminal's end of one stream: dropping it ends the stream.
49pub(crate) struct Listener(Arc<AtomicBool>);
50
51impl Listener {
52    /// A stream whose pushes reach `inbox` under `key`, and its listener.
53    pub(crate) fn new<T: Send + 'static>(inbox: &Arc<Inbox>, key: u64) -> (Self, Events<T>) {
54        let active = Arc::new(AtomicBool::new(true));
55        let events = Events {
56            inbox: Arc::clone(inbox),
57            key,
58            active: Arc::clone(&active),
59            event: PhantomData,
60        };
61        (Self(active), events)
62    }
63
64    /// False once the stream lagged: drop this listener, discard what the stream built and its
65    /// pushes still in the inbox, and subscribe again.
66    pub(crate) fn is_active(&self) -> bool {
67        self.0.load(Ordering::Relaxed)
68    }
69}
70
71impl Drop for Listener {
72    fn drop(&mut self) {
73        self.0.store(false, Ordering::Relaxed);
74    }
75}
76
77#[cfg(test)]
78mod tests {
79    use super::*;
80    use std::sync::atomic::AtomicUsize;
81
82    fn counting_wakes(inbox: &Inbox) -> Arc<AtomicUsize> {
83        let wakes = Arc::new(AtomicUsize::new(0));
84        let counter = Arc::clone(&wakes);
85        inbox.set_waker(move || {
86            counter.fetch_add(1, Ordering::Relaxed);
87        });
88        wakes
89    }
90
91    /// Each event with the key of its stream, in the order the inbox delivers them.
92    fn drain(inbox: &Inbox) -> Vec<(u64, u32)> {
93        let answers = inbox.drain().answers.into_iter();
94        answers
95            .map(|(key, event)| (key, *event.downcast::<u32>().unwrap()))
96            .collect()
97    }
98
99    /// Guards the memory bound and the repaint rate: a stream must deliver up to 10,000 pushes per
100    /// frame in push order and wake the UI once, the bound must count one stream's pushes since the
101    /// last frame (not its lifetime, not every stream's), and the push over it must end that
102    /// stream for both ends. A reordered or coalesced queue reverses an upsert and a remove; a
103    /// missing bound grows without limit while frames stop.
104    #[test]
105    fn a_stream_takes_10_000_pushes_per_frame_in_order_then_lags() {
106        let inbox = Arc::new(Inbox::default());
107        let wakes = counting_wakes(&inbox);
108        let (listener, events) = Listener::new::<u32>(&inbox, 1);
109        let (_other_listener, other) = Listener::new::<u32>(&inbox, 2);
110        let producer = events.clone();
111        std::thread::spawn(move || (0..10_000).for_each(|i| producer.push(i)))
112            .join()
113            .unwrap();
114        other.push(7);
115        assert_eq!(wakes.load(Ordering::Relaxed), 1);
116        let full_frame = Vec::from_iter((0..10_000).map(|i| (1, i)));
117        assert_eq!(drain(&inbox), [full_frame.clone(), vec![(2, 7)]].concat());
118
119        (0..10_000).for_each(|i| events.push(i));
120        assert!(events.is_active() && listener.is_active());
121        assert_eq!(wakes.load(Ordering::Relaxed), 2);
122        events.push(10_000);
123        other.push(8);
124        assert!(!events.is_active() && !listener.is_active());
125        assert!(other.is_active());
126        assert_eq!(drain(&inbox), [full_frame, vec![(2, 8)]].concat());
127    }
128
129    /// Guards the stop signal: once the terminal drops its listener (symbol or interval changed,
130    /// terminal dropped), every clone of the stream must read inactive so the host stops, and
131    /// pushes that still arrive must neither reach the inbox nor wake the UI.
132    #[test]
133    fn a_stopped_stream_is_inactive_and_ignored() {
134        let inbox = Arc::new(Inbox::default());
135        let wakes = counting_wakes(&inbox);
136        let (listener, events) = Listener::new::<u32>(&inbox, 1);
137        let producer = events.clone();
138        events.push(1);
139        drop(listener);
140        assert!(!events.is_active() && !producer.is_active());
141        assert_eq!(drain(&inbox), [(1, 1)]);
142        std::thread::spawn(move || producer.push(2)).join().unwrap();
143        assert_eq!(drain(&inbox), []);
144        assert_eq!(wakes.load(Ordering::Relaxed), 1);
145    }
146}
egui_charts/services/events.rs

egui_charts/services/
events.rs
