use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

use parking_lot::{Condvar, Mutex};

/// tells hydrant's own threads to wind down. they sleep on it instead of
/// `thread::sleep`, so a shutdown doesn't wait out a whole interval.
#[derive(Default)]
pub struct Stop {
    // the ingest shards check this per message, so it stays a plain load
    stopped: AtomicBool,
    lock: Mutex<()>,
    changed: Condvar,
}

impl Stop {
    pub fn trigger(&self) {
        // set under the lock, or a waiter that just checked could miss the notify
        let _lock = self.lock.lock();
        self.stopped.store(true, Ordering::Release);
        self.changed.notify_all();
    }

    pub fn is_set(&self) -> bool {
        self.stopped.load(Ordering::Acquire)
    }

    /// sleeps for `timeout`, or less if stopped meanwhile. true means stop.
    pub fn wait(&self, timeout: Duration) -> bool {
        let deadline = Instant::now() + timeout;
        let mut lock = self.lock.lock();
        while !self.is_set() {
            if self.changed.wait_until(&mut lock, deadline).timed_out() {
                break;
            }
        }
        self.is_set()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;

    #[test]
    fn wait_times_out_without_a_trigger() {
        let stop = Stop::default();
        let started = Instant::now();
        assert!(!stop.wait(Duration::from_millis(20)));
        assert!(started.elapsed() >= Duration::from_millis(20));
    }

    #[test]
    fn trigger_wakes_a_waiting_thread_early() {
        let stop = Arc::new(Stop::default());
        let waiter = std::thread::spawn({
            let stop = stop.clone();
            move || {
                let started = Instant::now();
                (stop.wait(Duration::from_secs(60)), started.elapsed())
            }
        });
        std::thread::sleep(Duration::from_millis(20));
        stop.trigger();
        let (stopped, waited) = waiter.join().unwrap();
        assert!(stopped);
        assert!(waited < Duration::from_secs(5));
        // and stays set for anyone who checks later
        assert!(stop.wait(Duration::from_secs(60)));
    }
}