sync: Add is_closed method to mpsc senders

tokio/src/sync/mpsc/bounded.rs

@@ -523,6 +523,21 @@ impl<T> Sender<T> {
         enter_handle.block_on(self.send(value)).unwrap()
     }
 
+    /// Checks if `Receiver` is still alive.
+    ///
+    /// ```
+    /// let (tx, rx) = tokio::sync::mpsc::channel::<()>(42);
+    /// assert!(!tx.is_closed());
+    /// let tx2 = tx.clone();
+    /// assert!(!tx2.is_closed());
+    /// std::mem::drop(rx);
+    /// assert!(tx.is_closed());
+    /// assert!(tx2.is_closed());
+    /// ```
+    pub fn is_closed(&self) -> bool {
+        self.chan.is_closed()
+    }
+
     /// Wait for channel capacity. Once capacity to send one message is
     /// available, it is reserved for the caller.
     ///

tokio/src/sync/mpsc/chan.rs

@@ -143,6 +143,10 @@ impl<T, S> Tx<T, S> {
 }
 
 impl<T, S: Semaphore> Tx<T, S> {
+    pub(crate) fn is_closed(&self) -> bool {
+        self.inner.semaphore.is_closed()
+    }
+
     pub(crate) async fn closed(&mut self) {
         use std::future::Future;
         use std::pin::Pin;

tokio/src/sync/mpsc/unbounded.rs

@@ -245,4 +245,18 @@ impl<T> UnboundedSender<T> {
     pub async fn closed(&mut self) {
         self.chan.closed().await
     }
+    /// Checks if `UnboundedReceiver` is still alive.
+    ///
+    /// ```
+    /// let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<()>();
+    /// assert!(!tx.is_closed());
+    /// let tx2 = tx.clone();
+    /// assert!(!tx2.is_closed());
+    /// std::mem::drop(rx);
+    /// assert!(tx.is_closed());
+    /// assert!(tx2.is_closed());
+    /// ```
+    pub fn is_closed(&self) -> bool {
+        self.chan.is_closed()
+    }
 }