diff --git a/tokio/src/sync/mpsc/chan.rs b/tokio/src/sync/mpsc/chan.rs index 7f7f8b16d..f9a72de60 100644 --- a/tokio/src/sync/mpsc/chan.rs +++ b/tokio/src/sync/mpsc/chan.rs @@ -436,7 +436,9 @@ impl Rx { } TryPopResult::Closed => return Err(TryRecvError::Disconnected), // If close() was called, an empty queue should report Disconnected. - TryPopResult::Empty if rx_fields.rx_closed => { + TryPopResult::Empty + if rx_fields.rx_closed && self.inner.semaphore.is_idle() => + { return Err(TryRecvError::Disconnected) } TryPopResult::Empty => return Err(TryRecvError::Empty), diff --git a/tokio/tests/sync_mpsc.rs b/tokio/tests/sync_mpsc.rs index 048b94eb5..27157288e 100644 --- a/tokio/tests/sync_mpsc.rs +++ b/tokio/tests/sync_mpsc.rs @@ -999,6 +999,19 @@ fn try_recv_after_receiver_close() { assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv()); } +#[test] +fn try_recv_after_receiver_close_with_permit() { + let (tx, mut rx) = mpsc::channel::<()>(5); + + let permit = tx.try_reserve().unwrap(); + + assert_eq!(Err(TryRecvError::Empty), rx.try_recv()); + rx.close(); + assert_eq!(Err(TryRecvError::Empty), rx.try_recv()); + drop(permit); + assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv()); +} + #[test] fn try_recv_close_while_empty_bounded() { let (tx, mut rx) = mpsc::channel::<()>(5);