ncd 0.1.3

use std::collections::VecDeque;

use parking_lot::{Condvar, Mutex};

pub type Task = Box<dyn FnOnce() + Send + 'static>;

struct QueueState {
    closed: bool,
    tasks: VecDeque<Task>,
}

pub struct TaskQueue {
    condvar: Condvar,
    state: Mutex<QueueState>,
}

impl TaskQueue {
    pub const fn new() -> Self {
        Self {
            condvar: Condvar::new(),
            state: Mutex::new(QueueState {
                closed: false,
                tasks: VecDeque::new(),
            }),
        }
    }

    pub fn push(&self, task: Task) -> Result<(), Task> {
        let mut state = self.state.lock();
        if state.closed {
            return Err(task);
        }

        state.tasks.push_back(task);
        drop(state);

        self.condvar.notify_one();
        Ok(())
    }

    pub fn pop(&self) -> Option<Task> {
        let mut state = self.state.lock();
        loop {
            if let Some(task) = state.tasks.pop_front() {
                return Some(task);
            }

            if state.closed {
                return None;
            }

            self.condvar.wait(&mut state);
        }
    }

    pub fn close(&self) {
        let mut state = self.state.lock();
        state.closed = true;
        drop(state);

        self.condvar.notify_all();
    }
}

#[cfg(test)]
mod tests {
    use std::{
        sync::{mpsc, Arc},
        thread,
        time::Duration,
    };

    use super::*;

    #[test]
    fn queue_is_fifo_and_drains_after_close() {
        let queue = TaskQueue::new();
        let (sender, receiver) = mpsc::channel();

        for value in 0..3 {
            let sender = sender.clone();
            assert!(queue
                .push(Box::new(move || sender.send(value).unwrap()))
                .is_ok());
        }

        queue.close();
        drop(sender);

        while let Some(task) = queue.pop() {
            task();
        }

        assert_eq!(receiver.iter().collect::<Vec<_>>(), vec![0, 1, 2]);
    }

    #[test]
    fn close_wakes_waiting_consumers() {
        let queue = Arc::new(TaskQueue::new());
        let worker_queue = Arc::clone(&queue);
        let (sender, receiver) = mpsc::channel();

        let worker = thread::spawn(move || sender.send(worker_queue.pop().is_none()).unwrap());
        thread::sleep(Duration::from_millis(10));
        queue.close();

        assert!(receiver.recv_timeout(Duration::from_secs(1)).unwrap());
        worker.join().unwrap();
    }

    #[test]
    fn closed_queue_rejects_tasks() {
        let queue = TaskQueue::new();
        queue.close();

        let task = queue.push(Box::new(|| {})).unwrap_err();
        drop(task);
        assert!(queue.pop().is_none());
    }
}