ek-event 0.1.2

//! P5 —— EVENT_TIMER 定时驱动模式:
//! paperaccount 的计数节流(timer_count 攒 N 次);timer 与业务事件交织不丢。

use ek_event::{EVENT_TIMER, Event, EventEngine};
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..400 {
        if cond() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    cond()
}

// T1: 计数节流 —— 每 N 次 timer 才干一次活(复刻 process_timer_event)
#[test]
fn t1_timer_count_throttle() {
    let engine = Arc::new(EventEngine::with_interval(Duration::from_millis(50)));
    engine.start();

    let fires = Arc::new(AtomicUsize::new(0)); // timer 到达总数(独立计数,不清零)
    let throttle = Arc::new(AtomicUsize::new(0)); // 节流计数(每次干活后归零,复刻 paperaccount timer_count)
    let works = Arc::new(AtomicUsize::new(0));
    let f = fires.clone();
    let th = throttle.clone();
    let w = works.clone();
    engine.register(
        EVENT_TIMER,
        Arc::new(move |_| {
            f.fetch_add(1, Ordering::SeqCst); // 总量独立累计
            let n = th.fetch_add(1, Ordering::SeqCst) + 1;
            if n < 3 {
                return;
            }
            w.fetch_add(1, Ordering::SeqCst);
            th.store(0, Ordering::SeqCst);
        }),
    );

    // 50ms 间隔 × 600ms 窗口 ≈ 11~12 次 timer;慢机器宽容 fired>=7,work≈floor(fired/3)
    std::thread::sleep(Duration::from_millis(600));
    let fired = fires.load(Ordering::SeqCst);
    let worked = works.load(Ordering::SeqCst);
    assert!(fired >= 7, "timer must fire repeatedly (fired {fired})");
    assert!(
        (2..=4).contains(&worked),
        "throttled work must fire roughly every 3rd tick (fired {fired}, worked {worked})"
    );
    engine.stop();
}

// T2: timer 与业务事件交织到达,业务流不丢不乱序
#[test]
fn t2_timer_interleaved_with_business_events() {
    let engine = Arc::new(EventEngine::with_interval(Duration::from_millis(30)));
    engine.start();

    let ticks_seen = Arc::new(AtomicUsize::new(0));
    let seq = Arc::new(Mutex::new(Vec::<usize>::new()));
    let s = seq.clone();
    let seen_for_handler = ticks_seen.clone();
    engine.register(
        "eTick.",
        Arc::new(move |e| {
            let v = *e.data.downcast_ref::<usize>().unwrap();
            s.lock().unwrap().push(v);
            seen_for_handler.fetch_add(1, Ordering::SeqCst);
        }),
    );
    let timer_seen = Arc::new(AtomicUsize::new(0));
    let t = timer_seen.clone();
    engine.register(
        EVENT_TIMER,
        Arc::new(move |_| {
            t.fetch_add(1, Ordering::SeqCst);
        }),
    );

    // 生产 10 个 tick,间隔 30ms,让 timer 事件穿插其中(0usize: 类型要和 downcast 一致)
    for i in 0..10usize {
        engine.put(Event::with_data("eTick.", Arc::new(i))).unwrap();
        std::thread::sleep(Duration::from_millis(30));
    }
    assert!(
        wait_for(|| ticks_seen.load(Ordering::SeqCst) == 10),
        "all 10 business events must be delivered"
    );
    assert!(
        timer_seen.load(Ordering::SeqCst) >= 3,
        "timer events must interleave with business events"
    );
    let got = seq.lock().unwrap().clone();
    let expected: Vec<usize> = (0..10usize).collect();
    assert_eq!(
        got, expected,
        "business order must be preserved despite timer interleave"
    );
    engine.stop();
}