pub mod environment;
pub mod io;
pub mod messaging;

use std::future::Future;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use tokio::runtime::{Handle, Runtime};

/// The Tokio runtime shared by the filesystem and network extensions.
pub struct Host(Option<Runtime>);

impl Host {
    pub fn new() -> std::io::Result<Self> {
        let runtime = tokio::runtime::Builder::new_multi_thread()
            // A second worker keeps decompression or a large message from stalling other connections.
            .worker_threads(2)
            .thread_name("avril:host")
            .enable_all()
            .build()?;
        Ok(Self(Some(runtime)))
    }

    pub fn handle(&self) -> Handle {
        self.0.as_ref().unwrap().handle().clone()
    }
}

impl Drop for Host {
    fn drop(&mut self) {
        // getaddrinfo may still be running in Tokio's blocking pool after an abort.
        if let Some(runtime) = self.0.take() {
            runtime.shutdown_background();
        }
    }
}

pub fn interrupted(scope: &avril_core::v8::PinScope, flag: &AtomicBool) -> bool {
    if flag.load(Ordering::SeqCst) {
        // V8 may not observe a termination request until this native call returns.
        scope.terminate_execution();
        true
    } else {
        scope.is_execution_terminating()
    }
}

struct WakeThread(std::thread::Thread);

impl std::task::Wake for WakeThread {
    fn wake(self: Arc<Self>) {
        self.0.unpark();
    }

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

// Worker and REPL loops may already be inside a futures executor. Poll only
// this op's native work, without entering an executor or pumping JavaScript.
pub fn wait<F: Future>(future: F, interrupted: impl Fn() -> bool) -> Option<F::Output> {
    let waker = std::task::Waker::from(Arc::new(WakeThread(std::thread::current())));
    let mut context = std::task::Context::from_waker(&waker);
    // A synchronous wait cannot yield to the outer task to replenish its budget.
    let mut pending = std::pin::pin!(tokio::task::unconstrained(future));
    loop {
        if interrupted() {
            return None;
        }
        if let std::task::Poll::Ready(result) = pending.as_mut().poll(&mut context) {
            return Some(result);
        }
        std::thread::park_timeout(std::time::Duration::from_millis(10));
    }
}

#[cfg(test)]
mod tests {
    use super::{Host, wait};
    use std::time::{Duration, Instant};

    #[test]
    fn synchronous_wait_drains_tokio_io_inside_an_async_task() {
        let host = Host::new().unwrap();
        let (sender, mut receiver) = tokio::sync::mpsc::channel(1_000);
        for value in 0..1_000 {
            sender.try_send(value).unwrap();
        }
        drop(sender);

        host.handle().block_on(async {
            let deadline = Instant::now() + Duration::from_secs(2);
            let result = wait(
                async {
                    let mut values = Vec::new();
                    while let Some(value) = receiver.recv().await {
                        values.push(value);
                    }
                    values
                },
                || Instant::now() >= deadline,
            );
            assert_eq!(result, Some((0..1_000).collect()));
        });
    }
}