Merge pull request #506 from stjepang/cleanup-stream

Cleanup stream module

src/io/mod.rs

@@ -19,8 +19,8 @@
 //! [`File`]s:
 //!
 //! ```no_run
-//! use async_std::prelude::*;
 //! use async_std::fs::File;
+//! use async_std::prelude::*;
 //!
 //! # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
 //! #
@@ -47,9 +47,9 @@
 //! coming from:
 //!
 //! ```no_run
-//! use async_std::io::prelude::*;
-//! use async_std::io::SeekFrom;
 //! use async_std::fs::File;
+//! use async_std::io::SeekFrom;
+//! use async_std::prelude::*;
 //!
 //! # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
 //! #
@@ -82,9 +82,9 @@
 //! methods to any reader:
 //!
 //! ```no_run
-//! use async_std::io::prelude::*;
-//! use async_std::io::BufReader;
 //! use async_std::fs::File;
+//! use async_std::io::BufReader;
+//! use async_std::prelude::*;
 //!
 //! # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
 //! #
@@ -104,9 +104,9 @@
 //! to [`write`][`Write::write`]:
 //!
 //! ```no_run
-//! use async_std::io::prelude::*;
-//! use async_std::io::BufWriter;
 //! use async_std::fs::File;
+//! use async_std::io::BufWriter;
+//! use async_std::io::prelude::*;
 //!
 //! # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
 //! #
@@ -179,9 +179,9 @@
 //! lines:
 //!
 //! ```no_run
-//! use async_std::prelude::*;
-//! use async_std::io::BufReader;
 //! use async_std::fs::File;
+//! use async_std::io::BufReader;
+//! use async_std::prelude::*;
 //!
 //! # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
 //! #

src/stream/extend.rs

@@ -65,10 +65,10 @@ pub trait Extend<A> {
 /// ```
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-pub async fn extend<'a, C, A, T>(collection: &mut C, stream: T)
+pub async fn extend<'a, C, T, S>(collection: &mut C, stream: S)
 where
-    C: Extend<A>,
-    T: IntoStream<Item = A> + 'a,
+    C: Extend<T>,
+    S: IntoStream<Item = T> + 'a,
 {
     Extend::extend(collection, stream).await
 }

src/stream/from_fn.rs

@@ -1,28 +1,21 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
-use std::future::Future;
-
-use pin_project_lite::pin_project;
 
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
-pin_project! {
-    /// A stream that yields elements by calling a closure.
-    ///
-    /// This stream is created by the [`from_fn`] function. See its
-    /// documentation for more.
-    ///
-    /// [`from_fn`]: fn.from_fn.html
-    #[derive(Debug)]
-    pub struct FromFn<F, Fut, T> {
-        f: F,
-        #[pin]
-        future: Option<Fut>,
-        __t: PhantomData<T>,
-    }
+/// A stream that yields elements by calling a closure.
+///
+/// This stream is created by the [`from_fn`] function. See its
+/// documentation for more.
+///
+/// [`from_fn`]: fn.from_fn.html
+#[derive(Clone, Debug)]
+pub struct FromFn<F> {
+    f: F,
 }
 
+impl<F> Unpin for FromFn<F> {}
+
 /// Creates a new stream where to produce each new element a provided closure is called.
 ///
 /// This allows creating a custom stream with any behaviour without using the more verbose
@@ -34,21 +27,15 @@ pin_project! {
 /// # async_std::task::block_on(async {
 /// #
 /// use async_std::prelude::*;
-/// use async_std::sync::{Arc, Mutex};
 /// use async_std::stream;
 ///
-/// let count = Arc::new(Mutex::new(0u8));
+/// let mut count = 0u8;
 /// let s = stream::from_fn(|| {
-///     let count = Arc::clone(&count);
-///
-///     async move {
-///         *count.lock().await += 1;
-///
-///         if *count.lock().await > 3 {
-///             None
-///         } else {
-///             Some(*count.lock().await)
-///         }
+///     count += 1;
+///     if count > 3 {
+///         None
+///     } else {
+///         Some(count)
 ///     }
 /// });
 ///
@@ -60,38 +47,21 @@ pin_project! {
 /// #
 /// # })
 /// ```
-pub fn from_fn<T, F, Fut>(f: F) -> FromFn<F, Fut, T>
+pub fn from_fn<T, F>(f: F) -> FromFn<F>
 where
-    F: FnMut() -> Fut,
-    Fut: Future<Output = Option<T>>,
+    F: FnMut() -> Option<T>,
 {
-    FromFn {
-        f,
-        future: None,
-        __t: PhantomData,
-    }
+    FromFn { f }
 }
 
-impl<F, Fut, T> Stream for FromFn<F, Fut, T>
+impl<T, F> Stream for FromFn<F>
 where
-    F: FnMut() -> Fut,
-    Fut: Future<Output = Option<T>>,
+    F: FnMut() -> Option<T>,
 {
     type Item = T;
 
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let mut this = self.project();
-        loop {
-            if this.future.is_some() {
-                let next =
-                    futures_core::ready!(this.future.as_mut().as_pin_mut().unwrap().poll(cx));
-                this.future.set(None);
-
-                return Poll::Ready(next);
-            } else {
-                let fut = (this.f)();
-                this.future.set(Some(fut));
-            }
-        }
+    fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let item = (&mut self.f)();
+        Poll::Ready(item)
     }
 }

src/stream/from_iter.rs

@@ -6,7 +6,7 @@ use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
 pin_project! {
-    /// A stream that created from iterator
+    /// A stream that was created from iterator.
     ///
     /// This stream is created by the [`from_iter`] function.
     /// See it documentation for more.

src/stream/mod.rs

@@ -306,9 +306,7 @@ pub use from_iter::{from_iter, FromIter};
 pub use once::{once, Once};
 pub use repeat::{repeat, Repeat};
 pub use repeat_with::{repeat_with, RepeatWith};
-pub use stream::{
-    Chain, Filter, Fuse, Inspect, Scan, Skip, SkipWhile, StepBy, Stream, Take, TakeWhile, Zip,
-};
+pub use stream::*;
 
 pub(crate) mod stream;
 

src/stream/once.rs

@@ -33,7 +33,7 @@ pin_project! {
     /// documentation for more.
     ///
     /// [`once`]: fn.once.html
-    #[derive(Debug)]
+    #[derive(Clone, Debug)]
     pub struct Once<T> {
         value: Option<T>,
     }

src/stream/repeat.rs

@@ -33,7 +33,7 @@ where
 /// documentation for more.
 ///
 /// [`repeat`]: fn.repeat.html
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct Repeat<T> {
     item: T,
 }

src/stream/repeat_with.rs

@@ -1,28 +1,21 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
-use std::future::Future;
-
-use pin_project_lite::pin_project;
 
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
-pin_project! {
-    /// A stream that repeats elements of type `T` endlessly by applying a provided closure.
-    ///
-    /// This stream is created by the [`repeat_with`] function. See its
-    /// documentation for more.
-    ///
-    /// [`repeat_with`]: fn.repeat_with.html
-    #[derive(Debug)]
-    pub struct RepeatWith<F, Fut, A> {
-        f: F,
-        #[pin]
-        future: Option<Fut>,
-        __a: PhantomData<A>,
-    }
+/// A stream that repeats elements of type `T` endlessly by applying a provided closure.
+///
+/// This stream is created by the [`repeat_with`] function. See its
+/// documentation for more.
+///
+/// [`repeat_with`]: fn.repeat_with.html
+#[derive(Clone, Debug)]
+pub struct RepeatWith<F> {
+    f: F,
 }
 
+impl<F> Unpin for RepeatWith<F> {}
+
 /// Creates a new stream that repeats elements of type `A` endlessly by applying the provided closure.
 ///
 /// # Examples
@@ -35,7 +28,7 @@ pin_project! {
 /// use async_std::prelude::*;
 /// use async_std::stream;
 ///
-/// let s = stream::repeat_with(|| async { 1 });
+/// let s = stream::repeat_with(|| 1);
 ///
 /// pin_utils::pin_mut!(s);
 ///
@@ -54,48 +47,38 @@ pin_project! {
 /// use async_std::prelude::*;
 /// use async_std::stream;
 ///
-/// let s = stream::repeat_with(|| async { 1u8 }).take(2);
+/// let mut n = 1;
+/// let s = stream::repeat_with(|| {
+///     let item = n;
+///     n *= 2;
+///     item
+/// })
+/// .take(4);
 ///
 /// pin_utils::pin_mut!(s);
 ///
 /// assert_eq!(s.next().await, Some(1));
-/// assert_eq!(s.next().await, Some(1));
+/// assert_eq!(s.next().await, Some(2));
+/// assert_eq!(s.next().await, Some(4));
+/// assert_eq!(s.next().await, Some(8));
 /// assert_eq!(s.next().await, None);
 /// # })
 /// ```
-pub fn repeat_with<F, Fut, A>(repeater: F) -> RepeatWith<F, Fut, A>
+pub fn repeat_with<T, F>(repeater: F) -> RepeatWith<F>
 where
-    F: FnMut() -> Fut,
-    Fut: Future<Output = A>,
+    F: FnMut() -> T,
 {
-    RepeatWith {
-        f: repeater,
-        future: None,
-        __a: PhantomData,
-    }
+    RepeatWith { f: repeater }
 }
 
-impl<F, Fut, A> Stream for RepeatWith<F, Fut, A>
+impl<T, F> Stream for RepeatWith<F>
 where
-    F: FnMut() -> Fut,
-    Fut: Future<Output = A>,
+    F: FnMut() -> T,
 {
-    type Item = A;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let mut this = self.project();
-        loop {
-            if this.future.is_some() {
-                let res = futures_core::ready!(this.future.as_mut().as_pin_mut().unwrap().poll(cx));
-
-                this.future.set(None);
-
-                return Poll::Ready(Some(res));
-            } else {
-                let fut = (this.f)();
+    type Item = T;
 
-                this.future.set(Some(fut));
-            }
-        }
+    fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let item = (&mut self.f)();
+        Poll::Ready(Some(item))
     }
 }

src/stream/stream/chain.rs

@@ -7,7 +7,7 @@ use crate::prelude::*;
 use crate::task::{Context, Poll};
 
 pin_project! {
-    /// Chains two streams one after another.
+    /// A stream that chains two streams one after another.
     ///
     /// This `struct` is created by the [`chain`] method on [`Stream`]. See its
     /// documentation for more.

src/stream/stream/cloned.rs

@@ -4,6 +4,7 @@ use pin_project_lite::pin_project;
 use std::pin::Pin;
 
 pin_project! {
+    /// A stream that clones the elements of an underlying stream.
     #[derive(Debug)]
     pub struct Cloned<S> {
         #[pin]

src/stream/stream/copied.rs

@@ -4,8 +4,8 @@ use pin_project_lite::pin_project;
 use std::pin::Pin;
 
 pin_project! {
-    #[doc(hidden)]
-    #[allow(missing_debug_implementations)]
+    /// A stream that copies the elements of an underlying stream.
+    #[derive(Debug)]
     pub struct Copied<S> {
         #[pin]
         stream: S,

src/stream/stream/cycle.rs

@@ -1,73 +1,54 @@
+use std::mem::ManuallyDrop;
 use std::pin::Pin;
 
-use pin_project_lite::pin_project;
-
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
-pin_project! {
-    /// A stream that will repeatedly yield the same list of elements
-    pub struct Cycle<S, T> {
-        #[pin]
-        source: S,
-        index: usize,
-        buffer: Vec<T>,
-        state: CycleState,
-    }
-}
-
-#[derive(Eq, PartialEq)]
-enum CycleState {
-    FromStream,
-    FromBuffer,
+/// A stream that will repeatedly yield the same list of elements.
+#[derive(Debug)]
+pub struct Cycle<S> {
+    orig: S,
+    source: ManuallyDrop<S>,
 }
 
-impl<S> Cycle<S, S::Item>
+impl<S> Cycle<S>
 where
-    S: Stream,
-    S::Item: Clone,
+    S: Stream + Clone,
 {
-    pub fn new(source: S) -> Cycle<S, S::Item> {
+    pub fn new(source: S) -> Cycle<S> {
         Cycle {
-            source,
-            index: 0,
-            buffer: Vec::new(),
-            state: CycleState::FromStream,
+            orig: source.clone(),
+            source: ManuallyDrop::new(source),
         }
     }
 }
 
-impl<S> Stream for Cycle<S, S::Item>
+impl<S> Drop for Cycle<S> {
+    fn drop(&mut self) {
+        unsafe {
+            ManuallyDrop::drop(&mut self.source);
+        }
+    }
+}
+
+impl<S> Stream for Cycle<S>
 where
-    S: Stream,
-    S::Item: Clone,
+    S: Stream + Clone,
 {
     type Item = S::Item;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let this = self.project();
-
-        let mut next;
-        if *this.state == CycleState::FromStream {
-            next = futures_core::ready!(this.source.poll_next(cx));
-
-            if let Some(val) = next {
-                this.buffer.push(val.clone());
-                next = Some(val)
-            } else {
-                *this.state = CycleState::FromBuffer;
-                next = this.buffer.get(*this.index).cloned();
+        unsafe {
+            let this = self.get_unchecked_mut();
+
+            match futures_core::ready!(Pin::new_unchecked(&mut *this.source).poll_next(cx)) {
+                Some(item) => Poll::Ready(Some(item)),
+                None => {
+                    ManuallyDrop::drop(&mut this.source);
+                    this.source = ManuallyDrop::new(this.orig.clone());
+                    Pin::new_unchecked(&mut *this.source).poll_next(cx)
+                }
             }
-        } else {
-            let mut index = *this.index;
-            if index == this.buffer.len() {
-                index = 0
-            }
-            next = Some(this.buffer[index].clone());
-
-            *this.index = index + 1;
         }
-
-        Poll::Ready(next)
     }
 }

src/stream/stream/enumerate.rs

@@ -6,8 +6,7 @@ use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
 pin_project! {
-    #[doc(hidden)]
-    #[allow(missing_debug_implementations)]
+    #[derive(Debug)]
     pub struct Enumerate<S> {
         #[pin]
         stream: S,

src/stream/stream/filter.rs

@@ -1,4 +1,3 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
 
 use pin_project_lite::pin_project;
@@ -15,25 +14,23 @@ pin_project! {
     /// [`filter`]: trait.Stream.html#method.filter
     /// [`Stream`]: trait.Stream.html
     #[derive(Debug)]
-    pub struct Filter<S, P, T> {
+    pub struct Filter<S, P> {
         #[pin]
         stream: S,
         predicate: P,
-        __t: PhantomData<T>,
     }
 }
 
-impl<S, P, T> Filter<S, P, T> {
+impl<S, P> Filter<S, P> {
     pub(super) fn new(stream: S, predicate: P) -> Self {
         Filter {
             stream,
             predicate,
-            __t: PhantomData,
         }
     }
 }
 
-impl<S, P> Stream for Filter<S, P, S::Item>
+impl<S, P> Stream for Filter<S, P>
 where
     S: Stream,
     P: FnMut(&S::Item) -> bool,

src/stream/stream/filter_map.rs

@@ -1,4 +1,3 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -7,29 +6,21 @@ use pin_project_lite::pin_project;
 use crate::stream::Stream;
 
 pin_project! {
-    #[doc(hidden)]
-    #[allow(missing_debug_implementations)]
-    pub struct FilterMap<S, F, T, B> {
+    #[derive(Debug)]
+    pub struct FilterMap<S, F> {
         #[pin]
         stream: S,
         f: F,
-        __from: PhantomData<T>,
-        __to: PhantomData<B>,
     }
 }
 
-impl<S, F, T, B> FilterMap<S, F, T, B> {
+impl<S, F> FilterMap<S, F> {
     pub(crate) fn new(stream: S, f: F) -> Self {
-        FilterMap {
-            stream,
-            f,
-            __from: PhantomData,
-            __to: PhantomData,
-        }
+        FilterMap { stream, f }
     }
 }
 
-impl<S, F, B> Stream for FilterMap<S, F, S::Item, B>
+impl<S, F, B> Stream for FilterMap<S, F>
 where
     S: Stream,
     F: FnMut(S::Item) -> Option<B>,

src/stream/stream/find.rs

@@ -1,31 +1,25 @@
-use std::marker::PhantomData;
-use std::pin::Pin;
 use std::future::Future;
+use std::pin::Pin;
 
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
 #[doc(hidden)]
 #[allow(missing_debug_implementations)]
-pub struct FindFuture<'a, S, P, T> {
+pub struct FindFuture<'a, S, P> {
     stream: &'a mut S,
     p: P,
-    __t: PhantomData<T>,
 }
 
-impl<'a, S, P, T> FindFuture<'a, S, P, T> {
+impl<'a, S, P> FindFuture<'a, S, P> {
     pub(super) fn new(stream: &'a mut S, p: P) -> Self {
-        FindFuture {
-            stream,
-            p,
-            __t: PhantomData,
-        }
+        FindFuture { stream, p }
     }
 }
 
-impl<S: Unpin, P, T> Unpin for FindFuture<'_, S, P, T> {}
+impl<S: Unpin, P> Unpin for FindFuture<'_, S, P> {}
 
-impl<'a, S, P> Future for FindFuture<'a, S, P, S::Item>
+impl<'a, S, P> Future for FindFuture<'a, S, P>
 where
     S: Stream + Unpin + Sized,
     P: FnMut(&S::Item) -> bool,

src/stream/stream/find_map.rs

@@ -1,33 +1,25 @@
-use std::marker::PhantomData;
+use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use std::future::Future;
 
 use crate::stream::Stream;
 
 #[doc(hidden)]
 #[allow(missing_debug_implementations)]
-pub struct FindMapFuture<'a, S, F, T, B> {
+pub struct FindMapFuture<'a, S, F> {
     stream: &'a mut S,
     f: F,
-    __b: PhantomData<B>,
-    __t: PhantomData<T>,
 }
 
-impl<'a, S, B, F, T> FindMapFuture<'a, S, F, T, B> {
+impl<'a, S, F> FindMapFuture<'a, S, F> {
     pub(super) fn new(stream: &'a mut S, f: F) -> Self {
-        FindMapFuture {
-            stream,
-            f,
-            __b: PhantomData,
-            __t: PhantomData,
-        }
+        FindMapFuture { stream, f }
     }
 }
 
-impl<S: Unpin, F, T, B> Unpin for FindMapFuture<'_, S, F, T, B> {}
+impl<S: Unpin, F> Unpin for FindMapFuture<'_, S, F> {}
 
-impl<'a, S, B, F> Future for FindMapFuture<'a, S, F, S::Item, B>
+impl<'a, S, B, F> Future for FindMapFuture<'a, S, F>
 where
     S: Stream + Unpin + Sized,
     F: FnMut(S::Item) -> Option<B>,

src/stream/stream/flat_map.rs

@@ -1,33 +1,36 @@
-use pin_project_lite::pin_project;
 use std::pin::Pin;
 
+use pin_project_lite::pin_project;
+
 use crate::prelude::*;
 use crate::stream::stream::map::Map;
 use crate::stream::{IntoStream, Stream};
 use crate::task::{Context, Poll};
 
 pin_project! {
+    /// A stream that maps each element to a stream, and yields the elements of the produced
+    /// streams.
+    ///
     /// This `struct` is created by the [`flat_map`] method on [`Stream`]. See its
     /// documentation for more.
     ///
     /// [`flat_map`]: trait.Stream.html#method.flat_map
     /// [`Stream`]: trait.Stream.html
-    #[allow(missing_debug_implementations)]
-    pub struct FlatMap<S, U, T, F> {
+    pub struct FlatMap<S, U, F> {
         #[pin]
-        stream: Map<S, F, T, U>,
+        stream: Map<S, F>,
         #[pin]
         inner_stream: Option<U>,
     }
 }
 
-impl<S, U, F> FlatMap<S, U, S::Item, F>
+impl<S, U, F> FlatMap<S, U, F>
 where
     S: Stream,
     U: IntoStream,
     F: FnMut(S::Item) -> U,
 {
-    pub(super) fn new(stream: S, f: F) -> FlatMap<S, U, S::Item, F> {
+    pub(super) fn new(stream: S, f: F) -> FlatMap<S, U, F> {
         FlatMap {
             stream: stream.map(f),
             inner_stream: None,
@@ -35,7 +38,7 @@ where
     }
 }
 
-impl<S, U, F> Stream for FlatMap<S, U, S::Item, F>
+impl<S, U, F> Stream for FlatMap<S, U, F>
 where
     S: Stream,
     S::Item: IntoStream<IntoStream = U, Item = U::Item>,

src/stream/stream/flatten.rs

@@ -1,30 +1,38 @@
-use pin_project_lite::pin_project;
+use std::fmt;
 use std::pin::Pin;
 
+use pin_project_lite::pin_project;
+
 use crate::stream::{IntoStream, Stream};
 use crate::task::{Context, Poll};
 
 pin_project! {
+    /// A stream that flattens one level of nesting in an stream of things that can be turned into
+    /// streams.
+    ///
     /// This `struct` is created by the [`flatten`] method on [`Stream`]. See its
     /// documentation for more.
     ///
     /// [`flatten`]: trait.Stream.html#method.flatten
     /// [`Stream`]: trait.Stream.html
-    #[allow(missing_debug_implementations)]
-    pub struct Flatten<S, U> {
+    pub struct Flatten<S>
+    where
+        S: Stream,
+        S::Item: IntoStream,
+    {
         #[pin]
         stream: S,
         #[pin]
-        inner_stream: Option<U>,
+        inner_stream: Option<<S::Item as IntoStream>::IntoStream>,
     }
 }
 
-impl<S> Flatten<S, S::Item>
+impl<S> Flatten<S>
 where
     S: Stream,
     S::Item: IntoStream,
 {
-    pub(super) fn new(stream: S) -> Flatten<S, S::Item> {
+    pub(super) fn new(stream: S) -> Flatten<S> {
         Flatten {
             stream,
             inner_stream: None,
@@ -32,7 +40,7 @@ where
     }
 }
 
-impl<S, U> Stream for Flatten<S, <S::Item as IntoStream>::IntoStream>
+impl<S, U> Stream for Flatten<S>
 where
     S: Stream,
     S::Item: IntoStream<IntoStream = U, Item = U::Item>,
@@ -56,3 +64,16 @@ where
         }
     }
 }
+
+impl<S, U> fmt::Debug for Flatten<S>
+where
+    S: fmt::Debug + Stream,
+    S::Item: IntoStream<IntoStream = U, Item = U::Item>,
+    U: fmt::Debug + Stream,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Flatten")
+            .field("inner", &self.stream)
+            .finish()
+    }
+}

src/stream/stream/fold.rs

@@ -1,6 +1,5 @@
-use std::marker::PhantomData;
-use std::pin::Pin;
 use std::future::Future;
+use std::pin::Pin;
 
 use pin_project_lite::pin_project;
 
@@ -8,29 +7,26 @@ use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
 pin_project! {
-    #[doc(hidden)]
-    #[allow(missing_debug_implementations)]
-    pub struct FoldFuture<S, F, T, B> {
+    #[derive(Debug)]
+    pub struct FoldFuture<S, F, B> {
         #[pin]
         stream: S,
         f: F,
         acc: Option<B>,
-        __t: PhantomData<T>,
     }
 }
 
-impl<S, F, T, B> FoldFuture<S, F, T, B> {
+impl<S, F, B> FoldFuture<S, F, B> {
     pub(super) fn new(stream: S, init: B, f: F) -> Self {
         FoldFuture {
             stream,
             f,
             acc: Some(init),
-            __t: PhantomData,
         }
     }
 }
 
-impl<S, F, B> Future for FoldFuture<S, F, S::Item, B>
+impl<S, F, B> Future for FoldFuture<S, F, B>
 where
     S: Stream + Sized,
     F: FnMut(B, S::Item) -> B,

src/stream/stream/for_each.rs

@@ -1,4 +1,3 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
 use std::future::Future;
 
@@ -10,25 +9,23 @@ use crate::task::{Context, Poll};
 pin_project! {
     #[doc(hidden)]
     #[allow(missing_debug_implementations)]
-    pub struct ForEachFuture<S, F, T> {
+    pub struct ForEachFuture<S, F> {
         #[pin]
         stream: S,
         f: F,
-        __t: PhantomData<T>,
     }
 }
 
-impl<S, F, T> ForEachFuture<S, F, T> {
+impl<S, F> ForEachFuture<S, F> {
     pub(super) fn new(stream: S, f: F) -> Self {
         ForEachFuture {
             stream,
             f,
-            __t: PhantomData,
         }
     }
 }
 
-impl<S, F> Future for ForEachFuture<S, F, S::Item>
+impl<S, F> Future for ForEachFuture<S, F>
 where
     S: Stream + Sized,
     F: FnMut(S::Item),

src/stream/stream/inspect.rs

@@ -1,4 +1,3 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
 
 use pin_project_lite::pin_project;
@@ -15,25 +14,23 @@ pin_project! {
     /// [`inspect`]: trait.Stream.html#method.inspect
     /// [`Stream`]: trait.Stream.html
     #[derive(Debug)]
-    pub struct Inspect<S, F, T> {
+    pub struct Inspect<S, F> {
         #[pin]
         stream: S,
         f: F,
-        __t: PhantomData<T>,
     }
 }
 
-impl<S, F, T> Inspect<S, F, T> {
+impl<S, F> Inspect<S, F> {
     pub(super) fn new(stream: S, f: F) -> Self {
         Inspect {
             stream,
             f,
-            __t: PhantomData,
         }
     }
 }
 
-impl<S, F> Stream for Inspect<S, F, S::Item>
+impl<S, F> Stream for Inspect<S, F>
 where
     S: Stream,
     F: FnMut(&S::Item),

src/stream/stream/map.rs

@@ -1,4 +1,3 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
 
 use pin_project_lite::pin_project;
@@ -7,29 +6,25 @@ use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
 pin_project! {
-    #[doc(hidden)]
-    #[allow(missing_debug_implementations)]
-    pub struct Map<S, F, T, B> {
+    /// A stream that maps value of another stream with a function.
+    #[derive(Debug)]
+    pub struct Map<S, F> {
         #[pin]
         stream: S,
         f: F,
-        __from: PhantomData<T>,
-        __to: PhantomData<B>,
     }
 }
 
-impl<S, F, T, B> Map<S, F, T, B> {
+impl<S, F> Map<S, F> {
     pub(crate) fn new(stream: S, f: F) -> Self {
         Map {
             stream,
             f,
-            __from: PhantomData,
-            __to: PhantomData,
         }
     }
 }
 
-impl<S, F, B> Stream for Map<S, F, S::Item, B>
+impl<S, F, B> Stream for Map<S, F>
 where
     S: Stream,
     F: FnMut(S::Item) -> B,

src/stream/stream/mod.rs

@@ -288,6 +288,7 @@ extension_trait! {
             Creates a stream that yields elements based on a predicate.
 
             # Examples
+
             ```
             # fn main() { async_std::task::block_on(async {
             #
@@ -300,12 +301,11 @@ extension_trait! {
             assert_eq!(s.next().await, Some(1));
             assert_eq!(s.next().await, Some(2));
             assert_eq!(s.next().await, None);
-
             #
             # }) }
             ```
         "#]
-        fn take_while<P>(self, predicate: P) -> TakeWhile<Self, P, Self::Item>
+        fn take_while<P>(self, predicate: P) -> TakeWhile<Self, P>
         where
             Self: Sized,
             P: FnMut(&Self::Item) -> bool,
@@ -410,10 +410,10 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn cloned<'a,T>(self) -> Cloned<Self>
+        fn cloned<'a, T>(self) -> Cloned<Self>
         where
             Self: Sized + Stream<Item = &'a T>,
-            T : 'a + Clone,
+            T: Clone + 'a,
         {
             Cloned::new(self)
         }
@@ -443,10 +443,10 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn copied<'a,T>(self) -> Copied<Self>
+        fn copied<'a, T>(self) -> Copied<Self>
         where
             Self: Sized + Stream<Item = &'a T>,
-            T : 'a + Copy,
+            T: Copy + 'a,
         {
             Copied::new(self)
         }
@@ -475,10 +475,9 @@ extension_trait! {
             # })
             ```
         "#]
-        fn cycle(self) -> Cycle<Self, Self::Item>
-            where
-                Self: Sized,
-                Self::Item: Clone,
+        fn cycle(self) -> Cycle<Self>
+        where
+            Self: Clone + Sized,
         {
             Cycle::new(self)
         }
@@ -505,7 +504,6 @@ extension_trait! {
             assert_eq!(s.next().await, Some((1, 'b')));
             assert_eq!(s.next().await, Some((2, 'c')));
             assert_eq!(s.next().await, None);
-
             #
             # }) }
             ```
@@ -540,7 +538,7 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn map<B, F>(self, f: F) -> Map<Self, F, Self::Item, B>
+        fn map<B, F>(self, f: F) -> Map<Self, F>
         where
             Self: Sized,
             F: FnMut(Self::Item) -> B,
@@ -565,17 +563,18 @@ extension_trait! {
             let s = stream::from_iter(vec![1, 2, 3, 4, 5]);
 
             let sum = s
-                    .inspect(|x| println!("about to filter {}", x))
-                    .filter(|x| x % 2 == 0)
-                    .inspect(|x| println!("made it through filter: {}", x))
-                    .fold(0, |sum, i| sum + i).await;
+               .inspect(|x| println!("about to filter {}", x))
+               .filter(|x| x % 2 == 0)
+               .inspect(|x| println!("made it through filter: {}", x))
+               .fold(0, |sum, i| sum + i)
+               .await;
 
             assert_eq!(sum, 6);
             #
             # }) }
             ```
         "#]
-        fn inspect<F>(self, f: F) -> Inspect<Self, F, Self::Item>
+        fn inspect<F>(self, f: F) -> Inspect<Self, F>
         where
             Self: Sized,
             F: FnMut(&Self::Item),
@@ -618,7 +617,6 @@ extension_trait! {
             #
             # }) }
             ```
-
         "#]
         fn last(
             self,
@@ -685,7 +683,7 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn filter<P>(self, predicate: P) -> Filter<Self, P, Self::Item>
+        fn filter<P>(self, predicate: P) -> Filter<Self, P>
         where
             Self: Sized,
             P: FnMut(&Self::Item) -> bool,
@@ -721,7 +719,7 @@ extension_trait! {
         "#]
         #[cfg(feature = "unstable")]
         #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-        fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, Self::Item, F>
+        fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, F>
             where
                 Self: Sized,
                 U: IntoStream,
@@ -755,7 +753,7 @@ extension_trait! {
         "#]
         #[cfg(feature = "unstable")]
         #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-        fn flatten(self) -> Flatten<Self, Self::Item>
+        fn flatten(self) -> Flatten<Self>
         where
             Self: Sized,
             Self::Item: IntoStream,
@@ -796,7 +794,7 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn filter_map<B, F>(self, f: F) -> FilterMap<Self, F, Self::Item, B>
+        fn filter_map<B, F>(self, f: F) -> FilterMap<Self, F>
         where
             Self: Sized,
             F: FnMut(Self::Item) -> Option<B>,
@@ -804,7 +802,7 @@ extension_trait! {
             FilterMap::new(self, f)
         }
 
-         #[doc = r#"
+        #[doc = r#"
             Returns the element that gives the minimum value with respect to the
             specified key function. If several elements are equally minimum,
             the first element is returned. If the stream is empty, `None` is returned.
@@ -828,19 +826,19 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn min_by_key<K>(
+        fn min_by_key<B, F>(
             self,
-            key_by: K,
-        ) -> impl Future<Output = Option<Self::Item>> [MinByKeyFuture<Self, Self::Item, K>]
+            key_by: F,
+        ) -> impl Future<Output = Option<Self::Item>> [MinByKeyFuture<Self, Self::Item, F>]
         where
             Self: Sized,
-            Self::Item: Ord,
-            K: FnMut(&Self::Item) -> Self::Item,
+            B: Ord,
+            F: FnMut(&Self::Item) -> B,
         {
             MinByKeyFuture::new(self, key_by)
         }
 
-         #[doc = r#"
+        #[doc = r#"
             Returns the element that gives the maximum value with respect to the
             specified key function. If several elements are equally maximum,
             the first element is returned. If the stream is empty, `None` is returned.
@@ -864,14 +862,14 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn max_by_key<K>(
+        fn max_by_key<B, F>(
             self,
-            key_by: K,
-        ) -> impl Future<Output = Option<Self::Item>> [MaxByKeyFuture<Self, Self::Item, K>]
+            key_by: F,
+        ) -> impl Future<Output = Option<Self::Item>> [MaxByKeyFuture<Self, Self::Item, F>]
         where
             Self: Sized,
-            Self::Item: Ord,
-            K: FnMut(&Self::Item) -> Self::Item,
+            B: Ord,
+            F: FnMut(&Self::Item) -> B,
         {
             MaxByKeyFuture::new(self, key_by)
         }
@@ -1043,7 +1041,7 @@ extension_trait! {
             n: usize,
         ) -> impl Future<Output = Option<Self::Item>> + '_ [NthFuture<'_, Self>]
         where
-            Self: Sized,
+            Self: Unpin + Sized,
         {
             NthFuture::new(self, n)
         }
@@ -1151,9 +1149,9 @@ extension_trait! {
         fn find<P>(
             &mut self,
             p: P,
-        ) -> impl Future<Output = Option<Self::Item>> + '_ [FindFuture<'_, Self, P, Self::Item>]
+        ) -> impl Future<Output = Option<Self::Item>> + '_ [FindFuture<'_, Self, P>]
         where
-            Self: Sized,
+            Self: Unpin + Sized,
             P: FnMut(&Self::Item) -> bool,
         {
             FindFuture::new(self, p)
@@ -1179,9 +1177,9 @@ extension_trait! {
         fn find_map<F, B>(
             &mut self,
             f: F,
-        ) -> impl Future<Output = Option<B>> + '_ [FindMapFuture<'_, Self, F, Self::Item, B>]
+        ) -> impl Future<Output = Option<B>> + '_ [FindMapFuture<'_, Self, F>]
         where
-            Self: Sized,
+            Self: Unpin + Sized,
             F: FnMut(Self::Item) -> Option<B>,
         {
             FindMapFuture::new(self, f)
@@ -1213,7 +1211,7 @@ extension_trait! {
             self,
             init: B,
             f: F,
-        ) -> impl Future<Output = B> [FoldFuture<Self, F, Self::Item, B>]
+        ) -> impl Future<Output = B> [FoldFuture<Self, F, B>]
         where
             Self: Sized,
             F: FnMut(B, Self::Item) -> B,
@@ -1248,7 +1246,7 @@ extension_trait! {
         fn for_each<F>(
             self,
             f: F,
-        ) -> impl Future<Output = ()> [ForEachFuture<Self, F, Self::Item>]
+        ) -> impl Future<Output = ()> [ForEachFuture<Self, F>]
         where
             Self: Sized,
             F: FnMut(Self::Item),
@@ -1389,7 +1387,7 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn skip_while<P>(self, predicate: P) -> SkipWhile<Self, P, Self::Item>
+        fn skip_while<P>(self, predicate: P) -> SkipWhile<Self, P>
         where
             Self: Sized,
             P: FnMut(&Self::Item) -> bool,
@@ -1472,7 +1470,7 @@ extension_trait! {
             use async_std::prelude::*;
             use async_std::stream;
 
-            let s = stream::from_iter(vec![1usize, 2, 3]);
+            let mut s = stream::from_iter(vec![1usize, 2, 3]);
             let sum = s.try_fold(0, |acc, v| {
                 if (acc+v) % 2 == 1 {
                     Ok(v+3)
@@ -1487,12 +1485,12 @@ extension_trait! {
             ```
         "#]
         fn try_fold<B, F, T, E>(
-            self,
+            &mut self,
             init: T,
             f: F,
-        ) -> impl Future<Output = Result<T, E>> [TryFoldFuture<Self, F, T>]
+        ) -> impl Future<Output = Result<T, E>> + '_ [TryFoldFuture<'_, Self, F, T>]
         where
-            Self: Sized,
+            Self: Unpin + Sized,
             F: FnMut(B, Self::Item) -> Result<T, E>,
         {
             TryFoldFuture::new(self, init, f)
@@ -1512,7 +1510,7 @@ extension_trait! {
 
             let (tx, rx) = channel();
 
-            let s = stream::from_iter(vec![1u8, 2, 3]);
+            let mut s = stream::from_iter(vec![1u8, 2, 3]);
             let s = s.try_for_each(|v| {
                 if v % 2 == 1 {
                     tx.clone().send(v).unwrap();
@@ -1533,11 +1531,11 @@ extension_trait! {
             ```
         "#]
         fn try_for_each<F, E>(
-            self,
+            &mut self,
             f: F,
-        ) -> impl Future<Output = E> [TryForEachFuture<Self, F, Self::Item, E>]
+        ) -> impl Future<Output = E> + 'a [TryForEachFuture<'_, Self, F>]
         where
-            Self: Sized,
+            Self: Unpin + Sized,
             F: FnMut(Self::Item) -> Result<(), E>,
         {
             TryForEachFuture::new(self, f)
@@ -1582,7 +1580,7 @@ extension_trait! {
         #[inline]
         fn zip<U>(self, other: U) -> Zip<Self, U>
         where
-            Self: Sized + Stream,
+            Self: Sized,
             U: Stream,
         {
             Zip::new(self, other)
@@ -1641,7 +1639,6 @@ extension_trait! {
         "#]
         #[cfg(feature = "unstable")]
         #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-        #[must_use = "if you really need to exhaust the iterator, consider `.for_each(drop)` instead (TODO)"]
         fn collect<'a, B>(
             self,
         ) -> impl Future<Output = B> + 'a [Pin<Box<dyn Future<Output = B> + 'a>>]
@@ -1740,28 +1737,28 @@ extension_trait! {
             use async_std::stream;
 
             let s = stream::from_iter(vec![1usize, 2, 3]);
-            let res = s.clone().position(|x| *x == 1).await;
+            let res = s.clone().position(|x| x == 1).await;
             assert_eq!(res, Some(0));
 
-            let res = s.clone().position(|x| *x == 2).await;
+            let res = s.clone().position(|x| x == 2).await;
             assert_eq!(res, Some(1));
 
-            let res = s.clone().position(|x| *x == 3).await;
+            let res = s.clone().position(|x| x == 3).await;
             assert_eq!(res, Some(2));
 
-            let res = s.clone().position(|x| *x == 4).await;
+            let res = s.clone().position(|x| x == 4).await;
             assert_eq!(res, None);
             #
             # }) }
             ```
         "#]
         fn position<P>(
-           self,
-           predicate: P
-        ) -> impl Future<Output = Option<usize>>  [PositionFuture<Self, P>]
+           &mut self,
+           predicate: P,
+        ) -> impl Future<Output = Option<usize>> + '_ [PositionFuture<'_, Self, P>]
         where
-            Self: Sized,
-            P: FnMut(&Self::Item) -> bool,
+            Self: Unpin + Sized,
+            P: FnMut(Self::Item) -> bool,
         {
             PositionFuture::new(self, predicate)
         }
@@ -1805,10 +1802,12 @@ extension_trait! {
             CmpFuture::new(self, other)
         }
 
-           #[doc = r#"
+        #[doc = r#"
             Determines if the elements of this `Stream` are lexicographically
             not equal to those of another.
+
             # Examples
+
             ```
             # fn main() { async_std::task::block_on(async {
             #
@@ -1833,7 +1832,7 @@ extension_trait! {
            other: S
         ) -> impl Future<Output = bool> [NeFuture<Self, S>]
         where
-            Self: Sized + Stream,
+            Self: Sized,
             S: Sized + Stream,
             <Self as Stream>::Item: PartialEq<S::Item>,
         {
@@ -2026,11 +2025,11 @@ extension_trait! {
         }
 
         #[doc = r#"
-            Sums the elements of an iterator.
+            Sums the elements of a stream.
 
             Takes each element, adds them together, and returns the result.
 
-            An empty iterator returns the zero value of the type.
+            An empty streams returns the zero value of the type.
 
             # Panics
 
@@ -2063,15 +2062,15 @@ extension_trait! {
         ) -> impl Future<Output = S> + 'a [Pin<Box<dyn Future<Output = S> + 'a>>]
         where
             Self: Sized + Stream<Item = S> + 'a,
-            S: Sum,
+            S: Sum<Self::Item>,
         {
             Sum::sum(self)
         }
 
         #[doc = r#"
-            Iterates over the entire iterator, multiplying all the elements
+            Multiplies all elements of the stream.
 
-            An empty iterator returns the one value of the type.
+            An empty stream returns the one value of the type.
 
             # Panics
 

src/stream/stream/position.rs

@@ -1,24 +1,21 @@
-use std::pin::Pin;
 use std::future::Future;
-
-use pin_project_lite::pin_project;
+use std::pin::Pin;
 
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
-pin_project! {
-    #[doc(hidden)]
-    #[allow(missing_debug_implementations)]
-    pub struct PositionFuture<S, P> {
-        #[pin]
-        stream: S,
-        predicate: P,
-        index:usize,
-    }
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct PositionFuture<'a, S, P> {
+    stream: &'a mut S,
+    predicate: P,
+    index: usize,
 }
 
-impl<S, P> PositionFuture<S, P> {
-    pub(super) fn new(stream: S, predicate: P) -> Self {
+impl<'a, S, P> Unpin for PositionFuture<'a, S, P> {}
+
+impl<'a, S, P> PositionFuture<'a, S, P> {
+    pub(super) fn new(stream: &'a mut S, predicate: P) -> Self {
         PositionFuture {
             stream,
             predicate,
@@ -27,23 +24,25 @@ impl<S, P> PositionFuture<S, P> {
     }
 }
 
-impl<S, P> Future for PositionFuture<S, P>
+impl<'a, S, P> Future for PositionFuture<'a, S, P>
 where
-    S: Stream,
-    P: FnMut(&S::Item) -> bool,
+    S: Stream + Unpin,
+    P: FnMut(S::Item) -> bool,
 {
     type Output = Option<usize>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.project();
-        let next = futures_core::ready!(this.stream.poll_next(cx));
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let next = futures_core::ready!(Pin::new(&mut self.stream).poll_next(cx));
 
         match next {
-            Some(v) if (this.predicate)(&v) => Poll::Ready(Some(*this.index)),
-            Some(_) => {
-                cx.waker().wake_by_ref();
-                *this.index += 1;
-                Poll::Pending
+            Some(v) => {
+                if (&mut self.predicate)(v) {
+                    Poll::Ready(Some(self.index))
+                } else {
+                    cx.waker().wake_by_ref();
+                    self.index += 1;
+                    Poll::Pending
+                }
             }
             None => Poll::Ready(None),
         }

src/stream/stream/skip_while.rs

@@ -1,4 +1,3 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
 
 use pin_project_lite::pin_project;
@@ -15,25 +14,23 @@ pin_project! {
     /// [`skip_while`]: trait.Stream.html#method.skip_while
     /// [`Stream`]: trait.Stream.html
     #[derive(Debug)]
-    pub struct SkipWhile<S, P, T> {
+    pub struct SkipWhile<S, P> {
         #[pin]
         stream: S,
         predicate: Option<P>,
-        __t: PhantomData<T>,
     }
 }
 
-impl<S, P, T> SkipWhile<S, P, T> {
+impl<S, P> SkipWhile<S, P> {
     pub(crate) fn new(stream: S, predicate: P) -> Self {
         SkipWhile {
             stream,
             predicate: Some(predicate),
-            __t: PhantomData,
         }
     }
 }
 
-impl<S, P> Stream for SkipWhile<S, P, S::Item>
+impl<S, P> Stream for SkipWhile<S, P>
 where
     S: Stream,
     P: FnMut(&S::Item) -> bool,

src/stream/stream/take_while.rs

@@ -1,4 +1,3 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
 
 use pin_project_lite::pin_project;
@@ -15,25 +14,23 @@ pin_project! {
     /// [`take_while`]: trait.Stream.html#method.take_while
     /// [`Stream`]: trait.Stream.html
     #[derive(Debug)]
-    pub struct TakeWhile<S, P, T> {
+    pub struct TakeWhile<S, P> {
         #[pin]
         stream: S,
         predicate: P,
-        __t: PhantomData<T>,
     }
 }
 
-impl<S, P, T> TakeWhile<S, P, T> {
+impl<S, P> TakeWhile<S, P> {
     pub(super) fn new(stream: S, predicate: P) -> Self {
         TakeWhile {
             stream,
             predicate,
-            __t: PhantomData,
         }
     }
 }
 
-impl<S, P> Stream for TakeWhile<S, P, S::Item>
+impl<S, P> Stream for TakeWhile<S, P>
 where
     S: Stream,
     P: FnMut(&S::Item) -> bool,

src/stream/stream/timeout.rs

@@ -22,7 +22,7 @@ pin_project! {
 }
 
 impl<S: Stream> Timeout<S> {
-    pub fn new(stream: S, dur: Duration) -> Timeout<S> {
+    pub(crate) fn new(stream: S, dur: Duration) -> Timeout<S> {
         let delay = Delay::new(dur);
 
         Timeout { stream, delay }

src/stream/stream/try_fold.rs

@@ -1,58 +1,51 @@
-use std::marker::PhantomData;
 use std::pin::Pin;
-use std::future::Future;
-
-use pin_project_lite::pin_project;
 
+use crate::future::Future;
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
-pin_project! {
-    #[doc(hidden)]
-    #[allow(missing_debug_implementations)]
-    pub struct TryFoldFuture<S, F, T> {
-        #[pin]
-        stream: S,
-        f: F,
-        acc: Option<T>,
-        __t: PhantomData<T>,
-    }
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct TryFoldFuture<'a, S, F, T> {
+    stream: &'a mut S,
+    f: F,
+    acc: Option<T>,
 }
 
-impl<S, F, T> TryFoldFuture<S, F, T> {
-    pub(super) fn new(stream: S, init: T, f: F) -> Self {
+impl<'a, S, F, T> Unpin for TryFoldFuture<'a, S, F, T> {}
+
+impl<'a, S, F, T> TryFoldFuture<'a, S, F, T> {
+    pub(super) fn new(stream: &'a mut S, init: T, f: F) -> Self {
         TryFoldFuture {
             stream,
             f,
             acc: Some(init),
-            __t: PhantomData,
         }
     }
 }
 
-impl<S, F, T, E> Future for TryFoldFuture<S, F, T>
+impl<'a, S, F, T, E> Future for TryFoldFuture<'a, S, F, T>
 where
-    S: Stream + Sized,
+    S: Stream + Unpin,
     F: FnMut(T, S::Item) -> Result<T, E>,
 {
     type Output = Result<T, E>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.project();
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         loop {
-            let next = futures_core::ready!(this.stream.as_mut().poll_next(cx));
+            let next = futures_core::ready!(Pin::new(&mut self.stream).poll_next(cx));
 
             match next {
                 Some(v) => {
-                    let old = this.acc.take().unwrap();
-                    let new = (this.f)(old, v);
+                    let old = self.acc.take().unwrap();
+                    let new = (&mut self.f)(old, v);
 
                     match new {
-                        Ok(o) => *this.acc = Some(o),
+                        Ok(o) => self.acc = Some(o),
                         Err(e) => return Poll::Ready(Err(e)),
                     }
                 }
-                None => return Poll::Ready(Ok(this.acc.take().unwrap())),
+                None => return Poll::Ready(Ok(self.acc.take().unwrap())),
             }
         }
     }

src/stream/stream/try_for_each.rs

@@ -1,52 +1,39 @@
 use std::future::Future;
-use std::marker::PhantomData;
 use std::pin::Pin;
 
-use pin_project_lite::pin_project;
-
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
-pin_project! {
-    #[doc(hidden)]
-    #[allow(missing_debug_implementations)]
-    pub struct TryForEachFuture<S, F, T, R> {
-        #[pin]
-        stream: S,
-        f: F,
-        __from: PhantomData<T>,
-        __to: PhantomData<R>,
-    }
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct TryForEachFuture<'a, S, F> {
+    stream: &'a mut S,
+    f: F,
 }
 
-impl<S, F, T, R> TryForEachFuture<S, F, T, R> {
-    pub(crate) fn new(stream: S, f: F) -> Self {
-        TryForEachFuture {
-            stream,
-            f,
-            __from: PhantomData,
-            __to: PhantomData,
-        }
+impl<'a, S, F> Unpin for TryForEachFuture<'a, S, F> {}
+
+impl<'a, S, F> TryForEachFuture<'a, S, F> {
+    pub(crate) fn new(stream: &'a mut S, f: F) -> Self {
+        TryForEachFuture { stream, f }
     }
 }
 
-impl<S, F, E> Future for TryForEachFuture<S, F, S::Item, E>
+impl<'a, S, F, E> Future for TryForEachFuture<'a, S, F>
 where
-    S: Stream,
-    S::Item: std::fmt::Debug,
+    S: Stream + Unpin,
     F: FnMut(S::Item) -> Result<(), E>,
 {
     type Output = Result<(), E>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.project();
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         loop {
-            let item = futures_core::ready!(this.stream.as_mut().poll_next(cx));
+            let item = futures_core::ready!(Pin::new(&mut self.stream).poll_next(cx));
 
             match item {
                 None => return Poll::Ready(Ok(())),
                 Some(v) => {
-                    let res = (this.f)(v);
+                    let res = (&mut self.f)(v);
                     if let Err(e) = res {
                         return Poll::Ready(Err(e));
                     }

src/stream/stream/zip.rs

@@ -7,7 +7,7 @@ use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
 pin_project! {
-    /// An iterator that iterates two other iterators simultaneously.
+    /// A stream that takes items from two other streams simultaneously.
     ///
     /// This `struct` is created by the [`zip`] method on [`Stream`]. See its
     /// documentation for more.