rustlavel_core/
events.rs

1//! The internal instrumentation bus.
2//!
3//! Every part of the framework reports what it does here; packages listen.
4//! Telescope, structured logging, and tracing exporters are all just
5//! subscribers, which is why nothing in core needs to know they exist.
6//!
7//! Events carry an open field map rather than a fixed enum so a package can
8//! record something core has never heard of.
9
10use crate::json::Json;
11use std::collections::BTreeMap;
12use std::sync::{Arc, OnceLock, RwLock};
13use std::time::{Duration, SystemTime};
14
15/// One recorded moment: a request, a query, a dispatched job, an exception.
16#[derive(Debug, Clone)]
17pub struct Event {
18    /// A dotted, stable identifier: `http.request`, `db.query`, `ai.call`.
19    pub kind: &'static str,
20    pub at: SystemTime,
21    /// How long the recorded work took, when it was a span rather than a point.
22    pub duration: Option<Duration>,
23    pub fields: BTreeMap<String, Json>,
24}
25
26impl Event {
27    pub fn new(kind: &'static str) -> Self {
28        Event { kind, at: SystemTime::now(), duration: None, fields: BTreeMap::new() }
29    }
30
31    pub fn with(mut self, key: &str, value: impl Into<Json>) -> Self {
32        self.fields.insert(key.to_string(), value.into());
33        self
34    }
35
36    pub fn took(mut self, duration: Duration) -> Self {
37        self.duration = Some(duration);
38        self
39    }
40
41    pub fn field(&self, key: &str) -> Option<&Json> {
42        self.fields.get(key)
43    }
44
45    /// Milliseconds elapsed, for renderers that show a duration column.
46    pub fn duration_ms(&self) -> Option<f64> {
47        self.duration.map(|d| d.as_secs_f64() * 1000.0)
48    }
49
50    /// Publish this event to every subscriber.
51    pub fn dispatch(self) {
52        dispatch(self);
53    }
54}
55
56/// Anything that wants to observe framework activity.
57pub trait Subscriber: Send + Sync + 'static {
58    fn handle(&self, event: &Event);
59
60    /// Return false to skip events a subscriber does not care about, before
61    /// the event is cloned or formatted.
62    fn interested_in(&self, _kind: &str) -> bool {
63        true
64    }
65}
66
67impl<F> Subscriber for F
68where
69    F: Fn(&Event) + Send + Sync + 'static,
70{
71    fn handle(&self, event: &Event) {
72        self(event)
73    }
74}
75
76type Subscribers = RwLock<Vec<Arc<dyn Subscriber>>>;
77
78fn registry() -> &'static Subscribers {
79    static REGISTRY: OnceLock<Subscribers> = OnceLock::new();
80    REGISTRY.get_or_init(|| RwLock::new(Vec::new()))
81}
82
83/// Register a subscriber for the lifetime of the process.
84pub fn subscribe(subscriber: impl Subscriber) {
85    registry().write().expect("event registry poisoned").push(Arc::new(subscriber));
86}
87
88/// Publish an event to every interested subscriber.
89///
90/// Dispatch is synchronous and best-effort: a subscriber that needs to do real
91/// work (writing to a database, shipping over the network) is expected to queue
92/// it, so instrumentation never slows down a request.
93pub fn dispatch(event: Event) {
94    let subscribers = registry().read().expect("event registry poisoned");
95    for subscriber in subscribers.iter() {
96        if subscriber.interested_in(event.kind) {
97            subscriber.handle(&event);
98        }
99    }
100}
101
102/// True when at least one subscriber is listening, so callers can skip building
103/// an expensive event payload nobody will read.
104pub fn has_subscribers() -> bool {
105    !registry().read().expect("event registry poisoned").is_empty()
106}
107
108/// Remove every subscriber. Intended for tests.
109pub fn clear_subscribers() {
110    registry().write().expect("event registry poisoned").clear();
111}
112
113/// Time a block and dispatch an event with its duration.
114pub fn timed<T>(kind: &'static str, fields: impl FnOnce() -> Vec<(String, Json)>, work: impl FnOnce() -> T) -> T {
115    if !has_subscribers() {
116        return work();
117    }
118    let started = std::time::Instant::now();
119    let result = work();
120    let elapsed = started.elapsed();
121
122    let mut event = Event::new(kind).took(elapsed);
123    for (key, value) in fields() {
124        event.fields.insert(key, value);
125    }
126    dispatch(event);
127    result
128}
129
130#[cfg(test)]
131mod tests {
132    use super::*;
133    use std::sync::atomic::{AtomicUsize, Ordering};
134
135    #[test]
136    fn subscribers_receive_dispatched_events() {
137        clear_subscribers();
138        let seen = Arc::new(AtomicUsize::new(0));
139        let counter = Arc::clone(&seen);
140
141        subscribe(move |event: &Event| {
142            assert_eq!(event.kind, "http.request");
143            assert_eq!(event.field("path").and_then(Json::as_str), Some("/users"));
144            counter.fetch_add(1, Ordering::SeqCst);
145        });
146
147        Event::new("http.request").with("path", "/users").dispatch();
148        assert_eq!(seen.load(Ordering::SeqCst), 1);
149        clear_subscribers();
150    }
151
152    #[test]
153    fn timed_records_a_duration() {
154        clear_subscribers();
155        let millis = Arc::new(RwLock::new(None));
156        let sink = Arc::clone(&millis);
157
158        subscribe(move |event: &Event| {
159            *sink.write().unwrap() = event.duration_ms();
160        });
161
162        let result = timed("db.query", || vec![("sql".to_string(), Json::from("select 1"))], || 21 * 2);
163
164        assert_eq!(result, 42);
165        assert!(millis.read().unwrap().is_some());
166        clear_subscribers();
167    }
168}
rustlavel_core/events.rs

rustlavel_core/
events.rs
