tokio-stream: implement FusedStream for Fuse (#8090)

tokio-stream/src/stream_ext/fuse.rs

@@ -1,5 +1,6 @@
 use crate::Stream;
 
+use futures_core::FusedStream;
 use pin_project_lite::pin_project;
 use std::pin::Pin;
 use std::task::{ready, Context, Poll};
@@ -51,3 +52,12 @@ where
         }
     }
 }
+
+impl<T> FusedStream for Fuse<T>
+where
+    T: Stream,
+{
+    fn is_terminated(&self) -> bool {
+        self.stream.is_none()
+    }
+}

tokio-stream/tests/stream_fuse.rs

@@ -1,3 +1,4 @@
+use futures_core::FusedStream;
 use tokio_stream::{Stream, StreamExt};
 
 use std::pin::Pin;
@@ -37,16 +38,22 @@ async fn basic_usage() {
     // however, once it is fused
     let mut stream = stream.fuse();
 
+    assert!(!stream.is_terminated());
     assert_eq!(stream.size_hint(), (0, None));
     assert_eq!(stream.next().await, Some(4));
 
+    assert!(!stream.is_terminated());
     assert_eq!(stream.size_hint(), (0, None));
     assert_eq!(stream.next().await, None);
 
+    assert!(stream.is_terminated());
+
     // it will always return `None` after the first time.
     assert_eq!(stream.size_hint(), (0, Some(0)));
     assert_eq!(stream.next().await, None);
     assert_eq!(stream.size_hint(), (0, Some(0)));
+
+    assert!(stream.is_terminated());
 }
 
 #[tokio::test]