use crate::{
FerroHeadlessPlatform, FerroHeadlessPlatformExtensions, FerroHeadlessPlatformOptions, FerroTestApplicationAttribute,
FerroTestIsolationAttribute, FerroTestIsolationLevel,
};
use ferroui_base::media::{FontManager, MediaContext};
use ferroui_base::reactive::IDisposable;
use ferroui_base::threading::{
CancellationToken, CancellationTokenSource, Dispatcher, DispatcherFrame, FerroSynchronizationContext,
OperationCanceledError,
};
use ferroui_base::{FerroLocator, LocatorExtensions};
use ferroui_controls::{AppBuilder, Application, Control, IToolTipService};
use ferroui_harfbuzz::HarfBuzzApplicationExtensions;
use std::any::Any;
use std::cell::RefCell;
use std::collections::{HashMap, VecDeque};
use std::future::Future;
use std::panic::{catch_unwind, resume_unwind, AssertUnwindSafe};
use std::sync::mpsc;
use std::sync::{Arc, Condvar, Mutex, MutexGuard, OnceLock, PoisonError};
use std::thread::{self, JoinHandle};
use std::time::Duration;
type Exception = Box<dyn Any + Send + 'static>;
#[derive(Debug)]
pub struct FerroTestAssembly {
pub name: &'static str,
pub test_application: Option<FerroTestApplicationAttribute>,
pub test_isolation: Option<FerroTestIsolationAttribute>,
}
enum TaskState<T> {
Running,
RanToCompletion(Option<T>),
Faulted(Option<Exception>),
Canceled,
}
struct TaskShared<T> {
state: Mutex<TaskState<T>>,
finished: Condvar,
}
impl<T> TaskShared<T> {
fn lock(&self) -> MutexGuard<'_, TaskState<T>> {
self.state.lock().unwrap_or_else(PoisonError::into_inner)
}
fn try_set(&self, state: TaskState<T>) -> bool {
let mut current = self.lock();
if !matches!(*current, TaskState::Running) {
return false;
}
*current = state;
self.finished.notify_all();
true
}
}
struct TaskCompletionSource<T>(Arc<TaskShared<T>>);
impl<T> TaskCompletionSource<T> {
fn new() -> (Self, HeadlessUnitTestTask<T>) {
let shared = Arc::new(TaskShared { state: Mutex::new(TaskState::Running), finished: Condvar::new() });
(Self(shared.clone()), HeadlessUnitTestTask { shared })
}
fn try_set_result(&self, result: T) -> bool {
self.0.try_set(TaskState::RanToCompletion(Some(result)))
}
fn try_set_exception(&self, exception: Exception) -> bool {
self.0.try_set(TaskState::Faulted(Some(exception)))
}
fn try_set_canceled(&self) -> bool {
self.0.try_set(TaskState::Canceled)
}
}
impl<T> Drop for TaskCompletionSource<T> {
fn drop(&mut self) {
self.try_set_canceled();
}
}
pub struct HeadlessUnitTestTask<T> {
shared: Arc<TaskShared<T>>,
}
impl<T> HeadlessUnitTestTask<T> {
pub fn is_completed(&self) -> bool {
!matches!(*self.shared.lock(), TaskState::Running)
}
pub fn wait(self) -> Result<T, OperationCanceledError> {
let mut state = self.shared.lock();
while matches!(*state, TaskState::Running) {
state = self.shared.finished.wait(state).unwrap_or_else(PoisonError::into_inner);
}
Self::take(&mut state)
}
pub fn wait_timeout(self, timeout: Duration) -> Result<T, OperationCanceledError> {
let state = self.shared.lock();
let (mut state, _) = self
.shared
.finished
.wait_timeout_while(state, timeout, |state| matches!(*state, TaskState::Running))
.unwrap_or_else(PoisonError::into_inner);
if matches!(*state, TaskState::Running) {
drop(state);
panic!("The operation has timed out.");
}
Self::take(&mut state)
}
fn take(state: &mut TaskState<T>) -> Result<T, OperationCanceledError> {
match state {
TaskState::RanToCompletion(result) => match result.take() {
Some(result) => Ok(result),
None => panic!("The result of the task has already been taken."),
},
TaskState::Faulted(exception) => match exception.take() {
Some(exception) => resume_unwind(exception),
None => panic!("The failure of the task has already been observed."),
},
TaskState::Canceled => Err(OperationCanceledError),
TaskState::Running => unreachable!("the task has finished"),
}
}
}
struct SessionThread {
app_builder: AppBuilder,
shared_context: RefCell<Option<Arc<FerroSynchronizationContext>>>,
}
type WorkItem = Box<dyn FnOnce(&SessionThread) + Send + 'static>;
#[derive(Default)]
struct QueueState {
items: VecDeque<WorkItem>,
adding_completed: bool,
}
#[derive(Default)]
struct BlockingQueue {
state: Mutex<QueueState>,
changed: Condvar,
}
impl BlockingQueue {
fn lock(&self) -> MutexGuard<'_, QueueState> {
self.state.lock().unwrap_or_else(PoisonError::into_inner)
}
fn add(&self, item: WorkItem) {
let mut state = self.lock();
if state.adding_completed {
drop(state);
panic!("The collection has been marked as complete with regards to additions.");
}
state.items.push_back(item);
self.changed.notify_all();
}
fn take(&self, cancellation_token_source: &CancellationTokenSource) -> Option<WorkItem> {
let mut state = self.lock();
loop {
if cancellation_token_source.is_cancellation_requested() {
return None;
}
if let Some(item) = state.items.pop_front() {
return Some(item);
}
state = self.changed.wait(state).unwrap_or_else(PoisonError::into_inner);
}
}
fn wake(&self) {
let _state = self.lock();
self.changed.notify_all();
}
fn complete_adding(&self) {
let dropped = {
let mut state = self.lock();
state.adding_completed = true;
std::mem::take(&mut state.items)
};
drop(dropped);
}
}
pub struct HeadlessUnitTestSession {
cancellation_token_source: CancellationTokenSource,
queue: Arc<BlockingQueue>,
dispatch_task: Mutex<Option<JoinHandle<()>>>,
isolated: bool,
}
fn sessions() -> &'static Mutex<HashMap<&'static str, Arc<HeadlessUnitTestSession>>> {
static SESSIONS: OnceLock<Mutex<HashMap<&'static str, Arc<HeadlessUnitTestSession>>>> = OnceLock::new();
SESSIONS.get_or_init(Default::default)
}
static HEADLESS_ASSEMBLY: FerroTestAssembly =
FerroTestAssembly { name: env!("CARGO_PKG_NAME"), test_application: None, test_isolation: None };
struct CancelOnDrop(CancellationTokenSource);
impl Drop for CancelOnDrop {
fn drop(&mut self) {
self.0.cancel();
}
}
struct ReleaseCompositorOnDrop;
impl Drop for ReleaseCompositorOnDrop {
fn drop(&mut self) {
crate::ferro_headless_platform::FerroHeadlessPlatform::release_compositor();
}
}
impl HeadlessUnitTestSession {
pub fn dispatch(
&self,
action: impl FnOnce() + Send + 'static,
cancellation_token: CancellationToken,
) -> HeadlessUnitTestTask<()> {
self.dispatch_core(
move || {
action();
std::future::ready(())
},
cancellation_token,
)
}
pub fn dispatch_result<TResult: Send + 'static>(
&self,
action: impl FnOnce() -> TResult + Send + 'static,
cancellation_token: CancellationToken,
) -> HeadlessUnitTestTask<TResult> {
self.dispatch_core(move || std::future::ready(action()), cancellation_token)
}
pub fn dispatch_task<F, Fut>(&self, action: F, cancellation_token: CancellationToken) -> HeadlessUnitTestTask<Fut::Output>
where
F: FnOnce() -> Fut + Send + 'static,
Fut: Future + 'static,
Fut::Output: Send + 'static,
{
self.dispatch_core(action, cancellation_token)
}
fn dispatch_core<F, Fut>(&self, action: F, cancellation_token: CancellationToken) -> HeadlessUnitTestTask<Fut::Output>
where
F: FnOnce() -> Fut + Send + 'static,
Fut: Future + 'static,
Fut::Output: Send + 'static,
{
if self.cancellation_token_source.is_cancellation_requested() {
panic!("Cannot access a disposed object.\nObject name: 'Session was already disposed.'.");
}
let token = self.cancellation_token_source.token();
let isolated = self.isolated;
let (tcs, task) = TaskCompletionSource::<Fut::Output>::new();
self.queue.add(Box::new(move |session: &SessionThread| {
let cts = CancellationTokenSource::new();
let global_cts = {
let cts = cts.clone();
token.register(move || cts.cancel())
};
let local_cts = {
let cts = cts.clone();
cancellation_token.register(move || cts.cancel())
};
let application = match catch_unwind(AssertUnwindSafe(|| {
if isolated {
session.ensure_isolated_application()
} else {
session.ensure_shared_application()
}
})) {
Ok(application) => application,
Err(ex) => {
tcs.try_set_exception(ex);
local_cts.dispose();
global_cts.dispose();
return; }
};
let mut should_cancel = false;
let mut caught: Option<Exception> = None;
let mut result: Option<Fut::Output> = None;
match catch_unwind(AssertUnwindSafe(|| {
let future = action();
let guard = CancelOnDrop(cts.clone());
let task = Dispatcher::ui_thread().to_task_scheduler().start_local(async move {
let _guard = guard;
future.await
});
if task.is_completed() {
return match task.result() {
Ok(result) => Some(result),
Err(canceled) => std::panic::panic_any(canceled),
};
}
if cts.is_cancellation_requested() {
should_cancel = true;
None
} else {
let frame = DispatcherFrame::new();
let inner_cts = {
let frame = frame.clone();
cts.token().register(move || frame.set_continue(false))
};
Dispatcher::ui_thread().push_frame(&frame);
inner_cts.dispose();
match task.result() {
Ok(result) => Some(result),
Err(canceled) => std::panic::panic_any(canceled),
}
}
})) {
Ok(value) => result = value,
Err(ex) => caught = Some(ex),
}
if let Err(ex) = catch_unwind(AssertUnwindSafe(|| application.dispose())) {
caught = Some(ex);
}
local_cts.dispose();
global_cts.dispose();
if let Some(caught) = caught {
tcs.try_set_exception(caught);
} else if should_cancel {
tcs.try_set_canceled();
} else if let Some(result) = result {
tcs.try_set_result(result);
}
}));
task
}
pub fn dispose(&self) {
self.cancellation_token_source.cancel();
self.queue.wake();
let dispatch_task = self.dispatch_task.lock().unwrap_or_else(PoisonError::into_inner).take();
if let Some(dispatch_task) = dispatch_task {
if let Err(ex) = dispatch_task.join() {
resume_unwind(ex);
}
}
self.queue.complete_adding();
}
pub async fn dispose_async(&self) {
self.dispose();
}
pub fn start_new(entry_point_type: fn() -> AppBuilder) -> Arc<HeadlessUnitTestSession> {
Self::start_new_with_isolation(entry_point_type, FerroTestIsolationLevel::PerTest)
}
pub fn start_new_with_isolation(
entry_point_type: fn() -> AppBuilder,
isolation_level: FerroTestIsolationLevel,
) -> Arc<HeadlessUnitTestSession> {
let (tcs, started) = mpsc::channel::<Result<(), Exception>>();
let cancellation_token_source = CancellationTokenSource::new();
let queue = Arc::new(BlockingQueue::default());
let task = {
let cancellation_token_source = cancellation_token_source.clone();
let queue = queue.clone();
thread::Builder::new()
.name("HeadlessUnitTestSession".to_owned())
.spawn(move || {
let _dispatcher = Dispatcher::unit_test_scope();
let _compositor = ReleaseCompositorOnDrop;
let session = match catch_unwind(|| {
let mut app_builder = entry_point_type();
if app_builder.windowing_subsystem_name().as_deref() != Some("Headless") {
app_builder = app_builder.use_headless(FerroHeadlessPlatformOptions::default());
}
if app_builder.text_shaping_subsystem_initializer().is_none() {
app_builder = app_builder.use_harfbuzz();
}
SessionThread { app_builder, shared_context: RefCell::new(None) }
}) {
Ok(session) => {
let _ = tcs.send(Ok(()));
session
}
Err(e) => {
let _ = tcs.send(Err(e));
return;
}
};
while !cancellation_token_source.is_cancellation_requested() {
if let Some(action) = queue.take(&cancellation_token_source) {
action(&session);
}
}
})
.expect("the thread of the headless unit test session")
};
match started.recv() {
Ok(Ok(())) => {}
Ok(Err(e)) => resume_unwind(e),
Err(_) => panic!("The thread of the headless unit test session ended before the session was created."),
}
let run_isolated = isolation_level == FerroTestIsolationLevel::PerTest;
Arc::new(HeadlessUnitTestSession {
cancellation_token_source,
queue,
dispatch_task: Mutex::new(Some(task)),
isolated: run_isolated,
})
}
pub fn get_or_start_for_assembly(assembly: Option<&FerroTestAssembly>) -> Arc<HeadlessUnitTestSession> {
let assembly = assembly.unwrap_or(&HEADLESS_ASSEMBLY);
let mut sessions = sessions().lock().unwrap_or_else(PoisonError::into_inner);
if let Some(session) = sessions.get(assembly.name) {
return session.clone();
}
let app_builder_entry_point_type =
assembly.test_application.map(|attribute| attribute.app_builder_entry_point_type());
let isolation_level = assembly
.test_isolation
.map_or(FerroTestIsolationLevel::PerTest, |attribute| attribute.isolation_level());
let session = match app_builder_entry_point_type {
Some(app_builder_entry_point_type) => {
Self::start_new_with_isolation(app_builder_entry_point_type, isolation_level)
}
None => Self::start_new_with_isolation(AppBuilder::configure::<Application>, isolation_level),
};
sessions.insert(assembly.name, session.clone());
session
}
}
impl IDisposable for HeadlessUnitTestSession {
fn dispose(&self) {
HeadlessUnitTestSession::dispose(self)
}
}
fn finish_composition_in_flight() {
let Some(media_context) = FerroLocator::current().get_service::<MediaContext>() else {
return;
};
for _ in 0..10 {
if !media_context.has_composition_batches_in_flight() {
return;
}
FerroHeadlessPlatform::force_render_timer_tick(1);
Dispatcher::ui_thread().run_jobs(None);
}
}
impl SessionThread {
fn ensure_shared_application(&self) -> Box<dyn IDisposable> {
let old_context = FerroSynchronizationContext::current();
if Application::current().is_none() {
self.app_builder.setup_unsafe();
*self.shared_context.borrow_mut() = FerroSynchronizationContext::current();
} else {
FerroSynchronizationContext::set_current(self.shared_context.borrow().clone());
}
Box::new(Cleanup::new(move || {
let jobs = catch_unwind(|| {
Dispatcher::ui_thread().run_jobs(None);
finish_composition_in_flight();
});
FerroSynchronizationContext::set_current(old_context);
if let Err(ex) = jobs {
resume_unwind(ex);
}
}))
}
fn ensure_isolated_application(&self) -> Box<dyn IDisposable> {
let scope = FerroLocator::enter_scope();
let old_context = FerroSynchronizationContext::current();
if let Err(ex) = catch_unwind(AssertUnwindSafe(|| {
Dispatcher::reset_before_unit_tests();
self.app_builder.setup_unsafe();
})) {
scope.dispose();
resume_unwind(ex);
}
Box::new(Cleanup::new(move || {
let services = catch_unwind(|| {
let tool_tip_service = FerroLocator::current().get_service::<dyn IToolTipService>();
if let Some(tool_tip_service) =
tool_tip_service.as_ref().and_then(|service| service.as_tool_tip_service())
{
tool_tip_service.dispose();
}
if let Some(font_manager) = FerroLocator::current().get_service::<FontManager>() {
font_manager.dispose();
}
Dispatcher::reset_for_unit_tests();
});
let scope = catch_unwind(AssertUnwindSafe(|| scope.dispose()));
Dispatcher::reset_before_unit_tests();
FerroSynchronizationContext::set_current(old_context);
Control::reset_loaded_queue_for_unit_tests();
if let Err(ex) = scope {
resume_unwind(ex);
}
if let Err(ex) = services {
resume_unwind(ex);
}
}))
}
}
struct Cleanup(RefCell<Option<Box<dyn FnOnce()>>>);
impl Cleanup {
fn new(action: impl FnOnce() + 'static) -> Self {
Self(RefCell::new(Some(Box::new(action))))
}
}
impl IDisposable for Cleanup {
fn dispose(&self) {
let action = self.0.borrow_mut().take();
if let Some(action) = action {
action();
}
}
}