From 247e1f083fff90ad09d433425de5e5d85656fb94 Mon Sep 17 00:00:00 2001
From: thement <40525767+thement@users.noreply.github.com>
Date: Tue, 25 Jul 2023 14:40:09 +0200
Subject: [PATCH] Add `len`, `is_empty` methods for `UnboundedSender` (#2750)

- add `len`, `is_empty` methods to inspect how many messages are
  enqueued in the message queue.
- add test for `len` and `is_empty`

Co-authored-by: Jakub Horak <jakub.horak@ibawizard.net>
---
 futures-channel/src/mpsc/mod.rs | 14 ++++++++++++++
 futures-channel/tests/mpsc.rs   | 23 +++++++++++++++++++++++
 2 files changed, 37 insertions(+)

diff --git a/futures-channel/src/mpsc/mod.rs b/futures-channel/src/mpsc/mod.rs
index cf45fe77f..ddf234042 100644
--- a/futures-channel/src/mpsc/mod.rs
+++ b/futures-channel/src/mpsc/mod.rs
@@ -842,6 +842,20 @@ impl<T> UnboundedSender<T> {
         let ptr = self.0.as_ref().map(|inner| inner.ptr());
         ptr.hash(hasher);
     }
+
+    /// Return the number of messages in the queue or 0 if channel is disconnected.
+    pub fn len(&self) -> usize {
+        if let Some(sender) = &self.0 {
+            decode_state(sender.inner.state.load(SeqCst)).num_messages
+        } else {
+            0
+        }
+    }
+
+    /// Return false is channel has no queued messages, true otherwise.
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
 }
 
 impl<T> Clone for Sender<T> {
diff --git a/futures-channel/tests/mpsc.rs b/futures-channel/tests/mpsc.rs
index 444c8e10f..f5d7198d2 100644
--- a/futures-channel/tests/mpsc.rs
+++ b/futures-channel/tests/mpsc.rs
@@ -632,3 +632,26 @@ fn send_backpressure_multi_senders() {
     let item = block_on(rx.next()).unwrap();
     assert_eq!(item, 2);
 }
+
+/// Test that empty channel has zero length and that non-empty channel has length equal to number
+/// of enqueued items
+#[test]
+fn unbounded_len() {
+    let (tx, mut rx) = mpsc::unbounded();
+    assert_eq!(tx.len(), 0);
+    assert!(tx.is_empty());
+    tx.unbounded_send(1).unwrap();
+    assert_eq!(tx.len(), 1);
+    assert!(!tx.is_empty());
+    tx.unbounded_send(2).unwrap();
+    assert_eq!(tx.len(), 2);
+    assert!(!tx.is_empty());
+    let item = block_on(rx.next()).unwrap();
+    assert_eq!(item, 1);
+    assert_eq!(tx.len(), 1);
+    assert!(!tx.is_empty());
+    let item = block_on(rx.next()).unwrap();
+    assert_eq!(item, 2);
+    assert_eq!(tx.len(), 0);
+    assert!(tx.is_empty());
+}