rws_server/events/
mod.rs

1//! A typed in-process event bus (18.35.0, #93): code that does something
2//! publishes an event value; code that cares subscribes to its type.
3//!
4//! ```rust
5//! # extern crate rws_server as rust_web_server;
6//! use std::sync::{Arc, Mutex};
7//! use rust_web_server::events::EventBus;
8//!
9//! #[derive(Debug, Clone)]
10//! struct UserSignedUp { email: String }
11//!
12//! let bus = EventBus::new();
13//! let welcomed = Arc::new(Mutex::new(Vec::new()));
14//! let w = Arc::clone(&welcomed);
15//! bus.subscribe(move |e: &UserSignedUp| w.lock().unwrap().push(e.email.clone()));
16//!
17//! assert_eq!(1, bus.publish(&UserSignedUp { email: "ann@example.com".into() }));
18//! assert_eq!(vec!["ann@example.com"], *welcomed.lock().unwrap());
19//! ```
20//!
21//! Subscribers run on the publishing thread, in the order they subscribed;
22//! one that panics is skipped (and reported on stderr) without stopping the
23//! others. Slow work belongs in a job the subscriber enqueues. Clones of a
24//! bus share its subscribers. [`EventBus::forward_to_mcp`] turns an event
25//! type into MCP notifications, [`EventBus::forward_to_channel`] into
26//! [channel](crate::channels) messages.
27
28#[cfg(test)]
29mod tests;
30
31use std::any::{Any, TypeId};
32use std::collections::HashMap;
33use std::panic::{catch_unwind, AssertUnwindSafe};
34use std::sync::atomic::{AtomicU64, Ordering};
35use std::sync::{Arc, RwLock};
36
37type Subscriber = Arc<dyn Fn(&dyn Any) + Send + Sync>;
38type Subscribers = HashMap<TypeId, Vec<(SubscriptionId, Subscriber)>>;
39
40/// What [`EventBus::subscribe`] returns: pass it to
41/// [`EventBus::unsubscribe`] to stop.
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
43pub struct SubscriptionId(u64);
44
45/// Subscribers by event type. See the [module docs](self).
46#[derive(Clone, Default)]
47pub struct EventBus {
48    subscribers: Arc<RwLock<Subscribers>>,
49    next_id: Arc<AtomicU64>,
50}
51
52impl EventBus {
53    pub fn new() -> Self {
54        EventBus::default()
55    }
56
57    /// Call `handler` with every event of type `E` published from now on.
58    pub fn subscribe<E, F>(&self, handler: F) -> SubscriptionId
59    where
60        E: Any,
61        F: Fn(&E) + Send + Sync + 'static,
62    {
63        let id = SubscriptionId(self.next_id.fetch_add(1, Ordering::Relaxed));
64        let subscriber: Subscriber = Arc::new(move |event: &dyn Any| {
65            if let Some(event) = event.downcast_ref::<E>() {
66                handler(event);
67            }
68        });
69        self.subscribers.write().unwrap_or_else(|e| e.into_inner()).entry(TypeId::of::<E>()).or_default().push((id, subscriber));
70        id
71    }
72
73    /// Stop a subscription; `false` when it was already gone.
74    pub fn unsubscribe(&self, id: SubscriptionId) -> bool {
75        let mut subscribers = self.subscribers.write().unwrap_or_else(|e| e.into_inner());
76        for list in subscribers.values_mut() {
77            if let Some(position) = list.iter().position(|(i, _)| *i == id) {
78                list.remove(position);
79                return true;
80            }
81        }
82        false
83    }
84
85    /// Hand `event` to every subscriber of its type, in subscription
86    /// order; returns how many ran to completion.
87    pub fn publish<E: Any>(&self, event: &E) -> usize {
88        // Cloned out, so a subscriber may subscribe or publish itself.
89        let subscribers: Vec<Subscriber> = match self.subscribers.read().unwrap_or_else(|e| e.into_inner()).get(&TypeId::of::<E>()) {
90            Some(list) => list.iter().map(|(_, s)| Arc::clone(s)).collect(),
91            None => return 0,
92        };
93        let mut completed = 0;
94        for subscriber in subscribers {
95            match catch_unwind(AssertUnwindSafe(|| subscriber(event))) {
96                Ok(()) => completed += 1,
97                Err(_) => eprintln!("[events] a subscriber to {} panicked", std::any::type_name::<E>()),
98            }
99        }
100        completed
101    }
102
103    /// How many subscribers events of type `E` have.
104    pub fn subscribers<E: Any>(&self) -> usize {
105        self.subscribers.read().unwrap_or_else(|e| e.into_inner()).get(&TypeId::of::<E>()).map_or(0, Vec::len)
106    }
107
108    /// Send each `E` to MCP clients as the notification `method`, with
109    /// `params(event)` (a JSON object) as its params
110    /// ([`McpServer::notify`](crate::mcp::McpServer::notify)).
111    pub fn forward_to_mcp<E, F>(&self, server: crate::mcp::McpServer, method: &str, params: F) -> SubscriptionId
112    where
113        E: Any,
114        F: Fn(&E) -> String + Send + Sync + 'static,
115    {
116        let method = method.to_string();
117        self.subscribe(move |event: &E| server.notify(&method, Some(&params(event))))
118    }
119
120    #[cfg(not(target_arch = "wasm32"))]
121    /// Publish each `E` to the channel topic `topic(event)` as
122    /// `message(event)` ([`Hub::publish`](crate::channels::Hub::publish)).
123    pub fn forward_to_channel<E, T, M>(&self, hub: crate::channels::Hub, topic: T, message: M) -> SubscriptionId
124    where
125        E: Any,
126        T: Fn(&E) -> String + Send + Sync + 'static,
127        M: Fn(&E) -> String + Send + Sync + 'static,
128    {
129        self.subscribe(move |event: &E| {
130            hub.publish(&topic(event), &message(event));
131        })
132    }
133}
rws_server/events/mod.rs

rws_server/events/
mod.rs
