ek-event 0.1.0

//! P3 —— UI 桥接生命周期:open 注册 / close 注销 / 重开重注册;
//! rpc 服务重启;handler 内部自注销(原版生态常见条件注销写法)。

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

// L1: UI open->close->open,注销后零到达,重注册恢复
#[test]
fn l1_ui_open_close_reopen() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

    let cnt = Arc::new(AtomicUsize::new(0));
    let c = cnt.clone();
    let mut widget_id: Option<HandlerId> = Some(engine.register("eKlineData", Arc::new(move |_| {
        c.fetch_add(1, Ordering::SeqCst);
    })));

    let fire = |engine: &EventEngine, n: usize| {
        for i in 0..n {
            engine.put(Event::with_data("eKlineData", Arc::new(i))).unwrap();
        }
    };

    fire(&engine, 3);
    assert!(wait_for(|| cnt.load(Ordering::SeqCst) == 3), "open: 3 events arrive");
    engine.unregister(widget_id.take().unwrap()); // close
    fire(&engine, 3);
    assert!(wait_for(|| engine.pending_events() == 0), "queued events get drained");
    std::thread::sleep(Duration::from_millis(50)); // 留出分发尾巴
    assert_eq!(cnt.load(Ordering::SeqCst), 3, "closed widget stays silent");

    let c = cnt.clone();
    widget_id = Some(engine.register("eKlineData", Arc::new(move |_| {
        c.fetch_add(1, Ordering::SeqCst);
    }))); // reopen
    fire(&engine, 3);
    assert!(wait_for(|| cnt.load(Ordering::SeqCst) == 6), "reopened widget receives again");
    engine.unregister(widget_id.take().unwrap()); // 离场语义:对齐 kline_widget 的 unregister
    engine.stop();
}

// L2: general 的重启语义(unregister_general / register_general)
#[test]
fn l2_general_restart_semantics() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

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

    for i in 0..3 {
        engine.put(Event::with_data("eTrade.", Arc::new(i))).unwrap();
    }
    assert!(wait_for(|| cnt.load(Ordering::SeqCst) == 3));
    assert!(engine.unregister_general(id), "unregister_general must succeed");

    for i in 0..3 {
        engine.put(Event::with_data("eTrade.", Arc::new(i))).unwrap();
    }
    assert!(
        wait_for(|| engine.pending_events() == 0) && cnt.load(Ordering::SeqCst) == 3,
        "after unregister_general nothing flows"
    );

    let c = cnt.clone();
    engine.register_general(Arc::new(move |_| {
        c.fetch_add(1, Ordering::SeqCst);
    }));
    for i in 0..2 {
        engine.put(Event::with_data("eTrade.", Arc::new(i))).unwrap();
    }
    assert!(wait_for(|| cnt.load(Ordering::SeqCst) == 5), "re-registered general resumes");
    engine.stop();
}

// L3: handler 内部注销自己:不死锁、自身不再收、同 type 伙伴照常、其它 type 照常
#[test]
fn l3_handler_self_unregister() {
    let engine = Arc::new(EventEngine::new());
    engine.start();

    let self_cnt = Arc::new(AtomicUsize::new(0));
    let partner_cnt = Arc::new(AtomicUsize::new(0));

    // 伙伴先注册(处理所有 eSelfKill 事件)
    let p = partner_cnt.clone();
    engine.register("eSelfKill", Arc::new(move |_| {
        p.fetch_add(1, Ordering::SeqCst);
    }));

    // 自杀 handler:首事件后注销自己
    struct KillSwitch {
        engine: Arc<EventEngine>,
        id_slot: Mutex<Option<HandlerId>>,
        fired: AtomicUsize,
    }
    let kill = Arc::new(KillSwitch {
        engine: engine.clone(),
        id_slot: Mutex::new(None),
        fired: AtomicUsize::new(0),
    });
    let k = kill.clone();
    let id = engine.register("eSelfKill", Arc::new(move |_| {
        let ks = &*k;
        ks.fired.fetch_add(1, Ordering::SeqCst);
        if let Some(hid) = ks.id_slot.lock().unwrap().take() {
            ks.engine.unregister(hid);
        }
    }));
    *kill.id_slot.lock().unwrap() = Some(id);

    for i in 0..3 {
        engine.put(Event::with_data("eSelfKill", Arc::new(i))).unwrap();
    }
    assert!(
        wait_for(|| partner_cnt.load(Ordering::SeqCst) == 3),
        "partner must receive all three events"
    );
    assert_eq!(self_cnt.load(Ordering::SeqCst), 0, "unused counter stays 0");
    assert_eq!(
        kill.fired.load(Ordering::SeqCst),
        1,
        "self-killing handler fires exactly once"
    );

    // 其它 type 照常工作
    let other = Arc::new(AtomicUsize::new(0));
    let o = other.clone();
    engine.register("eOther", Arc::new(move |_| {
        o.fetch_add(1, Ordering::SeqCst);
    }));
    engine.put(Event::new("eOther")).unwrap();
    assert!(wait_for(|| other.load(Ordering::SeqCst) == 1), "engine remains healthy");
    engine.stop();
}