ek_event/

engine.rs

//! `EventEngine`:无界队列 + 单分发线程 + 定时线程。
//!
//! 线程模型(对齐 enkeep_trade 的实际用法:各组件跨线程共享同一引擎):
//! - **`EventEngine` 满足 `Send + Sync`**:可经 `Arc<EventEngine>` 供任意多线程共享,
//!   `put / register / unregister / register_general / unregister_general / is_active /
//!   pending_events / start / stop` 全部可从任意线程调用
//! - 分发端保持**单线程串行**(对齐原版):同一 type 的事件按 put 顺序送达,
//!   同 type 的 handler 按注册顺序执行,general handler 随后
//! - handler 内部若需并行,由 handler 自行分流(引擎不替业务并行)—— 与原版一致
//!
//! 生命周期与原版对齐:
//! - `start()` 幂等;分发线程 `recv_timeout(1s)` 轮询 `active`,
//!   对齐原版 `_run` 的 `get(block=True, timeout=1)`
//! - `put()` 在 start **之前**投递合法(事件入队,随 start 分发,对齐原版 `Queue.put`)
//! - `stop()` 置 inactive 后 join 两线程;**队列残余事件被丢弃**(对齐原版),
//!   并把 `pending` 计数清零;stop 后引擎不可重启(receiver 已被消费,与原版线程死透一致)
//! - 使用方如需"等排空":所有 put 完成后轮询 [`EventEngine::pending_events`] 归零再 stop
//!
//! handler panic 被 catch_unwind 隔离(修正项,见 DESIGN.md §2)。
//!
//! 队列实现:crossbeam-channel `unbounded()`
//! - 对应原版 Python `queue.Queue()` 无界语义
//! - MPMC 无锁快路径(单字 CAS),百万级事件/秒热路径开销低
//! - Sender/Receiver 均可克隆 → 后续把 tick 等热点事件分流到独立 worker 时无需换型
//! - `send` 在无任何 Receiver 时报 `SendError` → 映射 [`PutError::Stopped`]

use crate::event::Event;
use crate::handler::{HandlerFn, HandlerId, HandlerRegistry};
use crossbeam_channel::{Receiver, RecvTimeoutError, SendError, Sender};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration;

/// put 失败原因(已 stop 或分发线程已退出)。
/// 注意:start **之前** put 是合法的(事件入队,随 start 分发,对齐原版)。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PutError {
    /// 事件未入队。
    Stopped,
}

/// 事件引擎。**Send + Sync**:可被 `Arc` 共享于多线程。
pub struct EventEngine {
    /// 定时间隔(默认 1s)。构造后不变,无需锁。
    interval: Duration,
    /// 队列发送端;`stop()` 写 None 断开,加速分发线程退出。
    /// `RwLock` 承担就地可变 + 让引擎满足 Sync(“crossbeam Sender 本身 Sync,锁只为 stop 能写”)。
    sender: RwLock<Option<Sender<Event>>>,
    /// 队列接收端,start 时 take 交给分发线程。
    /// crossbeam `Receiver` 是 Send 非 Sync,Mutex 包一层即取 Sync(Option<Receiver>: Send)。
    receiver: Mutex<Option<Receiver<Event>>>,
    /// handler 注册表(线程安全,自带 RwLock)。
    handlers: Arc<HandlerRegistry>,
    /// 引擎存活标志。
    active: Arc<AtomicBool>,
    /// 分发线程句柄(stop 时取出 join)。
    worker: Mutex<Option<std::thread::JoinHandle<()>>>,
    /// 定时线程句柄(stop 时取出 join)。
    timer: Mutex<Option<std::thread::JoinHandle<()>>>,
    /// 已入队未分发完的事件计数(近似值,供等排空;stop 后强制清零)。
    pending: Arc<AtomicU64>,
}

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

impl EventEngine {
    /// 新引擎,默认定时间隔 1s(对齐原版 `interval: int = 1`)。
    pub fn new() -> Self {
        Self::with_interval(Duration::from_secs(1))
    }

    /// 自定义定时间隔。
    pub fn with_interval(interval: Duration) -> Self {
        let (tx, rx) = crossbeam_channel::unbounded();
        Self {
            interval,
            sender: RwLock::new(Some(tx)),
            receiver: Mutex::new(Some(rx)),
            handlers: Arc::default(),
            active: Arc::new(AtomicBool::new(false)),
            worker: Mutex::new(None),
            timer: Mutex::new(None),
            pending: Arc::new(AtomicU64::new(0)),
        }
    }

    /// 启动分发线程与定时线程;已启动 / 已 stop 时为空操作(幂等)。
    /// 可从任意线程调用。
    pub fn start(&self) {
        // 先抢 receiver:双 start 并发时只有一方拿得到,天然去重;
        // 抢不到(未启动/已 stop)直接返回,避免把 active 置回 true 造成假活。
        let Some(rx) = self.receiver.lock().unwrap().take() else {
            return;
        };
        self.active.store(true, Ordering::SeqCst);

        let worker = std::thread::Builder::new()
            .name("ek-event-dispatch".into())
            .spawn({
                let handlers = self.handlers.clone();
                let active = self.active.clone();
                let pending = self.pending.clone();
                move || run_dispatch(rx, handlers, active, pending)
            })
            .expect("spawn dispatch thread");

        let timer = std::thread::Builder::new()
            .name("ek-event-timer".into())
            .spawn({
                let sender = self.sender.read().unwrap().clone();
                let active = self.active.clone();
                let step = self.interval;
                let pending = self.pending.clone();
                move || run_timer(sender, active, step, pending)
            })
            .expect("spawn timer thread");

        *self.worker.lock().unwrap() = Some(worker);
        *self.timer.lock().unwrap() = Some(timer);
    }

    /// 投递事件。start 前投递合法(事件入队,start 后分发,对齐原版);
    /// 已 stop / 通道断开时返回 [`PutError::Stopped`]。
    pub fn put(&self, event: Event) -> Result<(), PutError> {
        self.pending.fetch_add(1, Ordering::AcqRel);
        let sent = match self.sender.read().unwrap().as_ref() {
            Some(tx) => tx.send(event),
            None => {
                self.pending.fetch_sub(1, Ordering::AcqRel);
                return Err(PutError::Stopped);
            }
        };
        match sent {
            Ok(()) => Ok(()),
            Err(SendError(_)) => {
                self.pending.fetch_sub(1, Ordering::AcqRel);
                Err(PutError::Stopped)
            }
        }
    }

    /// 注册 event_type 精确 handler,返回 [`HandlerId`] 供注销。
    /// event_type 接受 `&str` / `String` / `format!(...)` 结果(对齐原版 str 语义)。
    pub fn register(&self, event_type: impl Into<String>, f: HandlerFn) -> HandlerId {
        self.handlers.register(event_type, f)
    }

    /// 注销任意位置的 handler;true 表示确实注销了某个。
    pub fn unregister(&self, id: HandlerId) -> bool {
        self.handlers.unregister(id)
    }

    /// 注册通用 handler(收所有事件)。
    pub fn register_general(&self, f: HandlerFn) -> HandlerId {
        self.handlers.register_general(f)
    }

    /// 注销通用 handler。
    pub fn unregister_general(&self, id: HandlerId) -> bool {
        self.handlers.unregister_general(id)
    }

    /// 引擎是否在运行。
    pub fn is_active(&self) -> bool {
        self.active.load(Ordering::SeqCst)
    }

    /// 已入队、尚未分发完的事件数(近似)。
    /// 使用方在 stop 前想要"等排空",轮询此处归零即可;
    /// stop 会强制清零(残余事件按"丢弃"语义不再计数)。
    pub fn pending_events(&self) -> u64 {
        self.pending.load(Ordering::SeqCst)
    }

    /// 停止引擎并 join 两线程;幂等,可从任意线程调用。
    /// 队列残余事件被丢弃(对齐原版);stop 后引擎不可重启。
    pub fn stop(&self) {
        if !self.active.swap(false, Ordering::SeqCst) {
            return; // 已停,幂等
        }
        // 丢弃发送端:put 从此报 Stopped;定时端的 send 也会失败而退出
        *self.sender.write().unwrap() = None;
        // 先取句柄再 join(join 期间不持有锁)
        let timer_handle = self.timer.lock().unwrap().take();
        if let Some(t) = timer_handle {
            let _ = t.join();
        }
        let worker_handle = self.worker.lock().unwrap().take();
        if let Some(w) = worker_handle {
            let _ = w.join();
        }
        // 残余事件按"丢弃"语义不再计数,防止调用方误以为还在排空
        self.pending.store(0, Ordering::SeqCst);
    }
}

/// 分发线程主体:对齐原版 `_run` 的 `get(block=True, timeout=1)`。
fn run_dispatch(
    rx: Receiver<Event>,
    handlers: Arc<HandlerRegistry>,
    active: Arc<AtomicBool>,
    pending: Arc<AtomicU64>,
) {
    const POLL: Duration = Duration::from_secs(1);
    loop {
        if !active.load(Ordering::SeqCst) {
            break;
        }
        match rx.recv_timeout(POLL) {
            Ok(event) => {
                pending.fetch_sub(1, Ordering::AcqRel);
                process_event(&event, &handlers);
            }
            Err(RecvTimeoutError::Timeout) => continue, // 轮询 active 旗标
            Err(RecvTimeoutError::Disconnected) => break,
        }
    }
}

/// 定时线程主体:对齐原版 `_run_timer`(sleep(interval) 后投递 TIMER 事件)。
fn run_timer(
    sender: Option<Sender<Event>>,
    active: Arc<AtomicBool>,
    step: Duration,
    pending: Arc<AtomicU64>,
) {
    let Some(sender) = sender else { return };
    loop {
        std::thread::sleep(step);
        if !active.load(Ordering::SeqCst) {
            break;
        }
        pending.fetch_add(1, Ordering::AcqRel);
        if sender.send(Event::new(crate::EVENT_TIMER)).is_err() {
            pending.fetch_sub(1, Ordering::AcqRel);
            break;
        }
    }
}

/// 分发一个事件:该 type 的 handler 先(注册顺序),general handler 后;
/// 单个 handler 的 panic 被 catch_unwind 隔离,不影响后续 handler 与事件。
fn process_event(event: &Event, handlers: &HandlerRegistry) {
    for f in handlers.snapshot(&event.event_type) {
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(event)));
    }
}