ek-event 0.1.2

//! P7 —— 阶段哨兵事件模式:
//! 下载成对事件 progress→complete,全部 complete 后发 all-done(对齐 datamaintenance);
//! 无载荷哨兵 + 带载荷哨兵两种形态(对齐 eContractFinish / ePositionDone)。

use ek_event::{Event, EventEngine};
use std::collections::HashSet;
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()
}

// E1: 每类资产 progress×n → complete;凑齐三类后聚合器发 all-complete;
// ack handler 收到时,聚合计数必须已是 3(单分发线程串行保证)
#[test]
fn e1_paired_progress_complete_pipeline() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

    let done = Arc::new(Mutex::new(HashSet::<&'static str>::new()));
    let ack_sent = Arc::new(AtomicUsize::new(0));

    let assets = ["stock", "etf", "index"];
    let d = done.clone();
    let a = ack_sent.clone();
    let engine_for_agg = engine.clone();
    for asset in assets {
        let handler_done = d.clone();
        let handler_ack = a.clone();
        let engine_agg = engine_for_agg.clone();
        engine.register(
            format!("e{asset}DownloadComplete"),
            Arc::new(move |_| {
                handler_done.lock().unwrap().insert(asset);
                if handler_done.lock().unwrap().len() == 3 {
                    engine_agg
                        .put(Event::new("eDataDownloadAllComplete"))
                        .unwrap();
                    handler_ack.fetch_add(1, Ordering::SeqCst);
                }
            }),
        );
    }

    let all_done_seen = Arc::new(AtomicUsize::new(0));
    let ad = all_done_seen.clone();
    let d2 = done.clone();
    engine.register(
        "eDataDownloadAllComplete",
        Arc::new(move |_| {
            // 串行分发保证:此时 done 集必然已满
            assert_eq!(
                d2.lock().unwrap().len(),
                3,
                "all-complete handler must run last"
            );
            ad.fetch_add(1, Ordering::SeqCst);
        }),
    );

    // 每 2 条 progress + 1 条 complete,顺序混着发
    for asset in assets {
        for i in 0..3 {
            engine
                .put(Event::new(format!("e{asset}DownloadProgress{i}")))
                .unwrap();
        }
        engine
            .put(Event::new(format!("e{asset}DownloadComplete")))
            .unwrap();
    }
    assert!(
        wait_for(|| all_done_seen.load(Ordering::SeqCst) == 1),
        "all-complete must fire exactly once (ack={})",
        ack_sent.load(Ordering::SeqCst)
    );
    assert_eq!(
        ack_sent.load(Ordering::SeqCst),
        1,
        "aggregator triggers once"
    );
    engine.stop();
}

// E2: 无载荷哨兵(data=())与带载荷哨兵(data=gateway_name)两种形态
#[test]
fn e2_sentinel_with_and_without_payload() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

    let finish_cnt = Arc::new(AtomicUsize::new(0));
    let names = Arc::new(Mutex::new(Vec::<String>::new()));

    let f = finish_cnt.clone();
    engine.register(
        "eContractFinish",
        Arc::new(move |e| {
            // 无载荷哨兵:Event::new 的 data 恒为 Arc<()>,可精确 downcast
            assert!(
                e.data.downcast_ref::<()>().is_some(),
                "empty sentinel must carry () payload"
            );
            f.fetch_add(1, Ordering::SeqCst);
        }),
    );
    let n = names.clone();
    engine.register(
        "ePositionDone.XT",
        Arc::new(move |e| {
            let gw = e
                .data
                .downcast_ref::<String>()
                .expect("gateway name payload");
            n.lock().unwrap().push(gw.clone());
        }),
    );

    engine.put(Event::new("eContractFinish")).unwrap();
    engine
        .put(Event::with_data(
            "ePositionDone.XT",
            Arc::new("XT".to_string()),
        ))
        .unwrap();

    assert!(wait_for(
        || finish_cnt.load(Ordering::SeqCst) == 1 && names.lock().unwrap().len() == 1
    ));
    assert_eq!(names.lock().unwrap()[0], "XT");
    engine.stop();
}