ek-core 0.5.2

//! dynamic_storm(0.2.0 补强):动态生命周期 × 高压风暴的联合 useContext。
//!
//! DS1 双阶段风暴:phase A→ 动态 add c2(车道即时通电,不加漏不重投)→ phase B
//! → 计划内 remove c1 → 断言:总量守恒 / 零溢出 / ePlugDown / 全程零死锁。
//! 判据:静态 c1 吃光全部 5000;动态 c2 恰好吃到 phase B 的 2000(装配完成后
//! forwarder 已在车道 → 之后的每一条都必须送达到位)。

mod common;

use common::{load, wait_until};
use ek_core::{MainEngine, PlugNotice, PluginState};
use ek_event::Event;
use std::sync::Arc;
use std::time::{Duration, Instant};

fn paced_put(bus: &Arc<ek_event::EventEngine>, ty: &str, n: u64, batch: u64) {
    let mut sent = 0u64;
    while sent < n {
        let take = (n - sent).min(batch);
        for i in 0..take {
            bus.put(Event::with_data(ty, Arc::new(sent + i))).unwrap();
        }
        sent += take;
        std::thread::sleep(Duration::from_millis(1));
    }
}

#[test]
fn ds1_two_phase_storm_with_dynamic_members() {
    let eng = MainEngine::new(None);
    let c1 = common::counter();
    let c1b = Arc::clone(&c1);
    eng.add_plugin(
        common::test_plugin("c1")
            .routes_str(&["eTick."])
            .cap(512)
            .mail(move |_e: &Event| {
                common::bump(&c1b);
            })
            .build(),
    )
    .unwrap();
    let downs = common::counter();
    let d2 = Arc::clone(&downs);
    eng.add_plugin(
        common::test_plugin("watch")
            .routes_str(&["ePlugDown"])
            .mail(move |e: &Event| {
                if let Some(n) = e.data.downcast_ref::<PlugNotice>() {
                    if n.name == "c1" {
                        common::bump(&d2);
                    }
                }
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();

    // ── Phase A:3000(static-only 时期)
    let t0 = Instant::now();
    paced_put(eng.event_engine(), "eTick.", 3000, 60);
    let _ = t0.elapsed();

    // ── 动态 add c2(装配同步完成 → 之后一条不漏)
    let c2 = common::counter();
    let c2b = Arc::clone(&c2);
    eng.add_plugin(
        common::test_plugin("c2")
            .routes_str(&["eTick."])
            .cap(512)
            .mail(move |_e: &Event| {
                common::bump(&c2b);
            })
            .build(),
    )
    .unwrap();
    assert_eq!(eng.get_state("c2"), Some(PluginState::Running));

    // ── Phase B:再 2000(所有在线订阅者各吃)
    paced_put(eng.event_engine(), "eTick.", 2000, 60);

    // ── 断言:守恒 + 无溢出 + 无死锁
    assert!(
        wait_until(Duration::from_secs(15), || load(&c1) == 5000),
        "c1 should see all 5000"
    );
    assert!(
        wait_until(Duration::from_secs(10), || load(&c2) == 2000),
        "c2 should see exactly phase-B 2000"
    );
    assert_eq!(eng.mail_stats("c1").map(|s| s.spilled), Some(0));
    assert_eq!(eng.mail_stats("c2").map(|s| s.spilled), Some(0));

    // ── 计划内移除 c1:风暴中途退场 → ePlugDown 广播 + 索引修剪 + 对端零死锁
    eng.remove_plugin("c1").unwrap();
    assert!(wait_until(Duration::from_secs(2), || load(&downs) == 1));
    for i in 0..200u64 {
        let _ = eng
            .event_engine()
            .put(Event::with_data("eTick.", Arc::new(i)));
    }
    // c2 继续工作(不因邻居退场受扰):精确总量 = phase B 2000 + 退场后 200 = 2200
    let ok = wait_until(Duration::from_secs(10), || load(&c2) >= 2200);
    assert!(
        ok,
        "c2 stalled: now={} c1={} c2_spilled={:?} pend={}",
        load(&c2),
        load(&c1),
        eng.mail_stats("c2").map(|s| s.spilled),
        eng.event_engine().pending_events()
    );
    assert!(load(&c2) <= 2200, "c2 must not over-deliver: {}", load(&c2));
    assert_eq!(load(&c1), 5000, "removed plugin frozen");
    assert_eq!(eng.get_state("c2"), Some(PluginState::Running));
    eng.close_all(Duration::from_secs(2));
}