mod_events/

dispatcher.rs

//! Main event dispatcher implementation

use crate::error::panic_payload_to_listener_error;
use crate::metrics::EventMetricsCounters;
use crate::{
    DispatchResult, Event, EventMetadata, ListenerError, ListenerId, ListenerWrapper,
    MiddlewareManager, Priority,
};
use parking_lot::RwLock;
use std::any::TypeId;
use std::collections::HashMap;
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

#[cfg(feature = "async")]
use crate::{AsyncEventResult, AsyncListenerWrapper};

#[cfg(feature = "async")]
type AsyncHandler = Arc<dyn for<'a> Fn(&'a dyn Event) -> AsyncEventResult<'a> + Send + Sync>;

/// High-performance event dispatcher
///
/// The main component of the Mod Events system. Thread-safe and optimized
/// for high-performance event dispatch with minimal overhead.
///
/// # Example
///
/// ```rust
/// use mod_events::{EventDispatcher, Event};
///
/// #[derive(Debug, Clone)]
/// struct MyEvent {
///     message: String,
/// }
///
/// impl Event for MyEvent {
///     fn as_any(&self) -> &dyn std::any::Any {
///         self
///     }
/// }
///
/// let dispatcher = EventDispatcher::new();
///
/// dispatcher.on(|event: &MyEvent| {
///     println!("Received: {}", event.message);
/// });
///
/// dispatcher.emit(MyEvent {
///     message: "Hello, World!".to_string(),
/// });
/// ```
pub struct EventDispatcher {
    listeners: Arc<RwLock<HashMap<TypeId, Vec<ListenerWrapper>>>>,
    #[cfg(feature = "async")]
    async_listeners: Arc<RwLock<HashMap<TypeId, Vec<AsyncListenerWrapper>>>>,
    next_id: AtomicUsize,
    // Per-event-type counters live behind an `Arc` so the dispatch hot
    // path can clone the counter pointer under a read lock and increment
    // its atomics without ever touching the outer map's write lock.
    metrics: Arc<RwLock<HashMap<TypeId, Arc<EventMetricsCounters>>>>,
    middleware: Arc<RwLock<MiddlewareManager>>,
}

impl EventDispatcher {
    /// Create a new event dispatcher.
    #[must_use]
    pub fn new() -> Self {
        Self {
            listeners: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(feature = "async")]
            async_listeners: Arc::new(RwLock::new(HashMap::new())),
            next_id: AtomicUsize::new(0),
            metrics: Arc::new(RwLock::new(HashMap::new())),
            middleware: Arc::new(RwLock::new(MiddlewareManager::new())),
        }
    }

    /// Subscribe to an event with a closure that can return errors.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mod_events::{EventDispatcher, Event};
    ///
    /// #[derive(Debug, Clone)]
    /// struct MyEvent {
    ///     message: String,
    /// }
    ///
    /// impl Event for MyEvent {
    ///     fn as_any(&self) -> &dyn std::any::Any {
    ///         self
    ///     }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// dispatcher.subscribe(|event: &MyEvent| {
    ///     if event.message.is_empty() {
    ///         return Err("Message cannot be empty".into());
    ///     }
    ///     println!("Message: {}", event.message);
    ///     Ok(())
    /// });
    /// ```
    pub fn subscribe<T, F>(&self, listener: F) -> ListenerId
    where
        T: Event + 'static,
        F: Fn(&T) -> Result<(), ListenerError> + Send + Sync + 'static,
    {
        self.subscribe_with_priority(listener, Priority::Normal)
    }

    /// Subscribe to an event with a specific priority.
    pub fn subscribe_with_priority<T, F>(&self, listener: F, priority: Priority) -> ListenerId
    where
        T: Event + 'static,
        F: Fn(&T) -> Result<(), ListenerError> + Send + Sync + 'static,
    {
        let type_id = TypeId::of::<T>();
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);

        let wrapper = ListenerWrapper::new(listener, priority, id);

        {
            let mut listeners = self.listeners.write();
            let event_listeners = listeners.entry(type_id).or_default();
            // Binary insertion preserves descending-priority order in O(n)
            // (one shift), avoiding the O(n log n) full re-sort the
            // previous push + sort_by_key combination performed.
            // `partition_point` finds the first index whose listener has
            // a strictly lower priority than the new one, which is also
            // where to insert. Equal-priority listeners run in
            // registration order (FIFO).
            let pos = event_listeners.partition_point(|existing| existing.priority >= priority);
            event_listeners.insert(pos, wrapper);
        }

        // Make sure a metrics entry exists for this type so `metrics()`
        // can report it even before the first dispatch.
        let _counters = self.counters_for::<T>();

        ListenerId::new(id, type_id)
    }

    /// Subscribe to an event with simple closure (no error handling).
    ///
    /// This is the most convenient method for simple event handling.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mod_events::{EventDispatcher, Event};
    ///
    /// #[derive(Debug, Clone)]
    /// struct MyEvent {
    ///     message: String,
    /// }
    ///
    /// impl Event for MyEvent {
    ///     fn as_any(&self) -> &dyn std::any::Any {
    ///         self
    ///     }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// dispatcher.on(|event: &MyEvent| {
    ///     println!("Received: {}", event.message);
    /// });
    /// ```
    pub fn on<T, F>(&self, listener: F) -> ListenerId
    where
        T: Event + 'static,
        F: Fn(&T) + Send + Sync + 'static,
    {
        self.subscribe(move |event: &T| {
            listener(event);
            Ok(())
        })
    }

    /// Subscribe an async listener at [`Priority::Normal`] (requires the
    /// `async` feature).
    ///
    /// The listener receives a borrowed event and returns a future that
    /// resolves to `Result<(), ListenerError>`.
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "async")]
    /// # {
    /// use mod_events::{Event, EventDispatcher};
    ///
    /// #[derive(Debug, Clone)]
    /// struct EmailSent { to: String }
    ///
    /// impl Event for EmailSent {
    ///     fn as_any(&self) -> &dyn std::any::Any { self }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// dispatcher.subscribe_async(|event: &EmailSent| {
    ///     let to = event.to.clone();
    ///     async move {
    ///         println!("delivered to {}", to);
    ///         Ok(())
    ///     }
    /// });
    /// # }
    /// ```
    #[cfg(feature = "async")]
    pub fn subscribe_async<T, F, Fut>(&self, listener: F) -> ListenerId
    where
        T: Event + 'static,
        F: Fn(&T) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = Result<(), ListenerError>> + Send + 'static,
    {
        self.subscribe_async_with_priority(listener, Priority::Normal)
    }

    /// Subscribe an async listener at a specific priority (requires the
    /// `async` feature).
    ///
    /// Higher-priority listeners are awaited first within a single
    /// [`Self::dispatch_async`] call.
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "async")]
    /// # {
    /// use mod_events::{Event, EventDispatcher, Priority};
    ///
    /// #[derive(Debug, Clone)]
    /// struct EmailSent { to: String }
    ///
    /// impl Event for EmailSent {
    ///     fn as_any(&self) -> &dyn std::any::Any { self }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// dispatcher.subscribe_async_with_priority(
    ///     |_event: &EmailSent| async move {
    ///         // logged before any other listener
    ///         Ok(())
    ///     },
    ///     Priority::High,
    /// );
    /// # }
    /// ```
    #[cfg(feature = "async")]
    pub fn subscribe_async_with_priority<T, F, Fut>(
        &self,
        listener: F,
        priority: Priority,
    ) -> ListenerId
    where
        T: Event + 'static,
        F: Fn(&T) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = Result<(), ListenerError>> + Send + 'static,
    {
        let type_id = TypeId::of::<T>();
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);

        let wrapper = AsyncListenerWrapper::new(listener, priority, id);

        {
            let mut async_listeners = self.async_listeners.write();
            let event_listeners = async_listeners.entry(type_id).or_default();
            // Binary insertion preserves descending-priority order in O(n).
            // See `subscribe_with_priority` for the rationale.
            let pos = event_listeners.partition_point(|existing| existing.priority >= priority);
            event_listeners.insert(pos, wrapper);
        }

        // Make sure a metrics entry exists for this type so `metrics()`
        // can report it even before the first dispatch.
        let _counters = self.counters_for::<T>();

        ListenerId::new(id, type_id)
    }

    /// Dispatch an event synchronously.
    ///
    /// Returns a [`DispatchResult`] containing per-listener outcomes.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mod_events::{EventDispatcher, Event};
    ///
    /// #[derive(Debug, Clone)]
    /// struct MyEvent {
    ///     message: String,
    /// }
    ///
    /// impl Event for MyEvent {
    ///     fn as_any(&self) -> &dyn std::any::Any {
    ///         self
    ///     }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// let result = dispatcher.dispatch(MyEvent {
    ///     message: "Hello".to_string(),
    /// });
    ///
    /// if result.all_succeeded() {
    ///     println!("All listeners handled the event successfully");
    /// }
    /// ```
    pub fn dispatch<T: Event>(&self, event: T) -> DispatchResult {
        // Update metrics
        self.update_metrics(&event);

        // Check middleware
        if !self.check_middleware(&event) {
            return DispatchResult::blocked();
        }

        let type_id = TypeId::of::<T>();
        let listeners = self.listeners.read();

        // Most dispatches succeed. `errors` stays empty (and
        // unallocated — `Vec::new()` does not allocate) on the
        // success path; we only push when a listener returns `Err`
        // or panics.
        let mut errors: Vec<ListenerError> = Vec::new();
        let mut listener_count = 0_usize;

        if let Some(event_listeners) = listeners.get(&type_id) {
            for listener in event_listeners {
                listener_count += 1;
                // `catch_unwind` keeps a panicking listener from
                // unwinding the dispatch thread (and the read lock we
                // are holding above) into the caller. Panics are
                // converted into the same `ListenerError` shape a
                // well-behaved listener would have returned, so the
                // caller's `DispatchResult::errors()` is the single
                // place to look for failures. `parking_lot::RwLock`
                // does not poison, so dropping the guard after a
                // caught panic leaves the registry in a usable state.
                match catch_unwind(AssertUnwindSafe(|| (listener.handler)(&event))) {
                    Ok(Ok(())) => {}
                    Ok(Err(err)) => errors.push(err),
                    Err(payload) => errors.push(panic_payload_to_listener_error(payload)),
                }
            }
        }

        DispatchResult::new(listener_count, errors)
    }

    /// Dispatch an event asynchronously and await every async listener
    /// in priority order (requires the `async` feature).
    ///
    /// Listeners are awaited **sequentially** in descending priority
    /// order. This preserves the priority contract — a high-priority
    /// listener completes (or errors) before the next listener is
    /// polled. Concurrent execution would lose ordering. If you want
    /// true concurrent execution, drive `spawn` (e.g. `tokio::spawn`,
    /// `async_std::task::spawn`) from inside each listener and return
    /// `Ok(())` immediately.
    ///
    /// Each listener future is wrapped in `catch_unwind`, mirroring
    /// the panic-safety guarantee of the sync [`Self::dispatch`]
    /// path. A panic during `.await` becomes a `ListenerError` in
    /// [`DispatchResult::errors`] with the prefix
    /// `"listener panicked: "`. Subsequent listeners still run.
    ///
    /// Sync listeners registered via [`Self::on`] / [`Self::subscribe`]
    /// are not invoked here; only listeners registered through
    /// [`Self::subscribe_async`] / [`Self::subscribe_async_with_priority`]
    /// are awaited.
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "async")]
    /// # async fn doc_example() {
    /// use mod_events::{Event, EventDispatcher};
    ///
    /// #[derive(Debug, Clone)]
    /// struct OrderShipped { order_id: u64 }
    ///
    /// impl Event for OrderShipped {
    ///     fn as_any(&self) -> &dyn std::any::Any { self }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// dispatcher.subscribe_async(|event: &OrderShipped| {
    ///     let id = event.order_id;
    ///     async move {
    ///         println!("notifying customer about order {}", id);
    ///         Ok(())
    ///     }
    /// });
    ///
    /// let result = dispatcher
    ///     .dispatch_async(OrderShipped { order_id: 42 })
    ///     .await;
    /// assert!(result.all_succeeded());
    /// # }
    /// ```
    #[cfg(feature = "async")]
    pub async fn dispatch_async<T: Event>(&self, event: T) -> DispatchResult {
        // Update metrics
        self.update_metrics(&event);

        // Check middleware
        if !self.check_middleware(&event) {
            return DispatchResult::blocked();
        }

        let type_id = TypeId::of::<T>();

        // Collect cloned handlers without holding the lock across await points.
        let handlers: Vec<AsyncHandler> = {
            let async_listeners = self.async_listeners.read();
            async_listeners
                .get(&type_id)
                .map(|event_listeners| {
                    event_listeners
                        .iter()
                        .map(|listener| listener.handler.clone())
                        .collect()
                })
                .unwrap_or_default()
        };

        // Wrap each listener future in `catch_unwind` so a panicking
        // listener becomes a `ListenerError` in `DispatchResult` rather
        // than unwinding the dispatching task. Mirrors the panic-safety
        // wrapping on the sync `dispatch` path.
        //
        // `AssertUnwindSafe` is required because the listener future
        // closes over arbitrary user state that may not implement
        // `UnwindSafe`. The contract is: if a listener panics, the
        // dispatcher catches it and reports it; the listener author is
        // responsible for not leaving captured state in a broken
        // condition before panicking. Same contract as the sync path.
        use futures_util::future::FutureExt;
        use std::panic::AssertUnwindSafe;

        // `errors` stays empty (and unallocated) on the success path.
        let listener_count = handlers.len();
        let mut errors: Vec<ListenerError> = Vec::new();
        for handler in handlers {
            match AssertUnwindSafe(handler(&event)).catch_unwind().await {
                Ok(Ok(())) => {}
                Ok(Err(err)) => errors.push(err),
                Err(payload) => errors.push(panic_payload_to_listener_error(payload)),
            }
        }

        DispatchResult::new(listener_count, errors)
    }

    /// Fire and forget — dispatch without inspecting the result.
    ///
    /// Use this when listeners' success or failure is not actionable at
    /// the call site (logging, fanout to passive observers). The errors
    /// returned by failing listeners are discarded; if you need them,
    /// call [`Self::dispatch`] instead.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mod_events::{EventDispatcher, Event};
    ///
    /// #[derive(Debug, Clone)]
    /// struct MyEvent {
    ///     message: String,
    /// }
    ///
    /// impl Event for MyEvent {
    ///     fn as_any(&self) -> &dyn std::any::Any {
    ///         self
    ///     }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// dispatcher.emit(MyEvent {
    ///     message: "Fire and forget".to_string(),
    /// });
    /// ```
    pub fn emit<T: Event>(&self, event: T) {
        // Update metrics
        self.update_metrics(&event);

        // Check middleware
        if !self.check_middleware(&event) {
            return;
        }

        let type_id = TypeId::of::<T>();
        let listeners = self.listeners.read();

        if let Some(event_listeners) = listeners.get(&type_id) {
            for listener in event_listeners {
                // Same panic-safety contract as `dispatch`, but the
                // outcome is intentionally discarded. This avoids
                // allocating the per-call result vector that `dispatch`
                // returns — the documented win of `emit` over
                // `dispatch` for fire-and-forget callers.
                let _ = catch_unwind(AssertUnwindSafe(|| (listener.handler)(&event)));
            }
        }
    }

    /// Add middleware that can block events.
    ///
    /// Middleware functions receive events and return `true` to allow
    /// processing or `false` to block the event.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mod_events::{EventDispatcher, Event};
    ///
    /// let dispatcher = EventDispatcher::new();
    /// dispatcher.add_middleware(|event: &dyn Event| {
    ///     println!("Processing event: {}", event.event_name());
    ///     true // Allow all events
    /// });
    /// ```
    pub fn add_middleware<F>(&self, middleware: F)
    where
        F: Fn(&dyn Event) -> bool + Send + Sync + 'static,
    {
        self.middleware.write().add(middleware);
    }

    /// Remove a previously registered listener.
    ///
    /// Returns `true` if the listener was found and removed, `false`
    /// if no listener with that id was registered (already removed,
    /// never registered, or registered against a different event type).
    ///
    /// # Example
    ///
    /// ```rust
    /// use mod_events::{Event, EventDispatcher};
    ///
    /// #[derive(Debug, Clone)]
    /// struct Tick;
    ///
    /// impl Event for Tick {
    ///     fn as_any(&self) -> &dyn std::any::Any { self }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// let id = dispatcher.on(|_: &Tick| {});
    /// assert!(dispatcher.unsubscribe(id));
    /// // Subsequent removals of the same id return false.
    /// assert!(!dispatcher.unsubscribe(id));
    /// ```
    pub fn unsubscribe(&self, listener_id: ListenerId) -> bool {
        // Try sync listeners first
        {
            let mut listeners = self.listeners.write();
            if let Some(event_listeners) = listeners.get_mut(&listener_id.type_id) {
                if let Some(pos) = event_listeners.iter().position(|l| l.id == listener_id.id) {
                    let _removed = event_listeners.remove(pos);
                    return true;
                }
            }
        }

        // Try async listeners
        #[cfg(feature = "async")]
        {
            let mut async_listeners = self.async_listeners.write();
            if let Some(event_listeners) = async_listeners.get_mut(&listener_id.type_id) {
                if let Some(pos) = event_listeners.iter().position(|l| l.id == listener_id.id) {
                    let _removed = event_listeners.remove(pos);
                    return true;
                }
            }
        }

        false
    }

    /// Get the total number of listeners (sync + async, when the
    /// `async` feature is enabled) registered for an event type.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mod_events::{Event, EventDispatcher};
    ///
    /// #[derive(Debug, Clone)]
    /// struct Tick;
    ///
    /// impl Event for Tick {
    ///     fn as_any(&self) -> &dyn std::any::Any { self }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// assert_eq!(dispatcher.listener_count::<Tick>(), 0);
    /// let _ = dispatcher.on(|_: &Tick| {});
    /// let _ = dispatcher.on(|_: &Tick| {});
    /// assert_eq!(dispatcher.listener_count::<Tick>(), 2);
    /// ```
    #[must_use]
    pub fn listener_count<T: Event + 'static>(&self) -> usize {
        let type_id = TypeId::of::<T>();
        let sync_count = self
            .listeners
            .read()
            .get(&type_id)
            .map(Vec::len)
            .unwrap_or(0);

        #[cfg(feature = "async")]
        let async_count = self
            .async_listeners
            .read()
            .get(&type_id)
            .map(Vec::len)
            .unwrap_or(0);

        #[cfg(not(feature = "async"))]
        let async_count = 0;

        sync_count + async_count
    }

    /// Get a snapshot of per-event-type [`EventMetadata`] keyed by
    /// `TypeId`.
    ///
    /// The returned map is a fresh snapshot. Subsequent dispatches do
    /// not mutate it, but the snapshot may be slightly behind the live
    /// counters because `dispatch_count` is read independently of
    /// `last_dispatch`.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mod_events::{Event, EventDispatcher};
    /// use std::any::TypeId;
    ///
    /// #[derive(Debug, Clone)]
    /// struct Tick;
    ///
    /// impl Event for Tick {
    ///     fn as_any(&self) -> &dyn std::any::Any { self }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// let _ = dispatcher.on(|_: &Tick| {});
    /// dispatcher.emit(Tick);
    /// dispatcher.emit(Tick);
    ///
    /// let snapshot = dispatcher.metrics();
    /// let meta = snapshot.get(&TypeId::of::<Tick>()).unwrap();
    /// assert_eq!(meta.dispatch_count, 2);
    /// ```
    #[must_use]
    pub fn metrics(&self) -> HashMap<TypeId, EventMetadata> {
        // Snapshot the metric counters first, then enrich each entry
        // with the live listener count derived from the registry. This
        // guarantees `listener_count` cannot drift because it is never
        // stored — it is computed at the moment of observation.
        let counters_map = self.metrics.read();
        let listeners_map = self.listeners.read();
        #[cfg(feature = "async")]
        let async_listeners_map = self.async_listeners.read();

        counters_map
            .iter()
            .map(|(type_id, counters)| {
                let mut snap = counters.snapshot();
                let sync_count = listeners_map.get(type_id).map(Vec::len).unwrap_or(0);
                #[cfg(feature = "async")]
                let async_count = async_listeners_map.get(type_id).map(Vec::len).unwrap_or(0);
                #[cfg(not(feature = "async"))]
                let async_count = 0;
                snap.listener_count = sync_count + async_count;
                (*type_id, snap)
            })
            .collect()
    }

    /// Drop every registered listener, both sync and async.
    ///
    /// Middleware and accumulated metrics are unaffected. Use
    /// [`Self::clear_middleware`] if you also need to drop the
    /// middleware chain.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mod_events::{Event, EventDispatcher};
    ///
    /// #[derive(Debug, Clone)]
    /// struct Tick;
    ///
    /// impl Event for Tick {
    ///     fn as_any(&self) -> &dyn std::any::Any { self }
    /// }
    ///
    /// let dispatcher = EventDispatcher::new();
    /// let _ = dispatcher.on(|_: &Tick| {});
    /// dispatcher.clear();
    /// assert_eq!(dispatcher.listener_count::<Tick>(), 0);
    /// ```
    pub fn clear(&self) {
        self.listeners.write().clear();

        #[cfg(feature = "async")]
        self.async_listeners.write().clear();
    }

    /// Drop every registered middleware function.
    ///
    /// Listeners and accumulated metrics are unaffected. Useful in
    /// test setup/teardown when you want to reset the middleware
    /// chain between cases without rebuilding the dispatcher.
    pub fn clear_middleware(&self) {
        self.middleware.write().clear();
    }

    /// Hot-path metric update. Tries a read-only fast path first; only
    /// promotes to a write lock if the entry doesn't exist yet.
    fn update_metrics<T: Event>(&self, _event: &T) {
        let counters = self.counters_for::<T>();
        counters.record_dispatch();
    }

    /// Look up (or create) the per-type counters. The fast path holds
    /// only a read lock; the slow path promotes to a write lock and
    /// double-checks the entry to avoid a torn-creation race.
    fn counters_for<T: Event + 'static>(&self) -> Arc<EventMetricsCounters> {
        let type_id = TypeId::of::<T>();

        if let Some(existing) = self.metrics.read().get(&type_id) {
            return Arc::clone(existing);
        }

        let mut metrics = self.metrics.write();
        Arc::clone(
            metrics
                .entry(type_id)
                .or_insert_with(|| Arc::new(EventMetricsCounters::new::<T>())),
        )
    }

    fn check_middleware(&self, event: &dyn Event) -> bool {
        self.middleware.read().process(event)
    }
}

impl Default for EventDispatcher {
    fn default() -> Self {
        Self::new()
    }
}