use ek_event::{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..300 {
if cond() {
return true;
}
std::thread::sleep(Duration::from_millis(10));
}
cond()
}
#[derive(Clone, Debug)]
#[allow(dead_code)] struct TickMsg {
symbol: String,
price: f64,
}
#[test]
fn b1_batch_list_payload() {
let engine = Arc::new(EventEngine::new());
engine.start();
let events_seen = Arc::new(AtomicUsize::new(0));
let rows = Arc::new(Mutex::new(Vec::<(String, usize)>::new()));
let r = rows.clone();
let ev = events_seen.clone();
engine.register(
"eTickStockBatch",
Arc::new(move |e| {
ev.fetch_add(1, Ordering::SeqCst);
let batch = e
.data
.downcast_ref::<Vec<TickMsg>>()
.expect("Vec<TickMsg> payload");
for (i, t) in batch.iter().enumerate() {
r.lock().unwrap().push((t.symbol.clone(), i));
}
}),
);
const N: usize = 500;
let batch: Vec<TickMsg> = (0..N)
.map(|i| TickMsg {
symbol: format!("S{i:04}"),
price: i as f64,
})
.collect();
engine
.put(Event::with_data("eTickStockBatch", Arc::new(batch)))
.unwrap();
assert!(
wait_for(|| events_seen.load(Ordering::SeqCst) == 1),
"batch event delivered"
);
assert!(
wait_for(|| rows.lock().unwrap().len() == N),
"all 500 rows consumed"
);
let got = rows.lock().unwrap().clone();
let idx_ok = got.iter().enumerate().all(|(k, (_, i))| k == *i);
assert!(idx_ok, "batch iteration order must match the list order");
assert_eq!(got[0].0, "S0000");
assert_eq!(got[N - 1].0, "S0499");
engine.stop();
}