ek-event

自包含的 Rust 事件引擎 —— 对 enkeep_trade 平台事件核心(enkeep_trade/event/engine.py)的逐语义 Rust 复刻。

本 crate 不依赖 enkeep_trade 任何模块,也不承载任何业务知识(Tick/Order 等)。 业务事件常量、Gateway 双投递、OmsEngine、UI 桥接等生态层,由使用方在本库之上自行实现 —— 与原 Python 平台相同的分层方式;真实使用姿势见 tests/ 目录(8 个生态使用模式还原测试)。

设计文档:DESIGN.md · 测试方案与执行记录:TESTPLAN.md

特性

• 单分发线程串行消费:无界队列;同一 type 的事件按投递顺序送达,handler 按注册顺序执行 —— 顺序语义与 Python 版完全对齐
• 独立定时线程:默认每 1 秒投递一个 EVENT_TIMER("eTimer")事件
• 精确匹配分发:该 type 的 handler 先执行,通用 handler 后执行
• 线程友好:EventEngine: Send + Sync,可经 Arc 供任意多线程共享投递/注册(对齐真实平台各组件共享同一引擎的用法)
• panic 隔离:单个 handler panic 不再杀死分发线程(对 Python 版的三项必要修正之一)
• handler 表全程 RwLock 保护(原版无锁,依赖 GIL)
• 零业务依赖:仅 crossbeam-channel 0.5 一个三方件;载荷为 Arc<dyn Any + Send + Sync>

快速上手

[dependencies]
ek-event = "0.1.0"

use ek_event::{Event, EventEngine};
use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};

struct Tick { symbol: String, price: f64 }

fn main() {
    // 引擎对象可任意共享:Arc<EventEngine>(put / register / start / stop 均为 &self)
    let engine = Arc::new(EventEngine::new()); // 定时间隔默认 1s
    engine.start();

    // handler 排第一个,panic 时引擎不会阵亡
    let hits = Arc::new(AtomicUsize::new(0));
    let h = hits.clone();
    let _oms = engine.register("eTick.", Arc::new(move |e: &Event| {
        let t = e.data.downcast_ref::<Tick>().unwrap();
        h.fetch_add(1, Ordering::SeqCst);
        println!("{t:?}");
    }));

    // 面板组件订阅某一只标的:注册精确 type "eTick.<symbol>"
    let panel = engine.register("eTick.600000", Arc::new(|e: &Event| {
        let _ = e.data.downcast_ref::<Tick>().unwrap(); // ... 面板刷新
    }));

    // 投递事件(无界队列;聚合事件的 type 就是 "eTick." 字符串本身)
    let tick = Arc::new(Tick { symbol: "600000".into(), price: 4100.0 });
    engine.put(Event::with_data("eTick.", tick.clone())).unwrap(); // 聚合 type
    engine.put(Event::with_data("eTick.600000", tick)).unwrap();   // 逐品种 type
    engine.unregister(panel); // 面板关闭时注销(和 UI 生命周期一致)

    engine.stop(); // 幂等;stop 后 put 返回 Err(PutError::Stopped)
}

要点:

• 聚合事件的 type 就是 "eTick." 字符串本身;"eTick.600000" 是另一个精确 key,没有前缀匹配。 OmsEngine 角色收聚合、面板角色逐品种订阅,两者并行不悖(与 BaseGateway.on_tick 双投递一致)。
• 定时事件:engine.register("eTimer", ...) 或用常量 ek_event::EVENT_TIMER。

线程安全矩阵

API 一览

EventEngine::new() / with_interval(Duration)      // 定时间隔,默认 1s
    .start(&self)                                 // 启动分发线程 + 定时线程;幂等
    .stop(&self)                                  // join 两线程;残余事件按"丢弃"语义清空;stop 后不可重启
    .put(Event) -> Result<(), PutError::Stopped>  // start 前投递合法(start 后分发,对齐原版 Queue.put)
    .register(type, handler) -> HandlerId         // 精确 type 注册;type 接受 &str / String / format!(...)
    .unregister(HandlerId) -> bool                // 同样可用于 general id;空类型表自动回收
    .register_general(handler) -> HandlerId       // 通用 handler:收所有事件(先 type 后 general)
    .unregister_general(HandlerId) -> bool
    .is_active() -> bool
    .pending_events() -> u64                      // 已入队未分发数(近似);stop 时清零

Event::new(type)                                  // 无载荷(data = Arc<()>),如哨兵事件
Event::with_data(type, payload)                   // 载荷 Arc<dyn Any + Send + Sync>,handler 内 downcast

与原 Python 版的行为差异

质量状态

• 27 断言组 / 19 个测试函数 / 9 个测试二进制全部通过;cargo clippy --all-targets -- -D warnings 零告警;连跑 10 次零失败
• 测试样板来自 enkeep_trade 的真实使用方式:Gateway 聚合+逐品种双投、UI open/close/open 生命周期、 rpc 全事件转发器过滤、定时计数节流、成对哨兵事件、批量 list 载荷、多引擎实例隔离、40 万事件多线程洪峰(不丢不重)—— 详见 TESTPLAN.md
• 工具链:Rust Edition 2024(rustc ≥ 1.85);依赖:crossbeam-channel 0.5.17(单 crate、零传递依赖)
• 构建 / 测试:cargo test(测试约 3 秒,调度细节与通过标准见 TESTPLAN.md §2)