sync: add buffer_len method for Sender in bounded

Fixes: tokio-rs#2642

tokio/src/sync/mpsc/bounded.rs

@@ -546,4 +546,8 @@ impl<T> Sender<T> {
             false
         }
     }
+    /// TODO: Add docs
+    pub fn buffer_len(&self) -> usize {
+        self.chan.available_permits()
+    }
 }

tokio/src/sync/mpsc/chan.rs

@@ -96,6 +96,8 @@ pub(crate) trait Semaphore {
     fn forget(&self, permit: &mut Self::Permit);
 
     fn close(&self);
+
+    fn available_permits(&self) -> usize;
 }
 
 struct Chan<T, S> {
@@ -199,6 +201,10 @@ where
     pub(crate) fn try_send(&mut self, value: T) -> Result<(), (T, TrySendError)> {
         self.inner.try_send(value, &mut self.permit)
     }
+
+    pub(crate) fn available_permits(&self) -> usize {
+        self.inner.semaphore.available_permits()
+    }
 }
 
 impl<T> Tx<T, (crate::sync::semaphore_ll::Semaphore, usize)> {
@@ -465,6 +471,10 @@ impl Semaphore for (crate::sync::semaphore_ll::Semaphore, usize) {
     fn close(&self) {
         self.0.close();
     }
+
+    fn available_permits(&self) -> usize {
+        self.0.available_permits()
+    }
 }
 
 // ===== impl Semaphore for AtomicUsize =====
@@ -528,4 +538,8 @@ impl Semaphore for AtomicUsize {
     fn close(&self) {
         self.fetch_or(1, Release);
     }
+
+    fn available_permits(&self) -> usize {
+        self.load(Relaxed)
+    }
 }

tokio/tests/sync_mpsc.rs

@@ -534,3 +534,12 @@ async fn blocking_send_async() {
     let (mut tx, _rx) = mpsc::channel::<()>(1);
     let _ = tx.blocking_send(());
 }
+
+#[tokio::test]
+async fn buffer_len_bounded() {
+    let buffer_size = 10;
+    let (mut tx, _rx) = mpsc::channel(buffer_size);
+    tx.try_send(0).unwrap();
+    tx.try_send(0).unwrap();
+    assert_eq!(tx.buffer_len(), buffer_size - 2);
+}