diff --git a/channeltest/src/main.rs b/channeltest/src/main.rs index 1db9515..24acf78 100644 --- a/channeltest/src/main.rs +++ b/channeltest/src/main.rs @@ -3,7 +3,7 @@ use std::time::Duration; use tokio::{sync::mpsc, time::sleep, task}; #[tokio::main(flavor = "current_thread")] -async fn main() { +async fn do_tasks_complete_when_one_side_of_the_channel_closes() { let (tx, mut rx) = mpsc::channel(10); let producer = task::spawn(async move { @@ -25,3 +25,43 @@ async fn main() { producer.await.expect("Producer died"); consumer.await.expect("Consumer died"); } + +#[tokio::main(flavor = "current_thread")] +async fn main() { + let (tx1, mut rx) = mpsc::channel(10); + + let tx2 = tx1.clone(); + let producer1 = task::spawn(async move { + for i in 10..20 { + println!("1: Produced {i}"); + tx1.send(i).await.expect("Failed to send t1"); + sleep(Duration::from_millis(i * 123)).await; + } + println!("==== P1 completed"); + }); + + let producer2 = task::spawn(async move { + for i in 0..10 { + println!("2: Produced {i}"); + tx2.send(i).await.expect("Failed to send t2"); + sleep(Duration::from_millis(i * 231)).await; + } + println!("==== P2 completed"); + }); + + let consumer = task::spawn(async move { + let mut data = Vec::new(); + while let Some(i) = rx.recv().await { + println!("Received {i}"); + data.push(i); + } + + println!("**** Channels closed, processing results"); + println!("{:?}", data); + }); + + + producer1.await.expect("P1 died"); + producer2.await.expect("P2 died"); + consumer.await.expect("Consumer died"); +}