Reproduce with slow producer

async-channel-repro.rs

use async_channel::{Receiver, Sender};
use async_executor::Executor;
use async_io::Timer;
use futures_lite::future;

use std::time::Duration;

// [dependencies]
// async-executor = "0.2.1"
// async-channel = "1.4.1"
// async-io = "0.2.6"
// futures-lite = "1"

const CONSUMER_NUMBER: usize = 3;

async fn producer(tx: Sender<usize>) {
    println!("Running producer");
    for ix in 0..10_000 {
        match tx.send(ix).await {
            Ok(_) => {}
            Err(e) => println!("Error while sending {:?}", e),
        }
        Timer::after(Duration::from_millis(30)).await;
    }
}

async fn consumer(rx: Receiver<usize>, id: usize) {
    println!("Running consumer {}", id);
    loop {
        Timer::after(Duration::from_millis(10)).await;
        let recv = rx.recv().await;
        match recv {
            Ok(time) => println!("[{}]Recieved {}", id, time),
            Err(e) => {
                println!("Error {:?}", e);
                return;
            }
        }
    }
}

fn main() {
    let ex = Executor::new();

    let (s, r) = async_channel::unbounded::<usize>();

    let mut tasks = vec![];
    println!("Spawning consumers");
    for id in 0..CONSUMER_NUMBER {
        let task = ex.spawn(consumer(r.clone(), id));
        tasks.push(task);
    }

    println!("Spawning producers");
    let producer = ex.spawn(producer(s));

    future::block_on(ex.run(producer));
    for task in tasks {
        future::block_on(ex.run(task));
    }
  
}