Build a Bounded Queue with Mutex and Condvar

use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;

struct QueueState<T> {
    items: VecDeque<T>,
    closed: bool,
}

struct BoundedQueue<T> {
    capacity: usize,
    state: Mutex<QueueState<T>>,
    ready: Condvar,
    space: Condvar,
}

impl<T> BoundedQueue<T> {
    fn new(capacity: usize) -> Arc<Self> {
        assert!(capacity > 0, "capacity must be positive");
        Arc::new(Self {
            capacity,
            state: Mutex::new(QueueState {
                items: VecDeque::with_capacity(capacity),
                closed: false,
            }),
            ready: Condvar::new(),
            space: Condvar::new(),
        })
    }

    fn push(&self, item: T) {
        let mut state = self.state.lock().expect("queue poisoned");
        while state.items.len() == self.capacity && !state.closed {
            state = self.space.wait(state).expect("queue poisoned");
        }
        if state.closed {
            return;
        }
        state.items.push_back(item);
        self.ready.notify_one();
    }

    fn pop(&self) -> Option<T> {
        let mut state = self.state.lock().expect("queue poisoned");
        loop {
            if let Some(item) = state.items.pop_front() {
                self.space.notify_one();
                return Some(item);
            }
            if state.closed {
                return None;
            }
            state = self.ready.wait(state).expect("queue poisoned");
        }
    }

    fn close(&self) {
        let mut state = self.state.lock().expect("queue poisoned");
        state.closed = true;
        self.ready.notify_all();
        self.space.notify_all();
    }
}

fn main() {
    let queue = BoundedQueue::new(2);
    let producer = {
        let queue = Arc::clone(&queue);
        thread::spawn(move || {
            for value in 0..6 {
                queue.push(value);
            }
            queue.close();
        })
    };
    let consumer = {
        let queue = Arc::clone(&queue);
        thread::spawn(move || {
            let mut seen = Vec::new();
            while let Some(value) = queue.pop() {
                seen.push(value);
                thread::sleep(Duration::from_millis(1));
            }
            seen
        })
    };

    producer.join().expect("producer panicked");
    let seen = consumer.join().expect("consumer panicked");
    println!("{seen:?}");
    assert_eq!(seen, vec![0, 1, 2, 3, 4, 5]);
}