# ek-event
**自包含的 Rust 事件引擎** —— 对 enkeep_trade 平台事件核心(`enkeep_trade/event/engine.py`)的逐语义 Rust 复刻。
> 本 crate **不依赖 enkeep_trade 任何模块**,也不承载任何业务知识(Tick/Order 等)。
> 业务事件常量、Gateway 双投递、OmsEngine、UI 桥接等生态层,由使用方在本库之上自行实现
> —— 与原 Python 平台相同的分层方式;真实使用姿势见 `tests/` 目录(8 个生态使用模式还原测试)。
设计文档:[DESIGN.md](DESIGN.md) · 测试方案与执行记录:[TESTPLAN.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>`
## 快速上手
```toml
[dependencies]
ek-event = "0.1.0"
```
```rust
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`。
## 线程安全矩阵
| `put()` | ✅ 任意线程 | crossbeam MPMC 无锁快路径;put 顺序即队列顺序 |
| `register / unregister`(含 general) | ✅ 任意线程 | 返回全局唯一 `HandlerId`,凭 id 精确注销 |
| `start / stop` | ✅ 任意线程 | 幂等;`stop` 期间并发的 `put` 返回 `Err`,不 panic |
| **分发(消费)** | **单线程串行**(刻意设计) | 同 type 保序、handler 保序;handler 需要并行则自行分流 |
## API 一览
```text
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 版的行为差异
| 定时事件 `"eTimer"` 默认 1s;精确 type 分发;type 后 general;无界队列;start 前 put 合法;stop 丢弃残余;stop 后不可重启 | ✅ 逐项对齐 |
| handler 表 RwLock 保护(原版无锁靠 GIL) | 必要修正 |
| 单 handler panic 不杀分发线程(原版会杀) | 必要修正 |
| HandlerId 注册/注销(原版依赖函数对象相等,Rust 不可复刻) | 必要修正 |
| `put` 在 stop 后显式报 `Err`(原版静默丢弃) | Rust 显式化 |
| `interval: int` 秒 → `Duration` | 更通用 |
| 引擎 `Send + Sync`,可 `Arc` 多线程共享 | 能力增强 |
## 质量状态
- **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)