grafton_ndi/

async_runtime.rs

//! Async runtime integration for Tokio and smol.
//!
//! This module provides async wrappers around the synchronous NDI receiver API,
//! allowing integration with async Rust applications using Tokio or smol runtimes.
//!
//! The NDI SDK operations are inherently synchronous and blocking, so these wrappers
//! use `spawn_blocking` internally to run NDI operations on a thread pool without
//! blocking the async runtime.
//!
//! For `capture` and `try_capture`, the timeout budget starts when the future
//! is first polled, rather than when it is constructed. The same deadline covers
//! blocking-pool queue delay, receiver capture-gate contention, and SDK waits.
//! This is a wait budget, not a hard completion deadline: scheduling, frame copying,
//! and other processing can delay the result beyond it.
//!
//! Dropping a future sets a cancellation flag. Work is skipped if the blocking
//! closure observes that flag at its entry gate. After passing that gate, work
//! continues, including waiting for receiver access; it cannot be interrupted by
//! dropping the future.
//!
//! # Features
//!
//! - `tokio` - Enable Tokio runtime support
//! - `smol` - Enable smol runtime support
//!
//! # Example with Tokio
//!
//! ```no_run
//! # #[cfg(feature = "tokio")]
//! # {
//! use grafton_ndi::{NDI, ReceiverOptionsBuilder, tokio::AsyncReceiver};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), grafton_ndi::Error> {
//!     let ndi = NDI::new()?;
//!     // ... obtain source from finder ...
//!     # let source = grafton_ndi::Source {
//!     #     name: "Test".into(),
//!     #     address: grafton_ndi::SourceAddress::None
//!     # };
//!
//!     let options = ReceiverOptionsBuilder::snapshot_preset(source).build();
//!     let receiver = grafton_ndi::Receiver::new(&ndi, &options)?;
//!
//!     let async_receiver = AsyncReceiver::new(receiver);
//!
//!     // Capture frame asynchronously without blocking the runtime
//!     let frame = async_receiver.video().capture(std::time::Duration::from_secs(5)).await?;
//!     println!("Captured {}x{} frame", frame.width(), frame.height());
//!
//!     Ok(())
//! }
//! # }
//! ```

use std::{
    future::Future,
    marker::PhantomData,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
    time::{Duration, Instant},
};

use crate::{
    capture::{AudioKind, CaptureKind, MetadataKind, VideoKind},
    to_ms_checked, ConnectionStats, Error, Receiver, Result,
};

// Dropping a future prevents its queued closure from starting SDK work. Work
// that has already passed this gate cannot be interrupted safely.
struct CancelOnDrop(Arc<AtomicBool>);

impl Drop for CancelOnDrop {
    fn drop(&mut self) {
        self.0.store(true, Ordering::Release);
    }
}

fn blocking_operation<F, T>(operation: F) -> (CancelOnDrop, impl FnOnce() -> Result<T> + Send)
where
    F: FnOnce() -> T + Send,
{
    let cancelled = Arc::new(AtomicBool::new(false));
    let guard = CancelOnDrop(Arc::clone(&cancelled));
    let work = move || {
        if cancelled.load(Ordering::Acquire) {
            return Err(Error::SpawnFailed(
                "Blocking operation cancelled before starting".into(),
            ));
        }
        std::panic::catch_unwind(std::panic::AssertUnwindSafe(operation)).map_err(|payload| {
            let message = payload
                .downcast_ref::<String>()
                .map(String::as_str)
                .or_else(|| payload.downcast_ref::<&str>().copied())
                .unwrap_or("non-string panic payload");
            Error::SpawnFailed(format!("Blocking operation panicked: {message}"))
        })
    };
    (guard, work)
}

/// Trait for async runtime spawn-blocking abstraction.
///
/// This trait enables runtime-agnostic async code by abstracting the spawn-blocking
/// mechanism. Each runtime (Tokio, smol) provides its own implementation.
///
/// The trait is sealed to prevent external implementations and ensure consistent
/// error handling across all supported runtimes.
pub trait SpawnBlocking: sealed::Sealed + Clone + Send + Sync + 'static {
    /// Spawns a blocking operation on a thread pool and returns its result.
    ///
    /// # Errors
    ///
    /// Returns `Err(Error::SpawnFailed)` if the blocking task panics or is cancelled.
    fn spawn_blocking<F, R>(f: F) -> impl Future<Output = Result<R>> + Send
    where
        F: FnOnce() -> R + Send + 'static,
        R: Send + 'static;
}

mod sealed {
    pub trait Sealed {}

    #[cfg(feature = "tokio")]
    impl Sealed for super::TokioRuntime {}

    #[cfg(feature = "smol")]
    impl Sealed for super::SmolRuntime {}
}

/// Tokio async runtime marker type.
///
/// Used as a type parameter for [`AsyncReceiverGeneric`] to select Tokio's
/// `spawn_blocking` implementation.
#[cfg(feature = "tokio")]
#[derive(Clone, Copy, Debug, Default)]
pub struct TokioRuntime;

#[cfg(feature = "tokio")]
impl SpawnBlocking for TokioRuntime {
    // Using `impl Future` instead of `async fn` in trait because we need explicit
    // Send bounds on the returned future. This pattern is intentional.
    #[allow(clippy::manual_async_fn)]
    fn spawn_blocking<F, R>(f: F) -> impl Future<Output = Result<R>> + Send
    where
        F: FnOnce() -> R + Send + 'static,
        R: Send + 'static,
    {
        async {
            let handle = ::tokio::runtime::Handle::try_current()
                .map_err(|e| Error::SpawnFailed(e.to_string()))?;
            let (_cancellation, work) = blocking_operation(f);
            handle
                .spawn_blocking(work)
                .await
                .map_err(|e| Error::SpawnFailed(e.to_string()))?
        }
    }
}

/// smol runtime marker type.
///
/// Used as a type parameter for [`AsyncReceiverGeneric`] to select smol's
/// `unblock` implementation.
#[cfg(feature = "smol")]
#[derive(Clone, Copy, Debug, Default)]
pub struct SmolRuntime;

#[cfg(feature = "smol")]
impl SpawnBlocking for SmolRuntime {
    // Using `impl Future` instead of `async fn` in trait because we need explicit
    // Send bounds on the returned future. This pattern is intentional.
    #[allow(clippy::manual_async_fn)]
    fn spawn_blocking<F, R>(f: F) -> impl Future<Output = Result<R>> + Send
    where
        F: FnOnce() -> R + Send + 'static,
        R: Send + 'static,
    {
        async {
            let (_cancellation, work) = blocking_operation(f);
            ::smol::unblock(work).await
        }
    }
}

/// Generic async receiver wrapper parameterized by runtime.
///
/// This struct provides async versions of the [`Receiver`] methods by running
/// blocking NDI operations on the runtime's thread pool using `spawn_blocking`.
///
/// # Type Parameters
///
/// - `R`: The async runtime type, implementing [`SpawnBlocking`]. Use
///   [`TokioRuntime`] or [`SmolRuntime`].
///
/// # Thread Safety
///
/// The underlying [`Receiver`] is wrapped in an [`Arc`] to allow sharing across
/// async tasks and safe cloning. The NDI SDK receiver is inherently thread-safe.
///
/// # Example
///
/// ```no_run
/// # #[cfg(feature = "tokio")]
/// # {
/// use grafton_ndi::{NDI, ReceiverOptionsBuilder, tokio::AsyncReceiver};
///
/// #[tokio::main]
/// async fn main() -> Result<(), grafton_ndi::Error> {
///     let ndi = NDI::new()?;
///     // ... obtain source ...
///     # let source = grafton_ndi::Source {
///     #     name: "Test".into(),
///     #     address: grafton_ndi::SourceAddress::None
///     # };
///
///     let options = ReceiverOptionsBuilder::snapshot_preset(source).build();
///     let receiver = grafton_ndi::Receiver::new(&ndi, &options)?;
///     let async_receiver = AsyncReceiver::new(receiver);
///
///     // Non-blocking async capture
///     match async_receiver.video().try_capture(std::time::Duration::from_millis(100)).await? {
///         Some(frame) => println!("Got frame: {}x{}", frame.width(), frame.height()),
///         None => println!("No frame available"),
///     }
///
///     Ok(())
/// }
/// # }
/// ```
pub struct AsyncReceiverGeneric<R: SpawnBlocking> {
    inner: Arc<Receiver>,
    _runtime: PhantomData<R>,
}

fn validated_deadline(timeout: Duration) -> Result<Instant> {
    to_ms_checked(timeout)?;
    Instant::now()
        .checked_add(timeout)
        .ok_or_else(|| Error::Timeout("Capture deadline cannot be represented".into()))
}

// Shared by both capture verbs so validation and queue accounting agree.
async fn with_timeout_budget<R, F, T>(timeout: Duration, operation: F) -> Result<T>
where
    R: SpawnBlocking,
    F: FnOnce(Instant) -> Result<T> + Send + 'static,
    T: Send + 'static,
{
    let deadline = validated_deadline(timeout)?;
    R::spawn_blocking(move || operation(deadline)).await?
}

impl<R: SpawnBlocking> AsyncReceiverGeneric<R> {
    /// Create a new async receiver wrapper.
    ///
    /// The receiver is wrapped in an [`Arc`] to allow sharing across async tasks.
    pub fn new(receiver: Receiver) -> Self {
        Self {
            inner: Arc::new(receiver),
            _runtime: PhantomData,
        }
    }

    /// Capture **video** frames without blocking the async runtime.
    ///
    /// Returns an [`AsyncCapture`] view; see [`AsyncCapture`] for its verbs
    /// ([`capture`](AsyncCapture::capture),
    /// [`try_capture`](AsyncCapture::try_capture)). Mirrors
    /// [`Receiver::video`], running each capture on the blocking pool.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # #[cfg(feature = "tokio")]
    /// # {
    /// # use grafton_ndi::{NDI, ReceiverOptions, tokio::AsyncReceiver};
    /// # use std::time::Duration;
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), grafton_ndi::Error> {
    /// # let ndi = NDI::new()?;
    /// # let source = grafton_ndi::Source {
    /// #     name: "Test".into(),
    /// #     address: grafton_ndi::SourceAddress::None
    /// # };
    /// # let options = ReceiverOptions::builder(source).build();
    /// # let receiver = grafton_ndi::Receiver::new(&ndi, &options)?;
    /// let async_receiver = AsyncReceiver::new(receiver);
    /// let frame = async_receiver.video().capture(Duration::from_secs(5)).await?;
    /// println!("Captured {}x{} frame", frame.width(), frame.height());
    /// # Ok(())
    /// # }
    /// # }
    /// ```
    #[must_use = "the AsyncCapture view does nothing until a capture verb is awaited"]
    pub fn video(&self) -> AsyncCapture<'_, R, VideoKind> {
        AsyncCapture::new(self)
    }

    /// Capture **audio** frames without blocking the async runtime.
    ///
    /// Returns an [`AsyncCapture`] view; see [`video`](Self::video) for usage.
    /// Mirrors [`Receiver::audio`].
    #[must_use = "the AsyncCapture view does nothing until a capture verb is awaited"]
    pub fn audio(&self) -> AsyncCapture<'_, R, AudioKind> {
        AsyncCapture::new(self)
    }

    /// Capture **metadata** frames without blocking the async runtime.
    ///
    /// Returns an [`AsyncCapture`] view; see [`video`](Self::video) for usage.
    /// Mirrors [`Receiver::metadata`].
    #[must_use = "the AsyncCapture view does nothing until a capture verb is awaited"]
    pub fn metadata(&self) -> AsyncCapture<'_, R, MetadataKind> {
        AsyncCapture::new(self)
    }

    /// Whether the underlying receiver currently has at least one active
    /// connection to its source. See [`Receiver::is_connected`].
    ///
    /// A cheap, non-blocking SDK query, so it runs inline rather than on
    /// the blocking pool.
    pub fn is_connected(&self) -> bool {
        self.inner.is_connected()
    }

    /// Connection and frame-throughput statistics for the underlying
    /// receiver. See [`Receiver::connection_stats`].
    ///
    /// A cheap, non-blocking SDK query, so it runs inline rather than on
    /// the blocking pool. `connection_stats().video_frames_received` is
    /// the canonical liveness signal: it advances as the receiver pulls
    /// frames off the network, independent of how often the caller
    /// captures, so a frozen counter means the feed itself has stalled.
    pub fn connection_stats(&self) -> ConnectionStats {
        self.inner.connection_stats()
    }

    /// Re-establish the underlying receiver's connection to its source
    /// in place. See [`Receiver::reconnect`].
    ///
    /// Unlike the liveness probes above, a reconnect takes the receiver's
    /// capture guard exclusively, so it can block until in-flight captures
    /// (which run on the blocking pool) drain. It therefore runs on the
    /// blocking pool too, never on the async runtime's threads, and is safe to
    /// call while captures on this receiver are in flight — they serialize
    /// instead of racing. Confirm recovery via [`Self::connection_stats`].
    ///
    /// # Cancellation
    ///
    /// Dropping this future skips work if the blocking closure observes the
    /// cancellation flag at its entry gate. After passing that gate, reconnect
    /// continues, including waiting for in-flight captures, and may reconnect
    /// after this future is dropped.
    pub async fn reconnect(&self) -> Result<()> {
        let receiver = Arc::clone(&self.inner);
        R::spawn_blocking(move || receiver.reconnect()).await?
    }
}

/// A typed view over an async receiver for capturing frames of one kind,
/// without blocking the async runtime.
///
/// Created by [`AsyncReceiverGeneric::video`], [`audio`](AsyncReceiverGeneric::audio),
/// and [`metadata`](AsyncReceiverGeneric::metadata). Each verb runs the
/// underlying synchronous capture on the runtime's blocking pool:
///
/// - [`capture`](Self::capture) — reliable owned capture with built-in retry.
///   The timeout budget starts when the future is first polled; `spawn_blocking`
///   queue delay is subtracted before the SDK begins waiting.
/// - [`try_capture`](Self::try_capture) — a single owned poll; `Ok(None)` when
///   no frame is ready. Its budget also includes blocking task queue delay.
///
/// There is no zero-copy `try_capture_ref` here: a borrowed frame is tied to
/// the receiver and cannot cross the `spawn_blocking` boundary. Use
/// [`Receiver::video`]'s [`Capture::try_capture_ref`](crate::Capture::try_capture_ref)
/// on the synchronous receiver for in-place processing.
pub struct AsyncCapture<'rx, R: SpawnBlocking, K: CaptureKind> {
    recv: &'rx AsyncReceiverGeneric<R>,
    _kind: PhantomData<K>,
}

impl<'rx, R: SpawnBlocking, K: CaptureKind> AsyncCapture<'rx, R, K>
where
    K::Owned: Send + 'static,
{
    fn new(recv: &'rx AsyncReceiverGeneric<R>) -> Self {
        Self {
            recv,
            _kind: PhantomData,
        }
    }

    /// Async version of [`Capture::capture`](crate::Capture::capture):
    /// reliable owned capture that retries across the SDK's initial-sync
    /// warm-up, run on the blocking pool.
    ///
    /// # Arguments
    ///
    /// * `timeout` - Total budget to wait for a frame, starting when the future
    ///   is first polled. Blocking-pool queue delay and capture-gate contention
    ///   count against the same budget. Completion can occur after this budget
    ///   because scheduling and frame processing are not bounded by it.
    ///   Must not exceed
    ///   [`crate::MAX_TIMEOUT`] (~49.7 days).
    ///
    /// # Returns
    ///
    /// * `Ok(frame)` - Successfully captured a frame
    /// * `Err(Error::FrameTimeout)` - No frame received within timeout (includes retry details)
    /// * `Err(Error::SpawnFailed)` - The blocking task panicked or was cancelled
    /// * `Err(_)` - Another error occurred during capture
    ///
    /// # Cancellation
    ///
    /// Dropping this future skips work if the blocking closure observes the
    /// cancellation flag at its entry gate. After passing that gate, work
    /// continues, including waiting for receiver access, and can consume a frame
    /// whose result is discarded.
    pub async fn capture(&self, timeout: Duration) -> Result<K::Owned> {
        let receiver = Arc::clone(&self.recv.inner);
        with_timeout_budget::<R, _, _>(timeout, move |deadline| {
            receiver.core.capture_kind_until::<K>(deadline)
        })
        .await
    }

    /// Async version of
    /// [`Capture::try_capture`](crate::Capture::try_capture): a single owned
    /// poll, run on the blocking pool.
    ///
    /// # Arguments
    ///
    /// * `timeout` - Total wait budget from the future's first poll, including
    ///   blocking-pool queue delay and capture-gate contention. With an exhausted
    ///   budget, an immediately available gate permits a zero-timeout SDK poll;
    ///   otherwise this returns `Ok(None)` without an SDK call. Scheduling and
    ///   frame processing can delay completion beyond this budget.
    ///   Must not exceed
    ///   [`crate::MAX_TIMEOUT`] (~49.7 days).
    ///
    /// # Returns
    ///
    /// * `Ok(Some(frame))` - Successfully captured a frame
    /// * `Ok(None)` - No frame available within timeout
    /// * `Err(Error::SpawnFailed)` - The blocking task panicked or was cancelled
    /// * `Err(_)` - An error occurred during capture
    ///
    /// # Cancellation
    ///
    /// Dropping this future skips work if the blocking closure observes the
    /// cancellation flag at its entry gate. After passing that gate, work
    /// continues, including waiting for receiver access, and can consume a frame
    /// whose result is discarded.
    pub async fn try_capture(&self, timeout: Duration) -> Result<Option<K::Owned>> {
        let receiver = Arc::clone(&self.recv.inner);
        with_timeout_budget::<R, _, _>(timeout, move |deadline| {
            receiver.core.try_capture_kind_until::<K>(deadline)
        })
        .await
    }
}

impl<R: SpawnBlocking> Clone for AsyncReceiverGeneric<R> {
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
            _runtime: PhantomData,
        }
    }
}

// Runtime-specific receiver aliases

#[cfg(feature = "tokio")]
pub mod tokio {
    //! Tokio async runtime integration.
    //!
    //! Provides [`AsyncReceiver`] wrapper that uses `tokio::task::spawn_blocking`
    //! to run NDI operations without blocking the Tokio runtime.
    //!
    //! # Example
    //!
    //! ```no_run
    //! # #[cfg(feature = "tokio")]
    //! # {
    //! use grafton_ndi::{NDI, ReceiverOptionsBuilder, tokio::AsyncReceiver};
    //!
    //! #[tokio::main]
    //! async fn main() -> Result<(), grafton_ndi::Error> {
    //!     let ndi = NDI::new()?;
    //!     // ... obtain source ...
    //!     # let source = grafton_ndi::Source {
    //!     #     name: "Test".into(),
    //!     #     address: grafton_ndi::SourceAddress::None
    //!     # };
    //!
    //!     let options = ReceiverOptionsBuilder::snapshot_preset(source).build();
    //!     let receiver = grafton_ndi::Receiver::new(&ndi, &options)?;
    //!     let async_receiver = AsyncReceiver::new(receiver);
    //!
    //!     // Non-blocking async capture
    //!     match async_receiver.video().try_capture(std::time::Duration::from_millis(100)).await? {
    //!         Some(frame) => println!("Got frame: {}x{}", frame.width(), frame.height()),
    //!         None => println!("No frame available"),
    //!     }
    //!
    //!     Ok(())
    //! }
    //! # }
    //! ```

    use super::{AsyncReceiverGeneric, TokioRuntime};

    /// Async receiver wrapper for Tokio runtime.
    ///
    /// This is a type alias for the generic async receiver parameterized with
    /// the Tokio runtime. It provides async versions of the [`crate::Receiver`]
    /// methods by running blocking NDI operations on Tokio's blocking thread
    /// pool using `spawn_blocking`.
    ///
    /// # Thread Safety
    ///
    /// The underlying `Receiver` is wrapped in an `Arc` to allow sharing across
    /// async tasks and safe cloning. The NDI SDK receiver is inherently thread-safe.
    ///
    /// # Error Handling
    ///
    /// All methods return [`crate::Result`], converting any task panic or cancellation
    /// into [`crate::Error::SpawnFailed`] rather than propagating the panic.
    pub type AsyncReceiver = AsyncReceiverGeneric<TokioRuntime>;
}

#[cfg(feature = "smol")]
pub mod smol {
    //! smol runtime integration.
    //!
    //! Provides [`AsyncReceiver`] wrapper that uses `::smol::unblock`
    //! to run NDI operations without blocking the smol runtime.
    //!
    //! # Example
    //!
    //! ```no_run
    //! # #[cfg(feature = "smol")]
    //! # {
    //! use grafton_ndi::{NDI, ReceiverOptionsBuilder, smol::AsyncReceiver};
    //!
    //! fn main() -> Result<(), grafton_ndi::Error> {
    //!     ::smol::block_on(async {
    //!         let ndi = NDI::new()?;
    //!         // ... obtain source ...
    //!         # let source = grafton_ndi::Source {
    //!         #     name: "Test".into(),
    //!         #     address: grafton_ndi::SourceAddress::None
    //!         # };
    //!
    //!         let options = ReceiverOptionsBuilder::snapshot_preset(source).build();
    //!         let receiver = grafton_ndi::Receiver::new(&ndi, &options)?;
    //!         let async_receiver = AsyncReceiver::new(receiver);
    //!
    //!         // Non-blocking async capture
    //!         match async_receiver.video().try_capture(std::time::Duration::from_millis(100)).await? {
    //!             Some(frame) => println!("Got frame: {}x{}", frame.width(), frame.height()),
    //!             None => println!("No frame available"),
    //!         }
    //!
    //!         Ok(())
    //!     })
    //! }
    //! # }
    //! ```

    use super::{AsyncReceiverGeneric, SmolRuntime};

    /// Async receiver wrapper for smol runtime.
    ///
    /// This is a type alias for the generic async receiver parameterized with
    /// the smol runtime. It provides async versions of the [`crate::Receiver`]
    /// methods by running blocking NDI operations on smol's blocking thread
    /// pool using `::smol::unblock`.
    ///
    /// # Thread Safety
    ///
    /// The underlying `Receiver` is wrapped in an `Arc` to allow sharing across
    /// async tasks and safe cloning. The NDI SDK receiver is inherently thread-safe.
    ///
    /// # Error Handling
    ///
    /// All methods return [`crate::Result`], converting blocking-operation
    /// panics into [`crate::Error::SpawnFailed`].
    pub type AsyncReceiver = AsyncReceiverGeneric<SmolRuntime>;
}

#[cfg(all(test, feature = "tokio"))]
mod timeout_tests {
    use super::*;
    use std::{
        sync::mpsc,
        task::{Context, Poll, Waker},
    };

    fn runtime() -> ::tokio::runtime::Runtime {
        ::tokio::runtime::Builder::new_current_thread()
            .max_blocking_threads(1)
            .build()
            .unwrap()
    }

    #[test]
    fn queued_work_uses_remaining_budget() {
        let runtime = runtime();
        runtime.block_on(async {
            let (started_tx, started_rx) = mpsc::channel();
            let (release_tx, release_rx) = mpsc::channel();
            let blocker = ::tokio::task::spawn_blocking(move || {
                started_tx.send(()).unwrap();
                release_rx.recv().unwrap();
            });
            started_rx.recv().unwrap();

            let budget = Duration::from_millis(20);
            let mut capture = Box::pin(with_timeout_budget::<TokioRuntime, _, _>(
                budget,
                |deadline| Ok(deadline.saturating_duration_since(Instant::now())),
            ));
            let mut context = Context::from_waker(Waker::noop());
            assert!(matches!(capture.as_mut().poll(&mut context), Poll::Pending));
            // Keep the sole worker occupied until the whole budget has elapsed.
            std::thread::sleep(budget + Duration::from_millis(5));
            release_tx.send(()).unwrap();
            assert_eq!(capture.await.unwrap(), Duration::ZERO);
            blocker.await.unwrap();
        });
    }

    #[test]
    fn invalid_budget_is_rejected_before_spawning() {
        // No entered runtime: trying to submit work would panic.
        let mut capture = Box::pin(with_timeout_budget::<TokioRuntime, _, ()>(
            crate::MAX_TIMEOUT + Duration::from_secs(1),
            |_| panic!("invalid timeout must not execute blocking work"),
        ));
        let mut context = Context::from_waker(Waker::noop());
        assert!(matches!(
            capture.as_mut().poll(&mut context),
            Poll::Ready(Err(_))
        ));
    }
}

#[cfg(all(test, feature = "smol"))]
mod smol_tests {
    use super::*;
    use std::{
        sync::mpsc,
        task::{Context, Poll, Waker},
    };

    #[test]
    fn dropping_unpolled_future_starts_no_work() {
        let called = Arc::new(AtomicBool::new(false));
        let worker_called = Arc::clone(&called);
        let operation = SmolRuntime::spawn_blocking(move || {
            worker_called.store(true, Ordering::Release);
        });
        drop(operation);
        assert!(!called.load(Ordering::Acquire));
    }

    #[test]
    fn dropping_started_future_allows_work_to_finish() {
        let (started_tx, started_rx) = mpsc::channel();
        let (release_tx, release_rx) = mpsc::channel();
        let (finished_tx, finished_rx) = mpsc::channel();
        let mut operation = Box::pin(SmolRuntime::spawn_blocking(move || {
            started_tx.send(()).unwrap();
            release_rx.recv().unwrap();
            finished_tx.send(42).unwrap();
        }));
        let mut context = Context::from_waker(Waker::noop());
        assert!(matches!(
            operation.as_mut().poll(&mut context),
            Poll::Pending
        ));
        started_rx.recv_timeout(Duration::from_secs(5)).unwrap();
        drop(operation);
        release_tx.send(()).unwrap();
        assert_eq!(
            finished_rx.recv_timeout(Duration::from_secs(5)).unwrap(),
            42
        );
    }

    #[test]
    fn invalid_budget_does_not_start_work() {
        let called = Arc::new(AtomicBool::new(false));
        let worker_called = Arc::clone(&called);
        let result = ::smol::block_on(with_timeout_budget::<SmolRuntime, _, _>(
            crate::MAX_TIMEOUT + Duration::from_secs(1),
            move |_| {
                worker_called.store(true, Ordering::Release);
                Ok(())
            },
        ));
        assert!(result.is_err());
        assert!(!called.load(Ordering::Acquire));
    }

    #[test]
    fn budget_starts_on_first_poll_and_keeps_one_deadline() {
        let budget = Duration::from_millis(20);
        let (started_tx, started_rx) = mpsc::channel();
        let (release_tx, release_rx) = mpsc::channel();
        let mut operation = Box::pin(with_timeout_budget::<SmolRuntime, _, _>(
            budget,
            move |deadline| {
                started_tx.send(deadline).unwrap();
                release_rx.recv().unwrap();
                Ok(deadline.saturating_duration_since(Instant::now()))
            },
        ));
        // Construction must not consume the wait budget.
        std::thread::sleep(budget);
        let before_poll = Instant::now();
        let mut context = Context::from_waker(Waker::noop());
        assert!(matches!(
            operation.as_mut().poll(&mut context),
            Poll::Pending
        ));
        let after_poll = Instant::now();
        let deadline = started_rx.recv_timeout(Duration::from_secs(5)).unwrap();
        assert!(deadline >= before_poll + budget);
        assert!(deadline <= after_poll + budget);
        // A wait inside blocking work must not reset the shared capture deadline.
        std::thread::sleep(deadline.saturating_duration_since(Instant::now()));
        release_tx.send(()).unwrap();
        assert_eq!(::smol::block_on(operation).unwrap(), Duration::ZERO);
    }
}

#[cfg(test)]
mod executor_tests {
    use super::*;

    #[test]
    fn cancellation_skips_queued_operation() {
        let (guard, work) = blocking_operation(|| panic!("cancelled work must not run"));
        drop(guard);
        assert!(matches!(work(), Err(Error::SpawnFailed(_))));
    }

    #[test]
    fn cancellation_allows_started_operation_to_finish() {
        let (started_tx, started_rx) = std::sync::mpsc::channel();
        let (finish_tx, finish_rx) = std::sync::mpsc::channel();
        let (guard, work) = blocking_operation(move || {
            started_tx.send(()).unwrap();
            finish_rx.recv().unwrap();
            42
        });
        let worker = std::thread::spawn(work);
        started_rx.recv().unwrap();
        drop(guard);
        finish_tx.send(()).unwrap();
        assert_eq!(worker.join().unwrap().unwrap(), 42);
    }

    #[cfg(feature = "tokio")]
    #[test]
    fn missing_tokio_context_returns_error() {
        use std::task::{Context, Poll, Waker};
        let mut operation = Box::pin(TokioRuntime::spawn_blocking(|| 42));
        let mut context = Context::from_waker(Waker::noop());
        assert!(matches!(
            operation.as_mut().poll(&mut context),
            Poll::Ready(Err(Error::SpawnFailed(_)))
        ));
    }

    #[cfg(feature = "tokio")]
    #[test]
    fn tokio_converts_blocking_panic() {
        let runtime = ::tokio::runtime::Builder::new_current_thread()
            .build()
            .unwrap();
        let result = runtime.block_on(TokioRuntime::spawn_blocking(|| panic!("test panic")));
        assert!(
            matches!(result, Err(Error::SpawnFailed(message)) if message.contains("test panic"))
        );
    }

    #[cfg(feature = "smol")]
    #[test]
    fn smol_converts_blocking_panic() {
        let result = ::smol::block_on(SmolRuntime::spawn_blocking(|| panic!("test panic")));
        assert!(
            matches!(result, Err(Error::SpawnFailed(message)) if message.contains("test panic"))
        );
    }

    #[cfg(feature = "tokio")]
    #[test]
    fn dropping_tokio_future_skips_queued_work() {
        use std::task::{Context, Poll, Waker};
        let runtime = ::tokio::runtime::Builder::new_current_thread()
            .max_blocking_threads(1)
            .build()
            .unwrap();
        runtime.block_on(async {
            let (started_tx, started_rx) = std::sync::mpsc::channel();
            let (release_tx, release_rx) = std::sync::mpsc::channel();
            let blocker = ::tokio::task::spawn_blocking(move || {
                started_tx.send(()).unwrap();
                release_rx.recv().unwrap();
            });
            started_rx.recv().unwrap();
            let called = Arc::new(AtomicBool::new(false));
            let worker_called = Arc::clone(&called);
            let mut operation = Box::pin(TokioRuntime::spawn_blocking(move || {
                worker_called.store(true, Ordering::Release)
            }));
            let mut context = Context::from_waker(Waker::noop());
            assert!(matches!(
                operation.as_mut().poll(&mut context),
                Poll::Pending
            ));
            drop(operation);
            release_tx.send(()).unwrap();
            blocker.await.unwrap();
            // The single worker processes the cancelled closure before this sentinel.
            ::tokio::task::spawn_blocking(|| ()).await.unwrap();
            assert!(!called.load(Ordering::Acquire));
        });
    }
}