ek-event 0.1.2

//! P4 —— rpc_server 的全事件转发器模式:
//! register_general 收一切事件,按 gateway_name 过滤,空载荷跳过(对齐 process_event);
//! 且同一事件先 type handler 后 general handler。

use ek_event::{EVENT_TIMER, Event, EventEngine};
use std::sync::{Arc, Mutex};
use std::time::Duration;

fn wait_for(cond: impl Fn() -> bool) -> bool {
    for _ in 0..300 {
        if cond() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    cond()
}

/// 模拟各 XxxData 的 gateway_name 字段
#[derive(Clone, Debug)]
struct Pub {
    gateway: &'static str,
    kind: &'static str,
}

/// 复刻 RpcServer.process_event:
/// empty 载荷(eTimer 之类)直接 return(对应 Python `if not data`)。
fn rpc_wants(event: &Event) -> bool {
    if event.data.downcast_ref::<()>().is_some() {
        return false; // if not data: return
    }
    match event.data.downcast_ref::<Pub>() {
        Some(p) => p.gateway == "hedge",
        None => false,
    }
}

// R1: 跨 type 保序转发 + gateway 过滤 + 空载荷跳过
#[test]
fn r1_general_forwarder_order_and_filter() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

    let fwd = Arc::new(Mutex::new(Vec::<String>::new()));
    let f = fwd.clone();
    engine.register_general(Arc::new(move |e| {
        if !rpc_wants(e) {
            return;
        }
        let p = e.data.downcast_ref::<Pub>().unwrap();
        f.lock().unwrap().push(p.kind.to_string());
    }));

    let engine = engine.as_ref();
    engine
        .put(Event::with_data(
            "eTick.",
            Arc::new(Pub {
                gateway: "hedge",
                kind: "tick1",
            }),
        ))
        .unwrap();
    engine.put(Event::new(EVENT_TIMER)).unwrap(); // 空载荷 → 应被跳过
    engine
        .put(Event::with_data(
            "eOrder.",
            Arc::new(Pub {
                gateway: "other",
                kind: "order_bad",
            }),
        ))
        .unwrap();
    engine
        .put(Event::with_data(
            "eTrade.",
            Arc::new(Pub {
                gateway: "hedge",
                kind: "trade2",
            }),
        ))
        .unwrap();
    engine
        .put(Event::with_data(
            "eLog",
            Arc::new(Pub {
                gateway: "hedge",
                kind: "log3",
            }),
        ))
        .unwrap();

    assert!(
        wait_for(|| {
            let got = fwd.lock().unwrap();
            *got == ["tick1", "trade2", "log3"]
        }),
        "forwarder must keep put order, filter gateway and skip empty payloads (got {:?})",
        fwd.lock().unwrap()
    );
    engine.stop();
}

// R2: 同一事件,type handler 先执行、general handler 后执行
#[test]
fn r2_type_then_general_on_same_event() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

    let log = Arc::new(Mutex::new(Vec::<&'static str>::new()));
    let l1 = log.clone();
    engine.register(
        "eTrade.",
        Arc::new(move |_| {
            l1.lock().unwrap().push("type");
        }),
    );
    let l2 = log.clone();
    engine.register_general(Arc::new(move |_| {
        l2.lock().unwrap().push("general");
    }));

    engine
        .put(Event::with_data(
            "eTrade.",
            Arc::new(Pub {
                gateway: "hedge",
                kind: "t",
            }),
        ))
        .unwrap();
    assert!(
        wait_for(|| *log.lock().unwrap() == ["type", "general"]),
        "type handler must run before general (got {:?})",
        log.lock().unwrap()
    );
    engine.stop();
}