use ferroui_reactive::linq::TimeoutException;
use ferroui_reactiveui::PanicException;
use ferroui_splat::AppBuilder;
use ferroui_system::threading::tasks::{run, Task, TaskCompletionSource};
use ferroui_system::ObservableError;
use std::cell::{Cell, RefCell};
use std::collections::VecDeque;
use std::future::{poll_fn, Future};
use std::pin::Pin;
use std::rc::Rc;
use std::task::Poll;
use std::time::Duration;
const DEFAULT_MAX_WAIT_MS: u64 = 60_000;
#[derive(Default)]
struct TestGate {
held: Cell<bool>,
waiters: RefCell<VecDeque<TaskCompletionSource<()>>>,
}
ferroui_system::ferro_thread_local! {
static TEST_GATE: Rc<TestGate> = Rc::new(TestGate::default());
}
impl TestGate {
fn current() -> Rc<Self> {
TEST_GATE.with(Rc::clone)
}
async fn wait(self: Rc<Self>, max_wait: Duration) -> bool {
if !self.held.get() {
self.held.set(true);
return true;
}
let turn = TaskCompletionSource::<()>::new();
let granted = turn.task();
self.waiters.borrow_mut().push_back(turn);
match first_of(granted.clone(), delay(max_wait)).await {
First::Left(_) => true,
First::Right(_) => {
self.waiters.borrow_mut().retain(|waiter| waiter.task().id() != granted.id());
granted.is_completed()
}
}
}
fn release(&self) {
let next = self.waiters.borrow_mut().pop_front();
match next {
Some(next) => next.set_result(()),
None => self.held.set(false),
}
}
}
enum First<A, B> {
Left(A),
Right(B),
}
async fn first_of<A: Clone + 'static, B: Clone + 'static>(
mut left: Task<A>,
mut right: Task<B>,
) -> First<Result<A, ObservableError>, Result<B, ObservableError>> {
poll_fn(move |context| {
if let Poll::Ready(outcome) = Pin::new(&mut left).poll(context) {
return Poll::Ready(First::Left(outcome));
}
if let Poll::Ready(outcome) = Pin::new(&mut right).poll(context) {
return Poll::Ready(First::Right(outcome));
}
Poll::Pending
})
.await
}
fn delay(duration: Duration) -> Task<()> {
run(move || {
std::thread::sleep(duration);
Ok(())
})
}
pub struct RxTest;
impl RxTest {
pub fn app_builder_test_async(test_body: impl FnOnce() -> Task<()> + 'static) -> Task<()> {
Self::app_builder_test_async_with(test_body, DEFAULT_MAX_WAIT_MS)
}
pub fn app_builder_test_async_with(test_body: impl FnOnce() -> Task<()> + 'static, max_wait_ms: u64) -> Task<()> {
let max_wait = Duration::from_millis(max_wait_ms);
Task::spawn_local(async move {
let gate = TestGate::current();
if !gate.clone().wait(max_wait).await {
return Err(Rc::new(TimeoutException) as ObservableError);
}
AppBuilder::reset_builder_state_for_tests();
let test_task = match PanicException::catch(test_body) {
Ok(task) => task,
Err((error, _payload)) => Task::from_exception(error),
};
let outcome = match first_of(test_task, delay(max_wait)).await {
First::Left(outcome) => outcome,
First::Right(_) => Err(Rc::new(TimeoutException) as ObservableError),
};
AppBuilder::reset_builder_state_for_tests();
gate.release();
outcome
})
}
}