test(mpsc): cover bounded sender wakeups

tests-integration/tests/mpsc_test.rs

@@ -15,15 +15,24 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::task::Poll;
 use std::time::Instant;
 
 use asyncband::mpsc;
 use asyncband::mpsc::RecvError;
 use asyncband::mpsc::TryRecvError;
 use asyncband::mpsc::TrySendError;
+use tests_integration::poll_once;
 use tests_integration::test_runtime;
 use tokio_test::assert_ok;
 
+fn expect_ready<T>(poll: Poll<T>) -> T {
+    match poll {
+        Poll::Ready(value) => value,
+        Poll::Pending => panic!("future should be ready"),
+    }
+}
+
 #[test]
 fn test_unbounded_pressure() {
     let n = 1024 * 1024;
@@ -269,6 +278,69 @@ async fn send_after_close_bounded() {
     assert_eq!(error.into_inner(), 2);
 }
 
+#[test]
+fn bounded_wakes_blocked_senders_one_at_a_time() {
+    let (tx, mut rx) = mpsc::bounded(1);
+    tx.try_send(0).unwrap();
+
+    let first_tx = tx.clone();
+    let second_tx = tx.clone();
+    let mut first = Box::pin(first_tx.send(1));
+    let mut second = Box::pin(second_tx.send(2));
+
+    assert!(poll_once(first.as_mut()).is_pending());
+    assert!(poll_once(second.as_mut()).is_pending());
+
+    assert_eq!(rx.try_recv(), Ok(0));
+    assert_eq!(expect_ready(poll_once(first.as_mut())), Ok(()));
+    assert!(poll_once(second.as_mut()).is_pending());
+
+    assert_eq!(rx.try_recv(), Ok(1));
+    assert_eq!(expect_ready(poll_once(second.as_mut())), Ok(()));
+    assert_eq!(rx.try_recv(), Ok(2));
+}
+
+#[test]
+fn bounded_cancelled_notified_sender_passes_slot_to_next_sender() {
+    let (tx, mut rx) = mpsc::bounded(1);
+    tx.try_send(0).unwrap();
+
+    let first_tx = tx.clone();
+    let second_tx = tx.clone();
+    let mut first = Box::pin(first_tx.send(1));
+    let mut second = Box::pin(second_tx.send(2));
+
+    assert!(poll_once(first.as_mut()).is_pending());
+    assert!(poll_once(second.as_mut()).is_pending());
+
+    assert_eq!(rx.try_recv(), Ok(0));
+    drop(first);
+
+    assert_eq!(expect_ready(poll_once(second.as_mut())), Ok(()));
+    assert_eq!(rx.try_recv(), Ok(2));
+}
+
+#[test]
+fn bounded_receiver_drop_returns_values_to_all_blocked_senders() {
+    let (tx, rx) = mpsc::bounded(1);
+    tx.try_send(0).unwrap();
+
+    let first_tx = tx.clone();
+    let second_tx = tx.clone();
+    let mut first = Box::pin(first_tx.send(1));
+    let mut second = Box::pin(second_tx.send(2));
+
+    assert!(poll_once(first.as_mut()).is_pending());
+    assert!(poll_once(second.as_mut()).is_pending());
+
+    drop(rx);
+
+    let first_error = expect_ready(poll_once(first.as_mut())).unwrap_err();
+    let second_error = expect_ready(poll_once(second.as_mut())).unwrap_err();
+    assert_eq!(first_error.into_inner(), 1);
+    assert_eq!(second_error.into_inner(), 2);
+}
+
 #[test]
 fn test_bounded_pressure() {
     let n = 1024 * 1024;