pub struct SemaPlusPlus<T> {
    queue_and_cv: Arc<(Mutex<VecDeque<T>>, Condvar)>,
}

impl<T> SemaPlusPlus<T> {
    pub fn send(&self, message: T) {
        let (queue_lock, cv) = &*self.queue_and_cv;
        let mut queue = queue_lock.lock().unwrap();
        let queue_was_empty = queue.is_empty();
        queue.push_back(message);
        // if a queue is not empty, then another thread must have notified and main()
        // must be awake
        if queue_was_empty {
            cv.notify_all();
        }
    }

    pub fn recv(&self) -> T {
        let (queue_lock, cv) = &*self.queue_and_cv;
        // wait_while dereferences the guard and pass the protected data to the closure,
        // when the condition is true, it releases the lock and sleep until another thread
        // notify_all
        let mut queue = cv
            .wait_while(queue_lock.lock().unwrap(), |queue| queue.is_empty())
            .unwrap();
        queue.pop_front().unwrap()
    }
}

fn main() {
    let sem: SemaPlusPlus<String> = SemaPlusPlus::new();
    let mut handlers = Vec::new();
    for i in 0..NUM_THREADS {
        let sem_clone = sem.clone();
        let handle = thread::spawn(move || {
            rand_sleep();
            sem_clone.send(format!("thread {} just finished!", i))
        });
        handlers.push(handle);
    }
    // main is the only consumer of the queue,
    // it wakes up when some thread notify_all, locks the queue, consume the message
    // and release the lock until it is notified again
    for _ in 0..NUM_THREADS {
        println!("{}", sem.recv())
    }

    for handle in handlers {
        handle.join().unwrap();
    }
}

/// Factor the number while receiving numbers from stdin
fn main() {
    let (sender, receiver) = crossbeam_channel::unbounded();
    let mut threads = Vec::new();
    for _ in 0..num_cpus::get() {
        // The channel maintains a reference counter for senders and for receivers
        // when receiver.clone(), the receiver counter increments.
        let receiver = receiver.clone();
        threads.push(thread::spawn(move || {
            while let Ok(next_num) = receiver.recv() {
                factor_number(next_num);
            }
        }));
    }

    let stdin = std::io::stdin();
    for line in stdin.lock().lines() {
        let num = line.unwrap().parse::<u32>().unwrap();
        sender.send(num).expect("Cannot send to channel when there is no receiver");
    }

    // decrement the sender counter, when no sender, the channel is closed
    drop(sender);

    for thread in threads {
        thread.join().expect("panic");
    }
}

/// A simplified version of Future definition
trait Future {
    type Output;
    // the executor thread calls poll() to start running until it cannot progress
    // cx includes a wake() function so that the future wakes up the executor when it can progress again
    // when wake() is called, the executor uses cx to identify the future that can make new progress
    fn poll(&mut self, cx: &mut Context) -> Poll<Self::Output>;
}

enum Poll<T> {
    Ready(T),
    /// Allow the single thread to work on other future
    Pending,
}

  // This is just an example, real futures are not composed like this
  let future = PlaceOnStoveFuture::new(meat)
    .then(|meat| CookOnSideFuture::new(meat))
    .then(|meat| FlipFuture::new(meat))
    .then(|meat| CookOnSideFuture::new(meat));

let cook_meal_future = futures::future::join(
    vec![
        CookMeatFuture::new(),
        CookMeatFuture::new(),
        // future may return Pending after complete 2 futures,
        // the next time when the executor calls poll(), the state is restored
        CookSoupFuture::new(),
        BakeBreadFuture::new()
    ]
);