ek-event 0.1.2

//! P2/P6 —— 多组件共享同一引擎 + 并发注册风暴 + 双实例隔离。
//! 对齐:OmsEngine/paperaccount/UI 共享 engine;LockTraderApiEngine 自建第二个 EventEngine。

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

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

#[derive(Clone, Debug)]
#[allow(dead_code)] // 仅计数,不读字段:保留字段以贴近真实 XxxData 形态
enum FeedMsg {
    Tick { symbol: String },
    Order { orderid: String, traded: f64 },
    Trade { tradeid: String },
    Log { msg: String },
}

// S1: gateway 生产 + oms/risk/转发 三组件共享消费,计数守恒
#[test]
fn s1_shared_engine_four_roles() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

    let oms_ticks = Arc::new(AtomicUsize::new(0));
    let oms_orders = Arc::new(AtomicUsize::new(0));
    let risk_orders = Arc::new(AtomicUsize::new(0));
    let general_total = Arc::new(AtomicUsize::new(0));

    let t = oms_ticks.clone();
    engine.register(
        "eTick.",
        Arc::new(move |_| {
            t.fetch_add(1, Ordering::SeqCst);
        }),
    );
    let o = oms_orders.clone();
    engine.register(
        "eOrder.",
        Arc::new(move |_| {
            o.fetch_add(1, Ordering::SeqCst);
        }),
    );
    let r = risk_orders.clone();
    engine.register(
        "eOrder.",
        Arc::new(move |_| {
            r.fetch_add(1, Ordering::SeqCst);
        }),
    );
    let g = general_total.clone();
    engine.register_general(Arc::new(move |_| {
        g.fetch_add(1, Ordering::SeqCst);
    }));

    // 生产:1000 混feed(来自"gateway 线程")
    const N_TICK: usize = 400;
    const N_ORDER: usize = 400;
    const N_TRADE: usize = 100;
    const N_LOG: usize = 100;
    let producer = {
        let e = engine.clone();
        std::thread::spawn(move || {
            for i in 0..N_TICK {
                e.put(Event::with_data(
                    "eTick.",
                    Arc::new(FeedMsg::Tick {
                        symbol: format!("S{i}"),
                    }),
                ))
                .unwrap();
            }
            for i in 0..N_ORDER {
                e.put(Event::with_data(
                    "eOrder.",
                    Arc::new(FeedMsg::Order {
                        orderid: format!("O{i}"),
                        traded: 0.0,
                    }),
                ))
                .unwrap();
            }
            for i in 0..N_TRADE {
                e.put(Event::with_data(
                    "eTrade.",
                    Arc::new(FeedMsg::Trade {
                        tradeid: format!("T{i}"),
                    }),
                ))
                .unwrap();
            }
            for i in 0..N_LOG {
                e.put(Event::with_data(
                    "eLog",
                    Arc::new(FeedMsg::Log {
                        msg: format!("L{i}"),
                    }),
                ))
                .unwrap();
            }
        })
    };
    producer.join().unwrap();

    let total = N_TICK + N_ORDER + N_TRADE + N_LOG;
    assert!(
        wait_for(|| general_total.load(Ordering::SeqCst) == total
            && oms_ticks.load(Ordering::SeqCst) == N_TICK
            && oms_orders.load(Ordering::SeqCst) == N_ORDER
            && risk_orders.load(Ordering::SeqCst) == N_ORDER),
        "all roles must receive exactly what was published"
    );
    engine.stop();
}

// S2: 注册/注销风暴 + 并发洪峰,引擎必须存活且计数守恒(仅 general 守恒可精确断言)
#[test]
fn s2_register_unregister_storm() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

    let general_total = Arc::new(AtomicUsize::new(0));
    let g = general_total.clone();
    engine.register_general(Arc::new(move |_| {
        g.fetch_add(1, Ordering::SeqCst);
    }));

    const TOTAL: usize = 20_000;
    let producer = {
        let e = engine.clone();
        std::thread::spawn(move || {
            for i in 0..TOTAL {
                let ty = format!("eStorm{}", i % 4);
                e.put(Event::with_data(ty, Arc::new(i))).unwrap();
            }
        })
    };

    // 4 个风暴线程:register -> sleep -> unregister 循环
    let mut storm = Vec::new();
    for k in 0..4 {
        let e = engine.clone();
        storm.push(std::thread::spawn(move || {
            for round in 0..50 {
                let cnt = Arc::new(AtomicUsize::new(0));
                let c = cnt.clone();
                let ty = format!("eStorm{k}");
                let id = e.register(
                    &ty,
                    Arc::new(move |_| {
                        c.fetch_add(1, Ordering::SeqCst);
                    }),
                );
                std::thread::sleep(Duration::from_millis(1 + round % 3));
                e.unregister(id);
                e.unregister(id); // 幂等
            }
        }));
    }
    producer.join().unwrap();
    for s in storm {
        s.join().unwrap();
    }
    assert!(
        wait_for(|| general_total.load(Ordering::SeqCst) == TOTAL),
        "general handler must see every event despite the storm"
    );
    engine.stop();
}

// S3: 双实例并存互不串扰 + 各自 stop 互不影响(对齐 LockTraderApiEngine 自建 ee)
// 载荷用引擎编号(0=A,1=B),handler 校验"自家计数 + 跨引擎泄漏为 0"
#[test]
fn s3_two_engines_isolated() {
    let ea = Arc::new(EventEngine::new());
    let eb = Arc::new(EventEngine::new());
    ea.start();
    eb.start();

    let own_a = Arc::new(AtomicUsize::new(0));
    let leaked_a = Arc::new(AtomicUsize::new(0));
    let own_b = Arc::new(AtomicUsize::new(0));
    let leaked_b = Arc::new(AtomicUsize::new(0));

    let (oa, la) = (own_a.clone(), leaked_a.clone());
    ea.register(
        "eTick.",
        Arc::new(move |e| {
            if *e.data.downcast_ref::<usize>().unwrap() == 0 {
                oa.fetch_add(1, Ordering::SeqCst);
            } else {
                la.fetch_add(1, Ordering::SeqCst);
            }
        }),
    );
    let (ob, lb) = (own_b.clone(), leaked_b.clone());
    eb.register(
        "eTick.",
        Arc::new(move |e| {
            if *e.data.downcast_ref::<usize>().unwrap() == 1 {
                ob.fetch_add(1, Ordering::SeqCst);
            } else {
                lb.fetch_add(1, Ordering::SeqCst);
            }
        }),
    );

    for i in 0..5 {
        // 聚合 type 精确为 "eTick.";载荷标记来源引擎,用于探测串扰
        ea.put(Event::with_data("eTick.", Arc::new(0usize)))
            .unwrap();
        eb.put(Event::with_data("eTick.", Arc::new(1usize)))
            .unwrap();
        let _ = i;
    }
    assert!(
        wait_for(|| own_a.load(Ordering::SeqCst) == 5 && own_b.load(Ordering::SeqCst) == 5),
        "each engine must receive its own 5 events"
    );
    assert_eq!(
        leaked_a.load(Ordering::SeqCst),
        0,
        "no cross-engine leakage into A"
    );
    assert_eq!(
        leaked_b.load(Ordering::SeqCst),
        0,
        "no cross-engine leakage into B"
    );

    eb.stop(); // 停 B 不影响 A
    for _ in 0..3 {
        assert!(
            ea.put(Event::with_data("eTick.", Arc::new(0usize))).is_ok(),
            "A must survive B's stop"
        );
    }
    assert!(wait_for(|| own_a.load(Ordering::SeqCst) == 8));
    assert!(
        eb.put(Event::new("eTick.")).is_err(),
        "stopped engine must reject put"
    );
    assert_eq!(own_b.load(Ordering::SeqCst), 5, "stopped B stays silent");
    ea.stop();
}