ek-event 0.1.2

//! 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();
}