ek-event 0.1.2

//! P1/P9 —— 还原 enkeep_trade Gateway 投递模式:
//! trader/gateway.py on_tick/on_order:聚合类型 + "类型+标识" 双投递;
//! UI/面板注册逐品种 handler;订单级 handler 用后注销。

use ek_event::{Event, EventEngine};
use std::sync::atomic::{AtomicUsize, Ordering};
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()
}

fn drain(engine: &EventEngine) {
    wait_for(|| engine.pending_events() == 0);
}

#[derive(Clone, Debug)]
#[allow(dead_code)] // 测试载荷字段只为形态贴近 TickData,不必全部读出
struct TickMsg {
    symbol: String,
    price: f64,
}

#[derive(Clone, Debug)]
#[allow(dead_code)] // 同上,status 仅用于记录订单阶段语义
struct OrderMsg {
    orderid: String,
    status: &'static str,
}

/// 复刻 BaseGateway.on_tick / on_order:聚合+逐实例双投
struct FakeGateway {
    engine: Arc<EventEngine>,
}

impl FakeGateway {
    fn on_tick(&self, tick: &TickMsg) {
        let engine = self.engine.as_ref();
        engine
            .put(Event::with_data("eTick.", Arc::new(tick.clone())))
            .unwrap();
        engine
            .put(Event::with_data(
                format!("eTick.{}", tick.symbol),
                Arc::new(tick.clone()),
            ))
            .unwrap();
    }
    fn on_order(&self, order: &OrderMsg) {
        let engine = self.engine.as_ref();
        engine
            .put(Event::with_data("eOrder.", Arc::new(order.clone())))
            .unwrap();
        engine
            .put(Event::with_data(
                format!("eOrder.{}", order.orderid),
                Arc::new(order.clone()),
            ))
            .unwrap();
    }
}

// G1: 聚合 handler 收全部 tick,逐品种 handler 只收本品种
#[test]
fn g1_aggregate_and_per_symbol_dispatch() {
    let engine = Arc::new(EventEngine::new());
    engine.start();
    let gw = FakeGateway {
        engine: engine.clone(),
    };

    let agg = Arc::new(Mutex::new(Vec::<String>::new()));
    let panel600 = Arc::new(AtomicUsize::new(0));

    let a = agg.clone();
    engine.register(
        "eTick.",
        Arc::new(move |e| {
            let t = e.data.downcast_ref::<TickMsg>().unwrap();
            a.lock().unwrap().push(t.symbol.clone());
        }),
    );
    let p = panel600.clone();
    engine.register(
        "eTick.600000",
        Arc::new(move |_| {
            p.fetch_add(1, Ordering::SeqCst);
        }),
    );

    for i in 0..4 {
        gw.on_tick(&TickMsg {
            symbol: "600000".into(),
            price: 10.0 + i as f64,
        });
        gw.on_tick(&TickMsg {
            symbol: "300750".into(),
            price: 200.0 + i as f64,
        });
    }

    assert!(
        wait_for(|| agg.lock().unwrap().len() == 8 && panel600.load(Ordering::SeqCst) == 4),
        "aggregate should see all 8 ticks, panel only its 4"
    );
    drain(&engine);
    assert_eq!(
        panel600.load(Ordering::SeqCst),
        4,
        "panel must not see other symbols"
    );
    // 到达序与投递序一致
    let seq = agg.lock().unwrap();
    let mut expected: Vec<&str> = Vec::new();
    for _ in 0..4 {
        expected.push("600000");
        expected.push("300750");
    }
    assert_eq!(*seq, expected);
    engine.stop();
}

// G2: 逐订单 handler,注销后不再收到同 id 订单
#[test]
fn g2_per_order_unregister_stops_delivery() {
    let engine = Arc::new(EventEngine::new());
    engine.start();
    let gw = FakeGateway {
        engine: engine.clone(),
    };

    let cnt = Arc::new(AtomicUsize::new(0));
    let c = cnt.clone();
    let id = engine.register(
        "eOrder.O1",
        Arc::new(move |_| {
            c.fetch_add(1, Ordering::SeqCst);
        }),
    );

    gw.on_order(&OrderMsg {
        orderid: "O1".into(),
        status: "SUBMITTING",
    });
    assert!(
        wait_for(|| cnt.load(Ordering::SeqCst) == 1),
        "per-order handler should fire once"
    );

    assert!(engine.unregister(id), "unregister must succeed");
    gw.on_order(&OrderMsg {
        orderid: "O1".into(),
        status: "ALLTRADED",
    });
    drain(&engine);
    assert_eq!(
        cnt.load(Ordering::SeqCst),
        1,
        "unregistered handler must stay silent"
    );
    engine.stop();
}

// G3: 先 start 后 register(对齐 MainEngine 启动序列):
// 已分发完的历史事件不会回放给新注册者,注册之后的事件必达
#[test]
fn g3_start_before_register_no_replay() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

    let cnt = Arc::new(AtomicUsize::new(0));
    for i in 0..5 {
        engine
            .put(Event::with_data("eEarly.", Arc::new(i)))
            .unwrap();
    }
    drain(&engine);

    let c = cnt.clone();
    engine.register(
        "eEarly.",
        Arc::new(move |_| {
            c.fetch_add(1, Ordering::SeqCst);
        }),
    );
    for i in 0..5 {
        engine
            .put(Event::with_data("eEarly.", Arc::new(i)))
            .unwrap();
    }
    assert!(
        wait_for(|| cnt.load(Ordering::SeqCst) == 5),
        "post-register events must arrive"
    );
    drain(&engine);
    assert_eq!(
        cnt.load(Ordering::SeqCst),
        5,
        "events consumed before registration must not be replayed"
    );
    engine.stop();
}