GitRust
mod.rs
242 lignes · 8213 octets
//! Generic pub/sub bus for real-time notifications. //! //! Central plumbing for features that need to push events to subscribed //! consumers — Server-Sent Events streams, live progress bars, CI logs, //! wall-activity feeds, etc. Each consumer app picks an event type `T` and //! gets `publish` + filtered `subscribe` on top of a single in-memory //! [`tokio::sync::broadcast`] channel. //! //! # Multi-instance deployments //! //! The built-in `InMemoryBus` is single-process only. For multi-instance //! (horizontal scaling) the trait is stable enough to add a `redis-bus` //! feature later with the same signatures — out of scope for v1. //! //! # Example //! //! ```no_run //! use std::sync::Arc; //! use rustwarden_core::notifications::{InMemoryBus, NotificationBus}; //! //! #[derive(Clone, serde::Serialize)] //! enum AppEvent { //! ImportProgress { job_id: uuid::Uuid, pct: u8 }, //! Ping, //! } //! //! # async fn usage() { //! let bus: Arc<dyn NotificationBus<AppEvent>> = Arc::new(InMemoryBus::<AppEvent>::new(128)); //! //! // Consumer: only "ImportProgress" events. //! let filter = Arc::new(|e: &AppEvent| matches!(e, AppEvent::ImportProgress { .. })); //! let _stream = bus.subscribe(filter); //! //! bus.publish(AppEvent::Ping).await; //! # } //! ``` pub mod sse; use async_trait::async_trait; use futures::stream::{Stream, StreamExt}; use std::pin::Pin; use std::sync::Arc; use tokio::sync::broadcast; use tokio_stream::wrappers::BroadcastStream; /// Filter predicate applied to each published event before delivery to a /// subscriber. `Arc<dyn Fn>` so subscribers can share the filter without /// boxing it again and so it is trivially clonable. pub type EventFilter<T> = Arc<dyn Fn(&T) -> bool + Send + Sync>; /// Abstract publish/subscribe bus for cloneable event types. /// /// Consumers hold an `Arc<dyn NotificationBus<T>>` so the concrete backend /// (in-memory today, Redis tomorrow) is swappable without touching call sites. #[async_trait] pub trait NotificationBus<T>: Send + Sync where T: Clone + Send + Sync + 'static, { /// Broadcast an event to every live subscriber. Drops the event silently /// when no subscriber is registered — the bus does not persist. async fn publish(&self, event: T); /// Subscribe with a filter predicate. Only events for which `filter` /// returns `true` reach the returned stream. /// /// A subscriber that falls behind the channel capacity drops the oldest /// missed events (standard `tokio::sync::broadcast` semantics); the /// stream keeps running. fn subscribe(&self, filter: EventFilter<T>) -> Pin<Box<dyn Stream<Item = T> + Send>>; } /// In-memory broadcast-backed implementation. Cheap, lock-free on the hot /// path, suitable for single-process deployments. pub struct InMemoryBus<T: Clone + Send + Sync + 'static> { tx: broadcast::Sender<T>, } impl<T: Clone + Send + Sync + 'static> InMemoryBus<T> { /// Build a bus with the given channel capacity. /// /// When more than `capacity` events pile up for a slow subscriber, the /// oldest are dropped — the stream yields nothing for them but does not /// terminate. Pick capacity proportional to peak burst rate × worst-case /// subscriber latency. pub fn new(capacity: usize) -> Self { let (tx, _) = broadcast::channel(capacity); Self { tx } } /// Current number of live subscribers. Useful for metrics / tests. pub fn receiver_count(&self) -> usize { self.tx.receiver_count() } } impl<T: Clone + Send + Sync + 'static> Default for InMemoryBus<T> { fn default() -> Self { Self::new(256) } } #[async_trait] impl<T> NotificationBus<T> for InMemoryBus<T> where T: Clone + Send + Sync + 'static, { async fn publish(&self, event: T) { // `send` returns Err only when there are zero receivers — that's fine // for a transient bus (no persistence guarantee). let _ = self.tx.send(event); } fn subscribe(&self, filter: EventFilter<T>) -> Pin<Box<dyn Stream<Item = T> + Send>> { let rx = self.tx.subscribe(); let stream = BroadcastStream::new(rx) // Drop Lagged events (slow consumer) instead of ending the stream. .filter_map(|res| async move { res.ok() }) .filter(move |event| { let keep = filter(event); async move { keep } }); Box::pin(stream) } } #[cfg(test)] mod tests { use super::*; use std::time::Duration; use tokio::time::timeout; #[derive(Clone, Debug, PartialEq, Eq)] enum TestEvent { Ping, Import { job: u32, pct: u8 }, } fn pass_all<T>() -> EventFilter<T> { Arc::new(|_| true) } #[tokio::test] async fn publish_with_no_subscribers_is_noop() { let bus: InMemoryBus<TestEvent> = InMemoryBus::new(4); bus.publish(TestEvent::Ping).await; assert_eq!(bus.receiver_count(), 0); } #[tokio::test] async fn subscriber_receives_published_events() { let bus: InMemoryBus<TestEvent> = InMemoryBus::new(4); let mut stream = bus.subscribe(pass_all()); bus.publish(TestEvent::Ping).await; bus.publish(TestEvent::Import { job: 1, pct: 50 }).await; let first = timeout(Duration::from_millis(200), stream.next()) .await .unwrap() .unwrap(); let second = timeout(Duration::from_millis(200), stream.next()) .await .unwrap() .unwrap(); assert_eq!(first, TestEvent::Ping); assert_eq!(second, TestEvent::Import { job: 1, pct: 50 }); } #[tokio::test] async fn multiple_subscribers_each_receive_the_event() { let bus: InMemoryBus<TestEvent> = InMemoryBus::new(4); let mut a = bus.subscribe(pass_all()); let mut b = bus.subscribe(pass_all()); bus.publish(TestEvent::Ping).await; let a_got = timeout(Duration::from_millis(200), a.next()) .await .unwrap() .unwrap(); let b_got = timeout(Duration::from_millis(200), b.next()) .await .unwrap() .unwrap(); assert_eq!(a_got, TestEvent::Ping); assert_eq!(b_got, TestEvent::Ping); } #[tokio::test] async fn filter_drops_non_matching_events() { let bus: InMemoryBus<TestEvent> = InMemoryBus::new(4); let only_imports: EventFilter<TestEvent> = Arc::new(|e| matches!(e, TestEvent::Import { .. })); let mut stream = bus.subscribe(only_imports); bus.publish(TestEvent::Ping).await; bus.publish(TestEvent::Import { job: 7, pct: 10 }).await; let got = timeout(Duration::from_millis(200), stream.next()) .await .unwrap() .unwrap(); assert_eq!(got, TestEvent::Import { job: 7, pct: 10 }); } #[tokio::test] async fn subscriber_drop_does_not_block_publisher() { let bus: InMemoryBus<TestEvent> = InMemoryBus::new(4); { let _stream = bus.subscribe(pass_all()); // _stream drops at end of scope. } // Publisher must not hang or panic after the receiver is gone. bus.publish(TestEvent::Ping).await; assert_eq!(bus.receiver_count(), 0); } #[tokio::test] async fn slow_subscriber_loses_old_events_but_stream_keeps_going() { // Capacity 2 → third event evicts the first for a stalled consumer. let bus: InMemoryBus<TestEvent> = InMemoryBus::new(2); let mut stream = bus.subscribe(pass_all()); for pct in [10, 20, 30, 40] { bus.publish(TestEvent::Import { job: 1, pct }).await; } // We should still eventually get a tail slice of events (exact number // depends on runtime scheduling, but it must be ≥ 1). let first = timeout(Duration::from_millis(200), stream.next()) .await .unwrap() .unwrap(); match first { TestEvent::Import { .. } => {} other => panic!("expected Import event, got {other:?}"), } } }