caelix_core/
events.rs

1use crate::{BoxFuture, Container, Injectable, Module, ModuleMetadata, Result};
2use futures_util::{StreamExt, stream, stream::BoxStream};
3use std::{
4    any::{Any, TypeId},
5    collections::HashMap,
6    marker::PhantomData,
7    sync::{Arc, RwLock},
8};
9use tokio::sync::broadcast;
10use tokio_stream::wrappers::BroadcastStream;
11use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
12
13/// Default capacity for per-event-type broadcast channels used by [`EventBus::subscribe`].
14const EVENT_BROADCAST_CAPACITY: usize = 256;
15
16#[diagnostic::on_unimplemented(
17    message = "`{Self}` is not an event handler for `{E}`",
18    label = "missing `impl EventHandler<{E}> for {Self}`",
19    note = "add `impl EventHandler<{E}> for {Self}` or use the correct event type in `.event_handler_for::<Event, Handler>()`"
20)]
21/// Public Caelix extension trait `EventHandler`.
22pub trait EventHandler<E>: Send + Sync + 'static {
23    /// Public Caelix API.
24    fn handle(&self, event: E) -> BoxFuture<'_, Result<()>>;
25}
26
27#[diagnostic::on_unimplemented(
28    message = "`{Self}` cannot be registered with `.event_handler::<{Self}>()` yet",
29    label = "missing `impl RegisterableEventHandler for {Self}`",
30    note = "add `impl RegisterableEventHandler for {Self} {{ type Event = YourEvent; }}` or use `.event_handler_for::<YourEvent, {Self}>()`"
31)]
32/// Public Caelix extension trait `RegisterableEventHandler`.
33pub trait RegisterableEventHandler: Injectable {
34    /// Public Caelix API.
35    type Event: Clone + Send + Sync + 'static;
36
37    /// Public Caelix API.
38    fn register_into(handler: Arc<Self>, bus: &EventBus) -> Result<()>
39    where
40        Self: Sized + EventHandler<Self::Event>,
41    {
42        bus.register::<Self::Event, Self>(handler)
43    }
44}
45
46trait ErasedHandler: Send + Sync {
47    fn handle_erased(&self, event: &dyn Any) -> BoxFuture<'_, Result<()>>;
48}
49
50struct TypedHandler<E, H> {
51    handler: Arc<H>,
52    _marker: PhantomData<E>,
53}
54
55impl<E, H> ErasedHandler for TypedHandler<E, H>
56where
57    E: Clone + Send + Sync + 'static,
58    H: EventHandler<E>,
59{
60    fn handle_erased(&self, event: &dyn Any) -> BoxFuture<'_, Result<()>> {
61        let event = match event.downcast_ref::<E>() {
62            Some(event) => event.clone(),
63            None => {
64                return Box::pin(async {
65                    Err(crate::exception::startup_error("event type mismatch"))
66                });
67            }
68        };
69        let handler = self.handler.clone();
70
71        Box::pin(async move { handler.handle(event).await })
72    }
73}
74
75struct EventChannel<E> {
76    tx: broadcast::Sender<E>,
77}
78
79/// Public Caelix type `EventBus`.
80pub struct EventBus {
81    handlers: RwLock<HashMap<TypeId, Vec<Arc<dyn ErasedHandler>>>>,
82    broadcasts: RwLock<HashMap<TypeId, Box<dyn Any + Send + Sync>>>,
83}
84
85impl EventBus {
86    /// Runs the `new` public API operation.
87    pub fn new() -> Self {
88        Self {
89            handlers: RwLock::new(HashMap::new()),
90            broadcasts: RwLock::new(HashMap::new()),
91        }
92    }
93
94    /// Runs the `register` public API operation.
95    pub fn register<E, H>(&self, handler: Arc<H>) -> Result<()>
96    where
97        E: Clone + Send + Sync + 'static,
98        H: EventHandler<E>,
99    {
100        let wrapped = Arc::new(TypedHandler::<E, H> {
101            handler,
102            _marker: PhantomData,
103        });
104
105        self.handlers
106            .write()
107            .map_err(|_| crate::exception::startup_error("event handler registry lock poisoned"))?
108            .entry(TypeId::of::<E>())
109            .or_default()
110            .push(wrapped);
111        Ok(())
112    }
113
114    /// Create the per-type broadcast channel if missing; used only by [`Self::subscribe`].
115    fn ensure_sender<E>(&self) -> Result<broadcast::Sender<E>>
116    where
117        E: Clone + Send + Sync + 'static,
118    {
119        let type_id = TypeId::of::<E>();
120
121        {
122            let broadcasts = self.broadcasts.read().map_err(|_| {
123                crate::exception::startup_error("event broadcast registry lock poisoned")
124            })?;
125            if let Some(channel) = broadcasts.get(&type_id) {
126                return channel
127                    .downcast_ref::<EventChannel<E>>()
128                    .ok_or_else(|| crate::exception::startup_error("event broadcast type mismatch"))
129                    .map(|channel| channel.tx.clone());
130            }
131        }
132
133        let mut broadcasts = self.broadcasts.write().map_err(|_| {
134            crate::exception::startup_error("event broadcast registry lock poisoned")
135        })?;
136        let channel = broadcasts.entry(type_id).or_insert_with(|| {
137            let (tx, _rx) = broadcast::channel::<E>(EVENT_BROADCAST_CAPACITY);
138            Box::new(EventChannel { tx })
139        });
140
141        channel
142            .downcast_ref::<EventChannel<E>>()
143            .ok_or_else(|| crate::exception::startup_error("event broadcast type mismatch"))
144            .map(|channel| channel.tx.clone())
145    }
146
147    /// Existing channel only — does not allocate. Used by [`Self::emit`].
148    fn existing_sender<E>(&self) -> Result<Option<broadcast::Sender<E>>>
149    where
150        E: Clone + Send + Sync + 'static,
151    {
152        self.broadcasts
153            .read()
154            .map_err(|_| crate::exception::startup_error("event broadcast registry lock poisoned"))?
155            .get(&TypeId::of::<E>())
156            .map(|channel| {
157                channel
158                    .downcast_ref::<EventChannel<E>>()
159                    .ok_or_else(|| crate::exception::startup_error("event broadcast type mismatch"))
160                    .map(|channel| channel.tx.clone())
161            })
162            .transpose()
163    }
164
165    /// Live stream of events of type `E`. Receives events after subscription;
166    /// earlier events are not replayed. Slow consumers may lag and drop events
167    /// (see broadcast capacity).
168    ///
169    /// Creates the broadcast channel for `E` on first subscribe. [`Self::emit`]
170    /// only fans out when a channel already exists (i.e. someone has subscribed).
171    pub fn subscribe<E>(&self) -> BoxStream<'static, Result<E>>
172    where
173        E: Clone + Send + Sync + 'static,
174    {
175        match self.ensure_sender::<E>() {
176            Ok(tx) => BroadcastStream::new(tx.subscribe())
177                .filter_map(|item| async move {
178                    match item {
179                        Ok(event) => Some(Ok(event)),
180                        Err(BroadcastStreamRecvError::Lagged(_)) => {
181                            tracing::warn!("event bus subscriber lagged; dropped events");
182                            None
183                        }
184                    }
185                })
186                .boxed(),
187            Err(err) => stream::once(async move { Err(err) }).boxed(),
188        }
189    }
190
191    /// Run registered handlers in order, then fan out to live subscribers.
192    ///
193    /// Handlers run first so a failed handler aborts `emit` and does **not**
194    /// publish to stream subscribers. Broadcast channels are only used when
195    /// created by a prior [`Self::subscribe`] — emit alone never allocates one.
196    pub async fn emit<E>(&self, event: E) -> Result<()>
197    where
198        E: Clone + Send + Sync + 'static,
199    {
200        let handlers = self
201            .handlers
202            .read()
203            .map_err(|_| crate::exception::startup_error("event handler registry lock poisoned"))?
204            .get(&TypeId::of::<E>())
205            .cloned();
206
207        if let Some(handlers) = handlers {
208            for handler in handlers {
209                handler.handle_erased(&event).await?;
210            }
211        }
212
213        if let Some(tx) = self.existing_sender::<E>()? {
214            let _ = tx.send(event);
215        }
216
217        Ok(())
218    }
219
220    /// Runs the `handler_count` public API operation.
221    pub fn handler_count<E>(&self) -> Result<usize>
222    where
223        E: Clone + Send + Sync + 'static,
224    {
225        Ok(self
226            .handlers
227            .read()
228            .map_err(|_| crate::exception::startup_error("event handler registry lock poisoned"))?
229            .get(&TypeId::of::<E>())
230            .map_or(0, Vec::len))
231    }
232}
233
234impl Default for EventBus {
235    fn default() -> Self {
236        Self::new()
237    }
238}
239
240impl Injectable for EventBus {
241    fn dependencies() -> Vec<crate::ProviderDependency> {
242        crate::provider_dependencies![]
243    }
244
245    fn create(_container: &Container) -> BoxFuture<'_, Result<Self>> {
246        Box::pin(async { Ok(Self::new()) })
247    }
248}
249
250/// Public Caelix type `EventModule`.
251pub struct EventModule;
252
253impl Module for EventModule {
254    fn register() -> ModuleMetadata {
255        ModuleMetadata::new()
256            .provider::<EventBus>()
257            .export::<EventBus>()
258    }
259}
260
261/// Public Caelix type `EventHandlerDef`.
262pub struct EventHandlerDef {
263    type_id: TypeId,
264    type_name: &'static str,
265    register_fn: Box<dyn Fn(&Container) -> Result<()> + Send + Sync>,
266}
267
268impl EventHandlerDef {
269    pub(crate) fn of<H>() -> Self
270    where
271        H: RegisterableEventHandler + EventHandler<H::Event>,
272    {
273        Self {
274            type_id: TypeId::of::<H>(),
275            type_name: std::any::type_name::<H>(),
276            register_fn: Box::new(|container| {
277                let handler = container.resolve::<H>()?;
278                let bus = container.resolve::<EventBus>()?;
279                H::register_into(handler, &bus)
280            }),
281        }
282    }
283
284    pub(crate) fn for_event<E, H>() -> Self
285    where
286        E: Clone + Send + Sync + 'static,
287        H: Injectable + EventHandler<E>,
288    {
289        Self {
290            type_id: TypeId::of::<H>(),
291            type_name: std::any::type_name::<H>(),
292            register_fn: Box::new(|container| {
293                let handler = container.resolve::<H>()?;
294                let bus = container.resolve::<EventBus>()?;
295                bus.register::<E, H>(handler)
296            }),
297        }
298    }
299
300    pub(crate) fn assert_registered_or_declared(
301        &self,
302        declared: &std::collections::HashSet<TypeId>,
303    ) -> Result<()> {
304        if declared.contains(&self.type_id) {
305            return Ok(());
306        }
307
308        Err(crate::exception::startup_error(format!(
309            "missing event handler provider at startup: {} was declared by module metadata but was not registered as a provider",
310            self.type_name
311        )))
312    }
313
314    pub(crate) fn register(&self, container: &Container) -> Result<()> {
315        (self.register_fn)(container)
316    }
317}
caelix_core/events.rs

caelix_core/
events.rs
