ek_event/

lib.rs

//! ek-event —— 自包含事件引擎(Rust 复刻 enkeep_trade/event/engine.py)。
//!
//! 语义对齐原版:
//! - 单分发线程串行处理(一个 type 的 handler 按注册顺序执行)
//! - 无界队列
//! - 独立定时线程,默认每 1 秒投递一个 [`EVENT_TIMER`] 事件
//! - `stop()` 后队列中未处理的事件被丢弃(对齐原版)
//!
//! 对原版的修正:
//! - handler 表全程 RwLock 保护
//! - handler panic 被 catch_unwind 隔离,不再杀掉分发线程
//! - 注册返回 [`HandlerId`],注销按 id(原版依赖函数对象相等,不可复刻)

pub mod engine;
pub mod event;
pub mod handler;

pub use engine::{EventEngine, PutError};
pub use event::{Event, EventData};
pub use handler::HandlerFn;
pub use handler::HandlerId;
pub use handler::HandlerRegistry;

/// 定时事件类型串(与原版 `EVENT_TIMER = "eTimer"` 对齐)。
pub const EVENT_TIMER: &str = "eTimer";

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex};
    use std::time::Duration;

    /// 轮询条件直至成立,默认上限约 2 秒;结束前再判定一次(最后一次可能恰好在睡眠窗口内完成)。
    fn wait_for(cond: impl Fn() -> bool) -> bool {
        for _ in 0..200 {
            if cond() {
                return true;
            }
            std::thread::sleep(Duration::from_millis(10));
        }
        cond()
    }

    // T3: type handler 先执行,general handler 后执行(各按注册顺序)
    #[test]
    fn dispatch_order_type_then_general() {
        let engine = EventEngine::new();
        engine.start();
        let seq: Arc<spin_free_seq::Seq> = Arc::default();
        let s1 = seq.clone();
        engine.register("eTick.", Arc::new(move |_| s1.record(0)));
        let s2 = seq.clone();
        engine.register_general(Arc::new(move |_| s2.record(1)));
        // 聚合事件的 type 就是 "eTick." 本身(对齐 enkeep gateway: on_event(EVENT_TICK, tick));
        // "eTick.600000" 是另一个精确 type,不会命中此处 handler。
        engine.put(Event::new("eTick.")).unwrap();
        for _ in 0..50 {
            if seq.is(0, 1) {
                break;
            }
            std::thread::sleep(Duration::from_millis(10));
        }
        // 期望值是 3(bit0 由 tick 置位,bit1 由 general handler 在 tick 或
        // 周期 TIMER 事件上都可能置位),不是仅按单事件算出的 2。
        assert_eq!(seq.0.load(Ordering::SeqCst) & 3, 3);
        engine.stop();
    }

    // T4: 定时事件周期
    #[test]
    fn timer_event_fires() {
        let engine = EventEngine::with_interval(Duration::from_millis(50));
        engine.start();
        let cnt = Arc::new(AtomicUsize::new(0));
        let c2 = cnt.clone();
        engine.register(
            EVENT_TIMER,
            Arc::new(move |_| {
                c2.fetch_add(1, Ordering::SeqCst);
            }),
        );
        std::thread::sleep(Duration::from_millis(400));
        let n = cnt.load(Ordering::SeqCst);
        assert!(n >= 3, "timer fired too few times: {n}");
        engine.stop();
    }

    // T1: HandlerId 语义——每次注册产出全局唯一 id(防重复由上层负责)
    #[test]
    fn duplicate_registration_yields_unique_ids() {
        let engine = EventEngine::new(); // 无需 start:注册与分发解耦
        let bump = Arc::new(AtomicUsize::new(0));
        let b1 = bump.clone();
        let id1 = engine.register(
            "eTick.",
            Arc::new(move |_| {
                b1.fetch_add(1, Ordering::SeqCst);
            }),
        );
        let b2 = bump.clone();
        let id2 = engine.register(
            "eTick.",
            Arc::new(move |_| {
                b2.fetch_add(1, Ordering::SeqCst);
            }),
        );
        assert_ne!(id1, id2, "each registration must get a unique HandlerId");
        assert!(engine.unregister(id1));
        assert!(engine.unregister(id2));
        assert!(!engine.unregister(id1), "double unregister must be false");
    }

    // T5: handler panic 被 catch_unwind 隔离——后续 handler 与后续事件继续处理
    #[test]
    fn handler_panic_does_not_kill_dispatcher() {
        let engine = EventEngine::new();
        engine.start();
        let cnt = Arc::new(AtomicUsize::new(0));
        // 排在第一个的 handler 必 panic
        let _ = engine.register(
            "eTick.",
            Arc::new(move |_| {
                panic!("boom");
            }),
        );
        let c2 = cnt.clone();
        engine.register(
            "eTick.",
            Arc::new(move |_| {
                c2.fetch_add(1, Ordering::SeqCst);
            }),
        );
        engine.put(Event::new("eTick.")).unwrap();
        assert!(
            wait_for(|| cnt.load(Ordering::SeqCst) >= 1),
            "handler after a panicking one must still run"
        );
        engine.put(Event::new("eTick.")).unwrap();
        assert!(
            wait_for(|| cnt.load(Ordering::SeqCst) >= 2),
            "events dispatched after a handler panic must still be processed"
        );
        engine.stop();
    }

    // T6: stop 后 put 报 Stopped;stop 重复调用幂等
    #[test]
    fn put_after_stop_fails_and_stop_is_idempotent() {
        let engine = EventEngine::new();
        engine.start();
        engine.stop();
        assert!(!engine.is_active());
        assert_eq!(engine.put(Event::new("eTick.")), Err(PutError::Stopped));
        engine.stop(); // 二次 stop:幂等,不 panic
        engine.stop(); // 三次
        assert_eq!(engine.pending_events(), 0, "stop must reset pending count");
    }

    // T8: 同一 type 的事件按 put 顺序送达(单分发线程串行)
    #[test]
    fn same_type_events_delivered_in_put_order() {
        let engine = EventEngine::new();
        engine.start();
        let seen: Arc<Mutex<Vec<usize>>> = Arc::default();
        let s = seen.clone();
        let _ = engine.register(
            "eTick.",
            Arc::new(move |e| {
                let v = *e.data.downcast_ref::<usize>().expect("usize payload");
                s.lock().unwrap().push(v);
            }),
        );
        const N: usize = 200;
        for i in 0..N {
            engine.put(Event::with_data("eTick.", Arc::new(i))).unwrap();
        }
        assert!(
            wait_for(|| seen.lock().unwrap().len() >= N),
            "all events must be delivered"
        );
        let expected: Vec<usize> = (0..N).collect();
        assert_eq!(
            *seen.lock().unwrap(),
            expected,
            "delivery must preserve put order"
        );
        engine.stop();
    }

    // T7(轻量版):多线程经 Arc 共享引擎并发 put / start 可在任意线程调用
    // —— 能编译即证明 EventEngine: Sync + Send(裸 Receiver / &mut stop 的旧缺陷在此断言)
    #[test]
    fn event_engine_is_sync_and_concurrent_put_works() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<EventEngine>();

        let engine = Arc::new(EventEngine::new());
        engine.start(); // &self,可在共享后启动
        let cnt = Arc::new(AtomicUsize::new(0));
        let c = cnt.clone();
        engine.register(
            "eTick.",
            Arc::new(move |_| {
                c.fetch_add(1, Ordering::SeqCst);
            }),
        );
        let mut handles = Vec::new();
        for _ in 0..4 {
            let e = engine.clone();
            handles.push(std::thread::spawn(move || {
                for _ in 0..50 {
                    e.put(Event::new("eTick.")).unwrap();
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        const N: usize = 200;
        assert!(
            wait_for(|| cnt.load(Ordering::SeqCst) >= N),
            "all cross-thread events must be delivered"
        );
        engine.stop();
    }

    // 内部测试用辅助:记录到达顺序(仅测试内使用)
    mod spin_free_seq {
        use std::sync::atomic::{AtomicI64, Ordering};
        pub struct Seq(pub AtomicI64);
        impl Seq {
            pub fn record(&self, i: i64) {
                let _ = self.0.fetch_or(1 << i, Ordering::SeqCst);
            }
            pub fn is(&self, a: i64, b: i64) -> bool {
                self.0.load(Ordering::SeqCst) == (1 << a) + (1 << b)
            }
        }
        impl Default for Seq {
            fn default() -> Self {
                Self(AtomicI64::new(0))
            }
        }
    }
}