use crate::core::async_context::LocalExecutor;
use crate::core::entity::EntityMap;
use crate::core::tasks::{Inbox, Waker as AppWaker};
use std::cell::{Cell, RefCell};
use std::future::Future;
use std::pin::Pin;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::task::{Context, Poll, Waker};
use std::time::Instant;
#[derive(Default)]
struct WakeCount(AtomicUsize);
impl AppWaker for WakeCount {
fn wake(&self) {
self.0.fetch_add(1, Ordering::SeqCst);
}
}
#[test]
fn inbox_wakes_once_per_burst_and_rearms_after_drain() {
let wakes = Arc::new(WakeCount::default());
let inbox = Inbox::new(wakes.clone());
let sender = inbox.sender();
for _ in 0..128 {
assert!(sender.send(|_| {}));
}
assert_eq!(wakes.0.load(Ordering::SeqCst), 1);
let mut drained = 0;
while inbox.try_next().is_some() {
drained += 1;
}
assert_eq!(drained, 128);
assert!(sender.send(|_| {}));
assert_eq!(wakes.0.load(Ordering::SeqCst), 2);
}
#[test]
fn replacing_inbox_waker_updates_existing_senders_and_wakes_pending_work() {
let mut inbox = Inbox::default();
let sender = inbox.sender();
sender.send(|_| {});
let wakes = Arc::new(WakeCount::default());
inbox.set_waker(wakes.clone());
assert_eq!(wakes.0.load(Ordering::SeqCst), 1);
while inbox.try_next().is_some() {}
sender.send(|_| {});
assert_eq!(wakes.0.load(Ordering::SeqCst), 2);
}
#[derive(Clone, Default)]
struct Probe {
polls: Rc<Cell<usize>>,
saved: Rc<RefCell<Option<Waker>>>,
complete: Rc<Cell<bool>>,
}
impl Future for Probe {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
self.polls.set(self.polls.get() + 1);
*self.saved.borrow_mut() = Some(cx.waker().clone());
if self.complete.get() { Poll::Ready(()) } else { Poll::Pending }
}
}
#[test]
fn repeated_wakes_poll_a_future_only_once_per_turn() {
let inbox = Inbox::default();
let sender = inbox.sender();
let mut executor = LocalExecutor::default();
let probe = Probe::default();
executor.spawn(probe.clone(), &sender);
executor.poll_ready(&sender);
for _ in 0..128 {
probe.saved.borrow().as_ref().unwrap().wake_by_ref();
}
executor.poll_ready(&sender);
assert_eq!(probe.polls.get(), 2);
assert!(!executor.poll_ready(&sender), "sin wake no se sondea");
}
#[test]
fn stale_waker_cannot_poll_a_reused_future_slot() {
let inbox = Inbox::default();
let sender = inbox.sender();
let mut executor = LocalExecutor::default();
let old = Probe::default();
old.complete.set(true);
executor.spawn(old.clone(), &sender);
executor.poll_ready(&sender);
let new = Probe::default();
executor.spawn(new.clone(), &sender);
executor.poll_ready(&sender);
old.saved.borrow().as_ref().unwrap().wake_by_ref();
assert!(!executor.poll_ready(&sender));
assert_eq!(new.polls.get(), 1);
}
#[test]
fn cancelling_a_sleeping_task_releases_its_future_without_an_external_wake() {
let mut entities = EntityMap::new();
let entity = entities.insert(());
let retained = Rc::new(());
let owned = retained.clone();
let task = entities
.update(entity, |_, cx| {
cx.spawn(|_, _| async move {
let _owned = owned;
std::future::pending::<()>().await;
})
})
.unwrap();
entities.run_pending(Instant::now());
assert_eq!(Rc::strong_count(&retained), 2);
drop(task);
entities.run_pending(Instant::now());
assert_eq!(entities.local.live_futures(), 0);
assert_eq!(Rc::strong_count(&retained), 1);
}
#[test]
fn cancelled_async_timers_do_not_keep_an_idle_deadline() {
let mut entities = EntityMap::new();
let context = entities.async_context();
let task = context.timer(std::time::Duration::from_secs(3600));
entities.run_pending(Instant::now());
assert!(entities.timers.next_deadline().is_some());
drop(task);
assert!(entities.timers.next_deadline().is_none());
entities.run_pending(Instant::now());
let task = context.timer(std::time::Duration::from_secs(3600));
drop(task);
entities.run_pending(Instant::now());
assert!(entities.timers.next_deadline().is_none(), "no registra un timer ya cancelado");
}
#[test]
fn wakes_during_poll_are_preserved_but_coalesced() {
let mut entities = EntityMap::new();
let polls = Rc::new(Cell::new(0));
let count = polls.clone();
entities.spawn_local(std::future::poll_fn(move |cx| {
count.set(count.get() + 1);
if count.get() == 1 {
for _ in 0..128 {
cx.waker().wake_by_ref();
}
Poll::Pending
} else {
Poll::Ready(())
}
}));
entities.run_pending(Instant::now());
assert_eq!(polls.get(), 2);
assert_eq!(entities.local.live_futures(), 0);
assert!(!entities.run_pending(Instant::now()));
}
#[test]
fn concurrent_inbox_producers_cannot_lose_a_wake_while_draining() {
struct Signal(std::sync::mpsc::Sender<()>);
impl AppWaker for Signal {
fn wake(&self) {
self.0.send(()).unwrap();
}
}
let (signal, wakes) = std::sync::mpsc::channel();
let inbox = Inbox::new(Arc::new(Signal(signal)));
let mut entities = EntityMap::new();
let delivered = Arc::new(AtomicUsize::new(0));
std::thread::scope(|scope| {
for _ in 0..4 {
let sender = inbox.sender();
let delivered = delivered.clone();
scope.spawn(move || {
for _ in 0..512 {
let delivered = delivered.clone();
assert!(sender.send(move |_| {
delivered.fetch_add(1, Ordering::SeqCst);
}));
std::thread::yield_now();
}
});
}
while delivered.load(Ordering::SeqCst) < 2048 {
wakes.recv_timeout(std::time::Duration::from_secs(5)).expect("wake perdido");
while let Some(job) = inbox.try_next() {
job(&mut entities);
}
}
});
assert_eq!(delivered.load(Ordering::SeqCst), 2048);
}
#[derive(Default)]
struct Model {
values: Vec<u32>,
latest: crate::core::tasks::Latest,
}
fn pump_for(entities: &mut EntityMap, millis: u64, stop: impl Fn(&EntityMap) -> bool) {
let end = Instant::now() + std::time::Duration::from_millis(millis);
while Instant::now() < end && !stop(entities) {
entities.run_pending(Instant::now());
std::thread::sleep(std::time::Duration::from_millis(2));
}
}
#[test]
fn a_job_returns_its_result_to_the_view_and_redraws() {
let mut entities = EntityMap::new();
let model = entities.insert(Model::default());
entities.update(model, |_, cx| {
cx.job(|_| 21 * 2, |this, value, _| this.values.push(value));
});
pump_for(&mut entities, 5_000, |e| e.read(model).is_some_and(|m| !m.values.is_empty()));
assert_eq!(entities.read(model).unwrap().values, [42]);
assert!(entities.is_dirty(), "el resultado marca la vista para redibujar");
}
#[test]
fn a_cancelled_job_never_reports() {
let mut entities = EntityMap::new();
let model = entities.insert(Model::default());
let job = entities
.update(model, |_, cx| {
cx.job(
|token| {
while !token.is_cancelled() {
std::thread::sleep(std::time::Duration::from_millis(1));
}
1
},
|this, value, _| this.values.push(value),
)
})
.unwrap();
job.cancel();
pump_for(&mut entities, 150, |_| false);
assert!(entities.read(model).unwrap().values.is_empty());
}
#[test]
fn a_blocking_job_runs_on_its_own_thread() {
let mut entities = EntityMap::new();
let model = entities.insert(Model::default());
entities.update(model, |_, cx| {
cx.job_blocking(
"aura-test",
|_| std::thread::current().name().map(str::to_owned),
|this, name, _| {
this.values.push(u32::from(name.as_deref() == Some("aura-test")));
},
);
});
pump_for(&mut entities, 5_000, |e| e.read(model).is_some_and(|m| !m.values.is_empty()));
assert_eq!(entities.read(model).unwrap().values, [1]);
}
#[test]
fn only_the_latest_result_is_applied() {
let mut entities = EntityMap::new();
let model = entities.insert(Model::default());
entities.update(model, |this, cx| {
let slow = |_: &crate::core::tasks::CancelToken| {
std::thread::sleep(std::time::Duration::from_millis(60));
1
};
this.latest.run(cx, slow, |t, v, _| t.values.push(v));
});
entities.update(model, |this, cx| this.latest.run(cx, |_| 2, |t, v, _| t.values.push(v)));
pump_for(&mut entities, 250, |_| false);
assert_eq!(entities.read(model).unwrap().values, [2]);
}