1 files changed, 111 insertions, 8 deletions
diff --git a/tests/sync_mpsc.rs b/tests/sync_mpsc.rs
index cd43ad4..1947d26 100644
--- a/tests/sync_mpsc.rs
+++ b/tests/sync_mpsc.rs
@@ -5,7 +5,7 @@
 use std::thread;
 use tokio::runtime::Runtime;
 use tokio::sync::mpsc;
-use tokio::sync::mpsc::error::TrySendError;
+use tokio::sync::mpsc::error::{TryRecvError, TrySendError};
 use tokio_test::task;
 use tokio_test::{
     assert_err, assert_ok, assert_pending, assert_ready, assert_ready_err, assert_ready_ok,
@@ -328,6 +328,27 @@ async fn try_send_fail() {
 }
 
 #[tokio::test]
+async fn try_send_fail_with_try_recv() {
+    let (tx, mut rx) = mpsc::channel(1);
+
+    tx.try_send("hello").unwrap();
+
+    // This should fail
+    match assert_err!(tx.try_send("fail")) {
+        TrySendError::Full(..) => {}
+        _ => panic!(),
+    }
+
+    assert_eq!(rx.try_recv(), Ok("hello"));
+
+    assert_ok!(tx.try_send("goodbye"));
+    drop(tx);
+
+    assert_eq!(rx.try_recv(), Ok("goodbye"));
+    assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
+}
+
+#[tokio::test]
 async fn try_reserve_fails() {
     let (tx, mut rx) = mpsc::channel(1);
 
@@ -389,13 +410,15 @@ fn dropping_rx_closes_channel_for_try() {
 
     drop(rx);
 
-    {
-        let err = assert_err!(tx.try_send(msg.clone()));
-        match err {
-            TrySendError::Closed(..) => {}
-            _ => panic!(),
-        }
-    }
+    assert!(matches!(
+        tx.try_send(msg.clone()),
+        Err(TrySendError::Closed(_))
+    ));
+    assert!(matches!(tx.try_reserve(), Err(TrySendError::Closed(_))));
+    assert!(matches!(
+        tx.try_reserve_owned(),
+        Err(TrySendError::Closed(_))
+    ));
 
     assert_eq!(1, Arc::strong_count(&msg));
 }
@@ -494,3 +517,83 @@ async fn permit_available_not_acquired_close() {
     drop(permit2);
     assert!(rx.recv().await.is_none());
 }
+
+#[test]
+fn try_recv_bounded() {
+    let (tx, mut rx) = mpsc::channel(5);
+
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    assert!(tx.try_send("hello").is_err());
+
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
+
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    assert_eq!(Ok("hello"), rx.try_recv());
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    assert!(tx.try_send("hello").is_err());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
+
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    tx.try_send("hello").unwrap();
+    drop(tx);
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Ok("hello"), rx.try_recv());
+    assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
+}
+
+#[test]
+fn try_recv_unbounded() {
+    for num in 0..100 {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+
+        for i in 0..num {
+            tx.send(i).unwrap();
+        }
+
+        for i in 0..num {
+            assert_eq!(rx.try_recv(), Ok(i));
+        }
+
+        assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
+        drop(tx);
+        assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
+    }
+}
+
+#[test]
+fn try_recv_close_while_empty_bounded() {
+    let (tx, mut rx) = mpsc::channel::<()>(5);
+
+    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
+    drop(tx);
+    assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
+}
+
+#[test]
+fn try_recv_close_while_empty_unbounded() {
+    let (tx, mut rx) = mpsc::unbounded_channel::<()>();
+
+    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
+    drop(tx);
+    assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
+}