update stream::fuse

Signed-off-by: Yoshua Wuyts <[email protected]>

Author: yoshuawuyts

src/stream/stream/fuse.rs

@@ -1,51 +1,29 @@
 use std::pin::Pin;
+use std::task::{Context, Poll};
 
-use crate::task::{Context, Poll};
-
-/// A `Stream` that is permanently closed
-/// once a single call to `poll` results in
-/// `Poll::Ready(None)`, returning `Poll::Ready(None)`
-/// for all future calls to `poll`.
+/// A `Stream` that is permanently closed once a single call to `poll` results in
+/// `Poll::Ready(None)`, returning `Poll::Ready(None)` for all future calls to `poll`.
 #[derive(Clone, Debug)]
 pub struct Fuse<S> {
-    stream: S,
-    done: bool,
+    pub(crate) stream: S,
+    pub(crate) done: bool,
 }
 
 impl<S: Unpin> Unpin for Fuse<S> {}
 
-impl<S: futures::Stream> Fuse<S> {
+impl<S: futures_core::Stream> Fuse<S> {
     pin_utils::unsafe_pinned!(stream: S);
     pin_utils::unsafe_unpinned!(done: bool);
-
-    /// Returns `true` if the underlying stream is fused.
-    ///
-    /// If this `Stream` is fused, all future calls to
-    /// `poll` will return `Poll::Ready(None)`.
-    pub fn is_done(&self) -> bool {
-        self.done
-    }
-
-    /// Consumes this `Fuse` and returns the inner
-    /// `Stream`, unfusing it if it had become
-    /// fused.
-    pub fn into_inner(self) -> S
-    where
-        S: Sized,
-    {
-        self.stream
-    }
 }
 
-
-impl<S: futures::Stream> futures::Stream for Fuse<S> {
+impl<S: futures_core::Stream> futures_core::Stream for Fuse<S> {
     type Item = S::Item;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> {
         if self.done {
             Poll::Ready(None)
         } else {
-            let next = futures::ready!(self.as_mut().stream().poll_next(cx));
+            let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
             if next.is_none() {
                 *self.as_mut().done() = true;
             }

src/stream/stream/mod.rs

@@ -248,29 +248,29 @@ pub trait Stream {
         Enumerate::new(self)
     }
 
-    /// Transforms this `Stream` into a "fused" `Stream`
-    /// such that after the first time `poll` returns
-    /// `Poll::Ready(None)`, all future calls to
-    /// `poll` will also return `Poll::Ready(None)`.
+    /// Transforms this `Stream` into a "fused" `Stream` such that after the first time `poll`
+    /// returns `Poll::Ready(None)`, all future calls to `poll` will also return
+    /// `Poll::Ready(None)`.
     ///
     /// # Examples
     ///
     /// ```
-    /// # #![feature(async_await)]
     /// # fn main() { async_std::task::block_on(async {
     /// #
     /// use async_std::prelude::*;
     /// use async_std::stream;
     ///
-    /// let mut s = stream::repeat(9).take(3);
-    ///
-    /// while let Some(v) = s.next().await {
-    ///     assert_eq!(v, 9);
-    /// }
+    /// let mut s = stream::once(1).fuse();
+    /// assert_eq!(s.next().await, Some(1));
+    /// assert_eq!(s.next().await, None);
+    /// assert_eq!(s.next().await, None);
     /// #
     /// # }) }
     /// ```
-    fn fuse(self) -> Fuse<Self> {
+    fn fuse(self) -> Fuse<Self>
+    where
+        Self: Sized,
+    {
         Fuse {
             stream: self,
             done: false,