Implement FusedStream for a few types

tokio/src/sync/mpsc/bounded.rs

@@ -177,6 +177,12 @@ impl<T> Receiver<T> {
 impl<T> Unpin for Receiver<T> {}
 
 cfg_stream! {
+    impl<T> futures_core::FusedStream for Receiver<T> {
+        fn is_terminated(&self) -> bool {
+            self.chan.is_closed()
+        }
+    }
+
     impl<T> crate::stream::Stream for Receiver<T> {
         type Item = T;
 

tokio/src/sync/mpsc/chan.rs

@@ -322,6 +322,15 @@ where
             }
         })
     }
+
+    /// Test if this receiver side of the channel is closed.
+    #[cfg(feature = "stream")]
+    pub(crate) fn is_closed(&self) -> bool {
+        self.inner
+            .rx_fields
+            .with(|rx_fields_ptr| unsafe { (*rx_fields_ptr).rx_closed })
+            && self.inner.semaphore.is_idle()
+    }
 }
 
 impl<T, S> Drop for Rx<T, S>

tokio/src/time/interval.rs

@@ -129,11 +129,19 @@ impl Interval {
     }
 }
 
-#[cfg(feature = "stream")]
-impl crate::stream::Stream for Interval {
-    type Item = Instant;
+cfg_stream! {
+    impl futures_core::FusedStream for Interval {
+        fn is_terminated(&self) -> bool {
+            // NB: intervals never terminate.
+            false
+        }
+    }
+
+    impl crate::stream::Stream for Interval {
+        type Item = Instant;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Instant>> {
-        Poll::Ready(Some(ready!(self.poll_tick(cx))))
+        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Instant>> {
+            Poll::Ready(Some(ready!(self.poll_tick(cx))))
+        }
     }
 }