fancellu · February 27, 2024 13:26 · fancellu · Feb 27, 2024
diff --git a/tokio_channels_demo.rs b/tokio_channels_demo.rs
 use tokio::sync::broadcast;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;

 #[tokio::main]
 async fn main() {
    let (tx, rx) = oneshot::channel();

    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        if let Err(_) = tx.send(99) {
            println!("the receiver dropped");
        }
    });

    println!("Waiting on rx");

    match rx.await {
        Ok(v) => println!("got oneshot = {:?}", v),
        Err(_) => println!("the sender dropped"),
    }

    // Buffer of 3 will create back pressure when full
    let (tx, mut rx) = mpsc::channel(3);

    tokio::spawn(async move {
        for i in 0..10 {
            println!("Sending {}", i);
            if let Err(_) = tx.send(i).await {
                println!("receiver dropped");
                return;
            }
        }
    });

    while let Some(i) = rx.recv().await {
        // We read slowly so that we can see sender block when buffer is full
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        println!("got mpsc = {}", i);
    }

    // Creating a broadcast channel for strings, multiple producer and consumers
    let (tx, mut _rx) = broadcast::channel::<String>(3);
    let tx1 = tx.clone();
    let tx2 = tx.clone();
    tokio::spawn(async move {
        tx1.send("hello".to_string()).unwrap();
        tx2.send("world".to_string()).unwrap();
    });

    let mut rx1 = tx.subscribe();
    let mut rx2 = tx.subscribe();

    // We see the sent messages
    for _i in 0..2 {
        let msg = rx1.recv().await.unwrap();
        println!("Got rx1 {}", msg);
    }
    // Sent messages are seen again, as we use a different receiver
    for _i in 0..2 {
        let msg = rx2.recv().await.unwrap();
        println!("Got rx2 {}", msg);
    }
 }
	use tokio::sync::broadcast;
	use tokio::sync::mpsc;
	use tokio::sync::oneshot;

	#[tokio::main]
	async fn main() {
	let (tx, rx) = oneshot::channel();

	tokio::spawn(async move {
	tokio::time::sleep(std::time::Duration::from_secs(1)).await;
	if let Err(_) = tx.send(99) {
	println!("the receiver dropped");
	}
	});

	println!("Waiting on rx");

	match rx.await {
	Ok(v) => println!("got oneshot = {:?}", v),
	Err(_) => println!("the sender dropped"),
	}

	// Buffer of 3 will create back pressure when full
	let (tx, mut rx) = mpsc::channel(3);

	tokio::spawn(async move {
	for i in 0..10 {
	println!("Sending {}", i);
	if let Err(_) = tx.send(i).await {
	println!("receiver dropped");
	return;
	}
	}
	});

	while let Some(i) = rx.recv().await {
	// We read slowly so that we can see sender block when buffer is full
	tokio::time::sleep(std::time::Duration::from_millis(500)).await;
	println!("got mpsc = {}", i);
	}

	// Creating a broadcast channel for strings, multiple producer and consumers
	let (tx, mut _rx) = broadcast::channel::<String>(3);
	let tx1 = tx.clone();
	let tx2 = tx.clone();
	tokio::spawn(async move {
	tx1.send("hello".to_string()).unwrap();
	tx2.send("world".to_string()).unwrap();
	});

	let mut rx1 = tx.subscribe();
	let mut rx2 = tx.subscribe();

	// We see the sent messages
	for _i in 0..2 {
	let msg = rx1.recv().await.unwrap();
	println!("Got rx1 {}", msg);
	}
	// Sent messages are seen again, as we use a different receiver
	for _i in 0..2 {
	let msg = rx2.recv().await.unwrap();
	println!("Got rx2 {}", msg);
	}
	}