some_executor 0.7.2

// SPDX-License-Identifier: MIT OR Apache-2.0

//! The channel between a spawned task and whoever spawned it.
//!
//! Spawning splits into two halves that share one allocation. The executor keeps
//! an `ObserverSender` and pushes the task's output into it once; the spawner
//! keeps a [`TypedObserver`], which reports progress through [`Observation`] and
//! resolves to a [`FinishedObservation`] when awaited.
//!
//! The halves also carry the cancellation protocol, in both directions:
//!
//! - Dropping the observer without [`Observer::detach`] asks for the task to be
//!   cancelled. That sets the task's [`IS_CANCELLED`](crate::task::IS_CANCELLED)
//!   token, and notifies the executor through [`ExecutorNotified`] so it can stop
//!   scheduling the task at all.
//! - Dropping the sender before the task produced a value means the task was
//!   cancelled, and the observer resolves to [`FinishedObservation::Cancelled`].
//!
//! A value only ever crosses once. [`Observation::Ready`] hands it to the first
//! caller that asks, and every later observation reports [`Observation::Done`].
//!
//! [`ObserverNotified`] is the push counterpart to polling an observer: the
//! executor calls it inline on completion, and drops it without calling it when
//! the task is cancelled.

use crate::task::{InFlightTaskCancellation, TaskID};
use atomic_waker::AtomicWaker;
use std::any::Any;
use std::convert::Infallible;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::task::{Context, Poll};
use wasm_lite_std::Mutex;

/// Represents the state of an observed task.
///
/// This enum is returned by [`Observer::observe`] to indicate the current status
/// of a task being observed. Once a task completes with [`Observation::Ready`],
/// subsequent observations will return [`Observation::Done`].
///
/// # Examples
///
/// ```
/// use some_executor::observer::Observation;
///
/// // An observation can be constructed from a value
/// let obs: Observation<i32> = Observation::from(42);
/// assert_eq!(obs, Observation::Ready(42));
///
/// // Observations can be converted to Option
/// let ready: Observation<String> = Observation::Ready("hello".to_string());
/// let opt: Option<String> = ready.into();
/// assert_eq!(opt, Some("hello".to_string()));
///
/// let pending: Observation<String> = Observation::Pending;
/// let opt: Option<String> = pending.into();
/// assert_eq!(opt, None);
/// ```
///
/// # Exhaustiveness
///
/// This enum is deliberately not `#[non_exhaustive]`. It describes a closed
/// lifecycle — the task is running, it has a value, its value has already been
/// taken, or it was cancelled — and every one of those calls for a different
/// decision by the caller. Forcing a `_` arm would let a state that deserves
/// handling be swallowed silently; if a variant is ever added, a compile error
/// at each match is the outcome worth having. Contrast [`Hint`](crate::hint::Hint),
/// which is an open-ended advisory set and is `#[non_exhaustive]` for that reason.
#[derive(Debug, PartialEq, Eq, Hash, Clone)]
pub enum Observation<T> {
    /// The task is still running and has not completed.
    Pending,
    /// The task has completed with a value.
    ///
    /// This value can only be observed once. After the value is taken,
    /// subsequent observations will return [`Observation::Done`].
    Ready(T),
    /// The task was previously completed and its value has already been observed.
    ///
    /// This state indicates that [`Observation::Ready`] was previously returned
    /// and the value has been consumed.
    Done,
    /// The task was cancelled before completion.
    ///
    /// Cancellation can occur when an [`Observer`] is dropped without calling
    /// [`Observer::detach`].
    Cancelled,
}

/// Represents the final state of an observed task.
///
/// This enum is returned when awaiting an [`Observer`] as a [`Future`].
/// Unlike [`Observation`], this only includes terminal states - either
/// the task completed successfully or was cancelled.
///
/// # Examples
///
/// ```
/// # use some_executor::observer::FinishedObservation;
/// # async fn example() {
/// # let observer: some_executor::observer::TypedObserver<String, std::convert::Infallible> = todo!();
/// // When awaiting an observer, you get a FinishedObservation
/// match observer.await {
///     FinishedObservation::Ready(value) => {
///         println!("Task completed with: {}", value);
///     }
///     FinishedObservation::Cancelled => {
///         println!("Task was cancelled");
///     }
/// }
/// # }
/// ```
///
/// # Exhaustiveness
///
/// Deliberately not `#[non_exhaustive]`, for the reason given on
/// [`Observation`]: these are the two terminal outcomes of a task, and a caller
/// that does not distinguish them is almost certainly wrong.
#[derive(Debug, PartialEq, Eq, Hash, Clone)]
pub enum FinishedObservation<T> {
    /// The task completed successfully with a value.
    Ready(T),
    /// The task was cancelled before completion.
    Cancelled,
}

#[derive(Debug)]
struct Shared<T> {
    //for now, we implement this with a mutex
    lock: Mutex<Observation<T>>,
    /**
    To be notified when the task is cancelled.
    */
    waker: AtomicWaker,
    /**
    Indicates the observer was dropped without detach.
    */
    observer_cancelled: AtomicBool,

    in_flight_task_cancellation: InFlightTaskCancellation,
}

/// Provides observation and control over spawned tasks.
///
/// An `Observer` allows you to:
/// - Check the current state of a task without blocking using [`observe`](Self::observe)
/// - Wait for task completion by awaiting the observer (it implements [`Future`])
/// - Cancel a task by dropping the observer
/// - Detach from a task to let it run independently using [`detach`](Self::detach)
///
/// # Cancellation
///
/// Dropping an observer requests cancellation of the associated task. Cancellation in
/// some_executor is optimistic and operates at three levels:
///
/// 1. **Lightweight cancellation**: some_executor guarantees that polls occurring logically
///    after cancellation will not be run. This is free and universal.
/// 2. **In-flight cancellation**: If the task is currently being polled, cancellation depends
///    on how the task itself reacts to cancellation through the `IS_CANCELLED` task local.
///    Task support for this is sporadic and not guaranteed.
/// 3. **Executor cancellation**: The executor may support cancellation by dropping the future
///    and not running it again. This is not guaranteed and depends on the executor implementation.
///
/// To allow a task to continue running without the observer, use [`detach`](Self::detach).
///
/// # Examples
///
/// ```
/// # use some_executor::observer::{Observer, TypedObserver};
/// # use some_executor::observer::{Observation, FinishedObservation};
/// # async fn example() {
/// # let observer: TypedObserver<String, std::convert::Infallible> = todo!();
/// // Check task state without blocking
/// match observer.observe() {
///     Observation::Pending => println!("Task still running"),
///     Observation::Ready(value) => println!("Task completed: {}", value),
///     Observation::Done => println!("Value already taken"),
///     Observation::Cancelled => println!("Task was cancelled"),
/// }
///
/// // Or wait for completion
/// # let observer: TypedObserver<String, std::convert::Infallible> = todo!();
/// match observer.await {
///     FinishedObservation::Ready(value) => println!("Got: {}", value),
///     FinishedObservation::Cancelled => println!("Cancelled"),
/// }
///
/// // Detach to let task run independently
/// # let observer: TypedObserver<String, std::convert::Infallible> = todo!();
/// observer.detach(); // Task continues running
/// # }
/// ```
#[must_use]
pub trait Observer: 'static + Future<Output = FinishedObservation<Self::Value>> {
    /// The type of value produced by the observed task.
    type Value;

    /// Checks the current state of the task without blocking.
    ///
    /// This method allows you to inspect the task's progress without waiting.
    /// Note that [`Observation::Ready`] can only be returned once - subsequent
    /// calls will return [`Observation::Done`] after the value has been taken.
    fn observe(&self) -> Observation<Self::Value>;

    /// Returns the unique identifier of the observed task.
    fn task_id(&self) -> &TaskID;

    /// Marks this observer as detached in place, so that dropping it will no longer
    /// cancel the task.
    ///
    /// This is the object-safe primitive underlying [`detach`](Self::detach); most
    /// code should call that method instead.
    fn detach_in_place(&mut self);

    /// Detaches from the task, allowing it to continue running independently.
    ///
    /// After calling this method, the observer is consumed and the task will
    /// continue to execute rather than being cancelled the way a plain drop
    /// would cancel it.
    fn detach(mut self)
    where
        Self: Sized,
    {
        self.detach_in_place();
    }
}

/// A concrete implementation of [`Observer`] for tasks that return a specific type.
///
/// `TypedObserver` is the primary way to observe tasks spawned on executors. It provides
/// both synchronous observation through [`observe`](Self::observe) and asynchronous
/// completion through its [`Future`] implementation.
///
/// The `ENotifier` type parameter allows executors to receive notifications about
/// observer lifecycle events (like cancellation requests). Most users can use
/// [`std::convert::Infallible`] for this parameter if executor notifications aren't needed.
///
/// # Examples
///
/// ```
/// # use some_executor::observer::TypedObserver;
/// # use std::convert::Infallible;
/// # async fn example() {
/// # let observer: TypedObserver<i32, Infallible> = todo!();
/// // Spawn a task and get an observer
/// // let observer = executor.spawn(task);
///
/// // Check status without blocking
/// use some_executor::observer::Observation;
/// if let Observation::Ready(value) = observer.observe() {
///     println!("Task completed with: {}", value);
/// }
///
/// // Or wait for completion
/// # let observer: TypedObserver<i32, Infallible> = todo!();
/// let result = observer.await;
/// println!("Final result: {:?}", result);
/// # }
/// ```
///
/// # Cancellation
///
/// Dropping a `TypedObserver` will request cancellation of the associated task:
///
/// ```
/// # use some_executor::SomeExecutor;
/// # use some_executor::current_executor::current_executor;
/// # use some_executor::observer::Observer;
/// # use some_executor::task::{Configuration, Task};
/// # let mut executor = current_executor();
/// # let observer = executor.spawn(Task::without_notifications(
/// #     "cancelled".to_string(),
/// #     Configuration::default(),
/// #     async {},
/// # ));
/// // This will cancel the task
/// drop(observer);
///
/// // To avoid cancellation, detach the observer
/// # let observer = executor.spawn(Task::without_notifications(
/// #     "detached".to_string(),
/// #     Configuration::default(),
/// #     async {},
/// # ));
/// observer.detach(); // Task continues running
/// ```
#[derive(Debug)]
pub struct TypedObserver<T, ENotifier: ExecutorNotified> {
    shared: Arc<Shared<T>>,
    task_id: TaskID,
    notifier: Option<ENotifier>,
    detached: bool,
}

impl<T, ENotifier: ExecutorNotified> Drop for TypedObserver<T, ENotifier> {
    fn drop(&mut self) {
        if !self.detached {
            let should_cancel = {
                let state = self.shared.lock.lock_sync();
                if matches!(*state, Observation::Pending) {
                    self.shared
                        .observer_cancelled
                        .store(true, std::sync::atomic::Ordering::Relaxed);
                    self.shared.in_flight_task_cancellation.cancel();
                    true
                } else {
                    false
                }
            };
            if should_cancel && let Some(mut n) = self.notifier.take() {
                n.request_cancel()
            }
        }
    }
}

impl<T: 'static, ENotifier: ExecutorNotified> Observer for TypedObserver<T, ENotifier> {
    type Value = T;

    fn observe(&self) -> Observation<Self::Value> {
        TypedObserver::observe(self)
    }

    fn task_id(&self) -> &TaskID {
        TypedObserver::task_id(self)
    }

    fn detach_in_place(&mut self) {
        self.notifier.take();
        self.detached = true;
    }
}

/**
The sender side of an observer.  This side is held by the executor, and is used to send values to the observer.
*/
#[derive(Debug)]
pub(crate) struct ObserverSender<T, Notifier> {
    shared: Arc<Shared<T>>,
    pub(crate) notifier: Option<Notifier>,
}

impl<T, Notifier> ObserverSender<T, Notifier> {
    pub(crate) fn send(&mut self, value: T)
    where
        Notifier: ObserverNotified<T>,
    {
        if let Some(n) = self.notifier.as_mut() {
            n.notify(&value)
        }
        {
            let mut lock = self.shared.lock.lock_sync();
            match *lock {
                Observation::Pending => {
                    *lock = Observation::Ready(value);
                }
                Observation::Ready(_) => {
                    panic!("Observer already has a value");
                }
                Observation::Done => {
                    panic!("Observer already completed");
                }
                Observation::Cancelled => {
                    panic!("Observer cancelled");
                }
            }
        }
        self.shared.waker.wake();
    }

    pub(crate) fn observer_cancelled(&self) -> bool {
        self.shared
            .observer_cancelled
            .load(std::sync::atomic::Ordering::Relaxed)
    }
}

impl<T, Notifier> Drop for ObserverSender<T, Notifier> {
    fn drop(&mut self) {
        let should_wake = {
            let mut lock = self.shared.lock.lock_sync();
            match *lock {
                Observation::Pending => {
                    //The task is being dropped before it produced a value, so it really
                    //was cancelled. Only mark the token in this case: a task that ran to
                    //completion was not cancelled, and clones of the token handed out
                    //through IS_CANCELLED would otherwise report cancellation on success.
                    self.shared.in_flight_task_cancellation.cancel();
                    *lock = Observation::Cancelled;
                    true
                }
                Observation::Ready(_) => {
                    //nothing to do
                    false
                }
                Observation::Done => {
                    //nothing to do
                    false
                }
                Observation::Cancelled => {
                    panic!("Observer cancelled");
                }
            }
        };
        if should_wake {
            self.shared.waker.wake();
        }
    }
}

impl<T, E: ExecutorNotified> TypedObserver<T, E> {
    /// Checks the current state of the observed task.
    ///
    /// This method provides a non-blocking way to check if a task has completed.
    /// The first call after task completion will return [`Observation::Ready`] with
    /// the value, and subsequent calls will return [`Observation::Done`].
    ///
    /// # Examples
    ///
    /// ```
    /// # use some_executor::SomeExecutor;
    /// # use some_executor::current_executor::current_executor;
    /// # use some_executor::observer::{Observation, Observer};
    /// # use some_executor::task::{Configuration, Task};
    /// # let mut executor = current_executor();
    /// # let observer = executor.spawn(Task::without_notifications(
    /// #     "greet".to_string(),
    /// #     Configuration::default(),
    /// #     async { "hello".to_string() },
    /// # ));
    /// match observer.observe() {
    ///     Observation::Pending => println!("Still working..."),
    ///     Observation::Ready(result) => println!("Got result: {}", result),
    ///     Observation::Done => println!("Already retrieved the result"),
    ///     Observation::Cancelled => println!("Task was cancelled"),
    /// }
    /// ```
    pub fn observe(&self) -> Observation<T> {
        let mut lock = self.shared.lock.lock_sync();
        match *lock {
            Observation::Pending => Observation::Pending,
            Observation::Ready(..) => std::mem::replace(&mut *lock, Observation::Done),
            Observation::Done => Observation::Done,
            Observation::Cancelled => Observation::Cancelled,
        }
    }

    /// Returns the unique identifier of the task being observed.
    ///
    /// Each spawned task has a unique [`TaskID`] that can be used for logging,
    /// debugging, or tracking purposes.
    pub fn task_id(&self) -> &TaskID {
        &self.task_id
    }

    /// Detaches from the task, allowing it to continue running independently.
    ///
    /// After calling this method, the task will continue to execute even after
    /// this observer is dropped. This is useful for "fire-and-forget" operations
    /// where you don't need the task's result and don't want to cancel it.
    ///
    /// # Examples
    ///
    /// ```
    /// # use some_executor::SomeExecutor;
    /// # use some_executor::current_executor::current_executor;
    /// # use some_executor::observer::Observer;
    /// # use some_executor::task::{Configuration, Task};
    /// # let mut executor = current_executor();
    /// # let observer = executor.spawn(Task::without_notifications(
    /// #     "background".to_string(),
    /// #     Configuration::default(),
    /// #     async {},
    /// # ));
    /// // Start a background task
    /// observer.detach();
    /// // The task continues running even though we no longer hold the observer
    /// ```
    pub fn detach(mut self) {
        self.notifier.take();
        self.detached = true;
    }
}

/// # Panics
///
/// [`Observation::Ready`] can only be produced once. Polling (or awaiting) the
/// observer after the value has already been taken — either by a previous
/// [`observe`](TypedObserver::observe) call that returned `Ready`, or by the
/// future itself having completed — panics.
impl<T, E> Future for TypedObserver<T, E>
where
    E: ExecutorNotified,
{
    type Output = FinishedObservation<T>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.shared.waker.register(cx.waker());
        let o = self.observe();
        match o {
            Observation::Pending => Poll::Pending,
            Observation::Ready(v) => Poll::Ready(FinishedObservation::Ready(v)),
            Observation::Done => {
                panic!(
                    "Observer polled after the value was already taken (e.g. by a prior observe() or await)"
                );
            }
            Observation::Cancelled => Poll::Ready(FinishedObservation::Cancelled),
        }
    }
}

/// Provides inline notifications when an observed task completes.
///
/// Unlike [`Observer`] which requires polling or awaiting, `ObserverNotified` allows
/// you to receive immediate notification when a task completes. The notifier's
/// [`notify`](Self::notify) method is called inline by the executor when the task
/// finishes.
///
/// # Cancellation Behavior
///
/// When a task is cancelled, the notifier is dropped without calling [`notify`](Self::notify).
/// This allows you to detect cancellation by implementing [`Drop`] for your notifier type.
///
/// # Design Note
///
/// This trait requires [`Unpin`] to allow the notifier to be moved during task execution.
/// This design choice enables notifiers to maintain mutable state without requiring
/// interior mutability patterns.
///
/// # Examples
///
/// ```
/// use some_executor::observer::ObserverNotified;
/// use std::sync::mpsc;
///
/// // A simple notifier that sends results through a channel
/// struct ChannelNotifier<T> {
///     sender: mpsc::Sender<T>,
/// }
///
/// impl<T: Clone + 'static> ObserverNotified<T> for ChannelNotifier<T> {
///     fn notify(&mut self, value: &T) {
///         let _ = self.sender.send(value.clone());
///     }
/// }
///
/// // A notifier that logs completion
/// struct LogNotifier {
///     task_name: String,
/// }
///
/// impl<T: std::fmt::Debug + ?Sized> ObserverNotified<T> for LogNotifier {
///     fn notify(&mut self, value: &T) {
///         println!("Task '{}' completed with: {:?}", self.task_name, value);
///     }
/// }
///
/// impl Drop for LogNotifier {
///     fn drop(&mut self) {
///         println!("Task '{}' was cancelled or notifier dropped", self.task_name);
///     }
/// }
/// ```
pub trait ObserverNotified<T: ?Sized>: Unpin + 'static {
    /// Called inline when the observed task completes successfully.
    ///
    /// This method is invoked by the executor immediately when the task finishes.
    /// It will not be called if the task is cancelled.
    ///
    /// # Parameters
    /// - `value`: A reference to the task's output value
    fn notify(&mut self, value: &T);
}

/// Allows executors to receive notifications about observer lifecycle events.
///
/// This trait enables executors to be notified when observers request task cancellation.
/// While implementing cancellation support is optional, it can improve efficiency by
/// allowing executors to stop polling cancelled tasks.
///
/// # Using Without Notifications
///
/// If your executor doesn't need notifications, use [`std::convert::Infallible`] as
/// the type parameter and pass `None` where executor notifiers are expected:
///
/// ```
/// use std::convert::Infallible;
/// use some_executor::observer::{Observer, ObserverNotified, TypedObserver};
/// use some_executor::{BoxedSendObserver, BoxedSendObserverFuture, DynExecutor, ObjSafeTask, SomeExecutor};
/// use some_executor::task::Task;
///
///
/// // Define an executor that doesn't use notifications
/// #[derive(Debug)]
/// struct MyExecutor;
///
/// impl SomeExecutor for MyExecutor {
///     type ExecutorNotifier = Infallible;
///
/// fn spawn<F: Future + Send + 'static, Notifier: ObserverNotified<F::Output> + Send>(&mut self, task: Task<F, Notifier>) -> impl Observer<Value=F::Output> + Send where Self: Sized, F::Output: Send + Unpin {
///          todo!() as TypedObserver::<F::Output, Infallible>
///     }
///
/// fn spawn_async<'s, F: Future + Send + 'static, Notifier: ObserverNotified<F::Output> + Send>(&'s mut self, task: Task<F, Notifier>) -> impl Future<Output=impl Observer<Value=F::Output>> + Send + 's where Self: Sized, F::Output: Send + Unpin {
///         async { todo!() as TypedObserver::<F::Output, Infallible>}
///     }
///
/// fn spawn_objsafe(&mut self, task: ObjSafeTask) -> BoxedSendObserver {
///         todo!()
///     }
///
/// fn spawn_objsafe_async<'s>(&'s mut self, task: ObjSafeTask) -> BoxedSendObserverFuture<'s> {
///         Box::new(async { todo!() })
///     }
///
/// fn clone_box(&self) -> Box<DynExecutor> {
///         todo!()
///     }
///
/// fn executor_notifier(&mut self) -> Option<Self::ExecutorNotifier> {
///         None
///     }
///     
///     // Implement the required methods...
///     // (implementation details omitted for brevity)
/// }
/// ```
///
/// # Examples
///
/// ```
/// use some_executor::observer::ExecutorNotified;
///
/// // A simple notifier that tracks cancellation requests
/// struct CancellationTracker {
///     task_id: u64,
///     cancelled: bool,
/// }
///
/// impl ExecutorNotified for CancellationTracker {
///     fn request_cancel(&mut self) {
///         self.cancelled = true;
///         println!("Task {} cancellation requested", self.task_id);
///         // Executor can now stop polling this task
///     }
/// }
/// ```
pub trait ExecutorNotified: 'static {
    /// Called when an observer requests cancellation of its associated task.
    ///
    /// Executors can use this notification to stop polling the task, though
    /// implementing this is optional. The cancellation is already tracked through
    /// other mechanisms, so this is purely an optimization opportunity.
    fn request_cancel(&mut self);
}

impl<T> ObserverNotified<T> for Infallible {
    fn notify(&mut self, _value: &T) {
        panic!("NoNotified should not be used");
    }
}
//support unboxing
impl ObserverNotified<Box<dyn std::any::Any + 'static>>
    for Box<dyn ObserverNotified<dyn std::any::Any + 'static>>
{
    fn notify(&mut self, value: &Box<dyn Any + 'static>) {
        let r = Box::as_mut(self);
        //`&**value` unboxes: passing `value` directly would unsize-coerce the *box*
        //into `&dyn Any`, and the erased notifier would then fail to downcast.
        r.notify(&**value);
    }
}

//I guess a few ways?

impl ObserverNotified<Box<dyn Any + Send + 'static>>
    for Box<dyn ObserverNotified<dyn Any + Send + 'static> + Send>
{
    fn notify(&mut self, value: &Box<dyn Any + Send + 'static>) {
        let r = Box::as_mut(self);
        //`&**value` unboxes: passing `value` directly would unsize-coerce the *box*
        //into `&dyn Any`, and the erased notifier would then fail to downcast.
        r.notify(&**value);
    }
}

impl ExecutorNotified for Infallible {
    fn request_cancel(&mut self) {
        panic!("NoNotified should not be used");
    }
}

pub(crate) fn observer_channel<R, ONotifier, ENotifier: ExecutorNotified>(
    observer_notify: Option<ONotifier>,
    executor_notify: Option<ENotifier>,
    task_cancellation: InFlightTaskCancellation,
    task_id: TaskID,
) -> (ObserverSender<R, ONotifier>, TypedObserver<R, ENotifier>) {
    let shared = Arc::new(Shared {
        lock: Mutex::new(Observation::Pending),
        waker: AtomicWaker::new(),
        observer_cancelled: AtomicBool::new(false),
        in_flight_task_cancellation: task_cancellation,
    });
    (
        ObserverSender {
            shared: shared.clone(),
            notifier: observer_notify,
        },
        TypedObserver {
            shared,
            task_id,
            notifier: executor_notify,
            detached: false,
        },
    )
}

/**
Allow a `Box<dyn ExecutorNotified>` to be used as an ExecutorNotified directly.

The implementation proceeds by dyanmic dispatch.
*/
impl ExecutorNotified for Box<dyn ExecutorNotified + '_> {
    fn request_cancel(&mut self) {
        (**self).request_cancel();
    }
}

/*
I don't really get why we need both of these... but we do!
 */
impl ExecutorNotified for Box<dyn ExecutorNotified + Send> {
    fn request_cancel(&mut self) {
        (**self).request_cancel();
    }
}

/*
boilerplates

Observer - avoid copy/clone, Eq, Hash, default (channel), from/into, asref/asmut, deref, etc.

Observation - we want clone, Eq, Hash.
default is not obvious to me – could be pending but idk
we could support from based on the value
 */
impl<T> From<T> for Observation<T> {
    fn from(value: T) -> Self {
        Observation::Ready(value)
    }
}

impl<T> From<Observation<T>> for Option<T> {
    fn from(value: Observation<T>) -> Self {
        match value {
            Observation::Ready(v) => Some(v),
            _ => None,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{ExecutorNotified, TypedObserver, observer_channel};
    use crate::task::{InFlightTaskCancellation, TaskID};
    use std::convert::Infallible;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    struct CancelCounter(Arc<AtomicUsize>);

    impl ExecutorNotified for CancelCounter {
        fn request_cancel(&mut self) {
            self.0.fetch_add(1, Ordering::Relaxed);
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    struct ReentrantWakeProbe {
        shared: Arc<super::Shared<u8>>,
        blocked_on_observer_lock: std::sync::atomic::AtomicBool,
    }

    #[cfg(not(target_arch = "wasm32"))]
    impl ReentrantWakeProbe {
        fn probe(&self) {
            use std::sync::mpsc;
            use std::time::Duration;

            let shared = self.shared.clone();
            let (acquired, wait_for_acquire) = mpsc::channel();
            std::thread::spawn(move || {
                let _guard = shared.lock.lock_sync();
                let _ = acquired.send(());
            });

            if wait_for_acquire
                .recv_timeout(Duration::from_millis(250))
                .is_err()
            {
                self.blocked_on_observer_lock.store(true, Ordering::Relaxed);
            }
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    impl std::task::Wake for ReentrantWakeProbe {
        fn wake(self: Arc<Self>) {
            self.probe();
        }

        fn wake_by_ref(self: &Arc<Self>) {
            self.probe();
        }
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test)]
    fn dropping_completed_observer_does_not_request_cancellation() {
        let cancellation_requests = Arc::new(AtomicUsize::new(0));
        let (mut sender, observer) = observer_channel::<u8, Infallible, CancelCounter>(
            None,
            Some(CancelCounter(cancellation_requests.clone())),
            InFlightTaskCancellation::default(),
            TaskID::from_u64(1),
        );

        sender.send(42);
        drop(sender);
        assert_eq!(observer.observe(), super::Observation::Ready(42));
        drop(observer);

        assert_eq!(
            cancellation_requests.load(Ordering::Relaxed),
            0,
            "dropping an observer after successful completion must not request cancellation"
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn observer_state_lock_is_released_before_waking() {
        use std::future::Future;
        use std::pin::Pin;
        use std::task::{Context, Poll, Waker};

        let (mut sender, mut observer) = observer_channel::<u8, Infallible, Infallible>(
            None,
            None,
            InFlightTaskCancellation::default(),
            TaskID::from_u64(2),
        );
        let probe = Arc::new(ReentrantWakeProbe {
            shared: observer.shared.clone(),
            blocked_on_observer_lock: std::sync::atomic::AtomicBool::new(false),
        });
        let waker = Waker::from(probe.clone());
        let mut context = Context::from_waker(&waker);

        assert_eq!(Pin::new(&mut observer).poll(&mut context), Poll::Pending);
        sender.send(7);

        assert!(
            !probe.blocked_on_observer_lock.load(Ordering::Relaxed),
            "the observer's waker ran while the shared state mutex was still locked"
        );
        observer.detach();
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test)]
    fn completing_task_leaves_cancellation_token_unset() {
        let cancellation = InFlightTaskCancellation::default();
        let (mut sender, observer) = observer_channel::<u8, Infallible, Infallible>(
            None,
            None,
            cancellation.clone(),
            TaskID::from_u64(3),
        );

        sender.send(42);
        drop(sender);

        assert!(
            !cancellation.is_cancelled(),
            "a task that ran to completion must not mark its cancellation token"
        );
        drop(observer);
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test)]
    fn dropping_sender_before_completion_sets_cancellation_token() {
        let cancellation = InFlightTaskCancellation::default();
        let (sender, observer) = observer_channel::<u8, Infallible, Infallible>(
            None,
            None,
            cancellation.clone(),
            TaskID::from_u64(4),
        );

        drop(sender);

        assert!(
            cancellation.is_cancelled(),
            "a task dropped before completion must mark its cancellation token"
        );
        drop(observer);
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test)]
    fn boxed_notifier_unboxes_the_value() {
        use std::any::Any;

        struct Inner(Arc<AtomicUsize>);
        impl super::ObserverNotified<dyn Any + Send> for Inner {
            fn notify(&mut self, value: &(dyn Any + Send)) {
                assert_eq!(
                    value.downcast_ref::<u32>().copied(),
                    Some(7),
                    "the erased notifier must receive the value, not the box around it"
                );
                self.0.fetch_add(1, Ordering::Relaxed);
            }
        }

        let notifications = Arc::new(AtomicUsize::new(0));
        let mut boxed: Box<dyn super::ObserverNotified<dyn Any + Send> + Send> =
            Box::new(Inner(notifications.clone()));
        let value: Box<dyn Any + Send> = Box::new(7u32);

        super::ObserverNotified::notify(&mut boxed, &value);

        assert_eq!(notifications.load(Ordering::Relaxed), 1);
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test)]
    fn test_send() {
        /* observer can send when the underlying value can */
        #[allow(unused)]
        fn ex<T: Send, E: ExecutorNotified + Send>(_observer: TypedObserver<T, E>) {
            fn assert_send<T: Send>() {}
            assert_send::<TypedObserver<T, E>>();
        }
    }
    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test)]
    fn test_unpin() {
        /* observer can unpin */
        #[allow(unused)]
        fn ex<T, E: ExecutorNotified + Unpin>(_observer: TypedObserver<T, E>) {
            fn assert_unpin<T: Unpin>() {}
            assert_unpin::<TypedObserver<T, E>>();
        }
    }
}