Repository: tokio-rs/tokio
File: tokio/tests/sync_mpsc.rs (Rust)

Code before:
    .await;

    let _ = actor_handle.await;
}

static NUM_DROPPED: AtomicUsize = AtomicUsize::new(0);

#[derive(Debug)]
struct Msg;

impl Drop for Msg {
    fn drop(&mut self) {
        NUM_DROPPED.fetch_add(1, Release);
    }
}

// Tests that no pending messages are put onto the channel after `Rx` was
// dropped.
//
// Note: After the introduction of `WeakSender`, which internally
// used `Arc` and doesn't call a drop of the channel after the last strong
// `Sender` was dropped while more than one `WeakSender` remains, we want to
// ensure that no messages are kept in the channel, which were sent after
// the receiver was dropped.
#[tokio::test(start_paused = true)]
async fn test_msgs_dropped_on_rx_drop() {
    fn ms(millis: u64) -> Duration {
        Duration::from_millis(millis)
    }

    let (tx, mut rx) = mpsc::channel(3);

    let rx_handle = tokio::spawn(async move {
        let _ = rx.recv().await.unwrap();
        let _ = rx.recv().await.unwrap();
        time::sleep(ms(1)).await;
        drop(rx);

        time::advance(ms(1)).await;
    });

    let tx_handle = tokio::spawn(async move {
        let _ = tx.send(Msg {}).await.unwrap();
        let _ = tx.send(Msg {}).await.unwrap();

        // This msg will be pending and should be dropped when `rx` is dropped
        let _ = tx.send(Msg {}).await.unwrap();
        time::advance(ms(1)).await;

        // This msg will not be put onto `Tx` list anymore, since `Rx` is closed.
        time::sleep(ms(1)).await;

Review comment (bug):
One of the `unwrap` calls in test this fails on wasm. I have not investigated why, but you might want to change them to `expect` with a message so you can tell which one failed.

Code after:
static NUM_DROPPED: AtomicUsize = AtomicUsize::new(0);

#[derive(Debug)]
struct Msg;

impl Drop for Msg {
    fn drop(&mut self) {
        NUM_DROPPED.fetch_add(1, Release);
    }
}

// Tests that no pending messages are put onto the channel after `Rx` was
// dropped.
//
// Note: After the introduction of `WeakSender`, which internally
// used `Arc` and doesn't call a drop of the channel after the last strong
// `Sender` was dropped while more than one `WeakSender` remains, we want to
// ensure that no messages are kept in the channel, which were sent after
// the receiver was dropped.
#[tokio::test]
async fn test_msgs_dropped_on_rx_drop() {
    let (tx, mut rx) = mpsc::channel(3);

    let _ = tx.send(Msg {}).await.unwrap();
    let _ = tx.send(Msg {}).await.unwrap();

    // This msg will be pending and should be dropped when `rx` is dropped
    let sent_fut = tx.send(Msg {});

    let _ = rx.recv().await.unwrap();
    let _ = rx.recv().await.unwrap();

    let _ = sent_fut.await.unwrap();

    drop(rx);

    assert_eq!(NUM_DROPPED.load(Acquire), 3);

    // This msg will not be put onto `Tx` list anymore, since `Rx` is closed.
    assert!(tx.send(Msg {}).await.is_err());

    assert_eq!(NUM_DROPPED.load(Acquire), 4);
}

// Tests that a `WeakSender` is upgradeable when other `Sender`s exist.
#[tokio::test]
async fn downgrade_upgrade_sender_success() {
    let (tx, _rx) = mpsc::channel::<i32>(1);
    let weak_tx = tx.downgrade();
    assert!(weak_tx.upgrade().is_some());
}