from/into stream

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

update examples

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

impl collect

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

compiles!

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

layout base for collect into vec

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

fmt

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

progress

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

compiles!

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

define failing test

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

cargo fmt

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

stuck again

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

fix trait bounds!

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

cargo fmt

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

hide dyn fut impl

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

dyn ret for vec

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

cargo fmt

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

collect docs

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

remove macro from vec::from_stream

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

shorten collect trait bound

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

Remove some Unpin and Send bounds

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

Author: yoshuawuyts

src/lib.rs

@@ -42,4 +42,6 @@ pub mod stream;
 pub mod sync;
 pub mod task;
 
+mod vec;
+
 pub(crate) mod utils;

src/stream/from_stream.rs

@@ -0,0 +1,28 @@
+use super::IntoStream;
+
+use std::pin::Pin;
+
+/// Conversion from a `Stream`.
+///
+/// By implementing `FromStream` for a type, you define how it will be created from a stream.
+/// This is common for types which describe a collection of some kind.
+///
+/// See also: [`IntoStream`].
+///
+/// [`IntoStream`]: trait.IntoStream.html
+pub trait FromStream<T> {
+    /// Creates a value from a stream.
+    ///
+    /// # Examples
+    ///
+    /// Basic usage:
+    ///
+    /// ```
+    /// // use async_std::stream::FromStream;
+    ///
+    /// // let _five_fives = async_std::stream::repeat(5).take(5);
+    /// ```
+    fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
+        stream: S,
+    ) -> Pin<Box<dyn core::future::Future<Output = Self> + 'a>>;
+}

src/stream/into_stream.rs

@@ -0,0 +1,35 @@
+use futures::stream::Stream;
+
+/// Conversion into a `Stream`.
+///
+/// By implementing `IntoIterator` for a type, you define how it will be
+/// converted to an iterator. This is common for types which describe a
+/// collection of some kind.
+///
+/// [`from_stream`]: #tymethod.from_stream
+/// [`Stream`]: trait.Stream.html
+/// [`collect`]: trait.Stream.html#method.collect
+///
+/// See also: [`FromStream`].
+///
+/// [`FromStream`]: trait.FromStream.html
+pub trait IntoStream {
+    /// The type of the elements being iterated over.
+    type Item;
+
+    /// Which kind of stream are we turning this into?
+    type IntoStream: Stream<Item = Self::Item>;
+
+    /// Creates a stream from a value.
+    fn into_stream(self) -> Self::IntoStream;
+}
+
+impl<I: Stream> IntoStream for I {
+    type Item = I::Item;
+    type IntoStream = I;
+
+    #[inline]
+    fn into_stream(self) -> I {
+        self
+    }
+}

src/stream/mod.rs

@@ -22,11 +22,15 @@
 //! ```
 
 pub use empty::{empty, Empty};
+pub use from_stream::FromStream;
+pub use into_stream::IntoStream;
 pub use once::{once, Once};
 pub use repeat::{repeat, Repeat};
 pub use stream::{Stream, Take};
 
 mod empty;
+mod from_stream;
+mod into_stream;
 mod min_by;
 mod once;
 mod repeat;

src/stream/stream.rs

@@ -26,6 +26,7 @@ use std::pin::Pin;
 
 use cfg_if::cfg_if;
 
+use super::from_stream::FromStream;
 use super::min_by::MinBy;
 use crate::future::Future;
 use crate::task::{Context, Poll};
@@ -51,6 +52,21 @@ cfg_if! {
     }
 }
 
+cfg_if! {
+    if #[cfg(feature = "docs")] {
+        #[doc(hidden)]
+        pub struct DynFuture<'a, T>(std::marker::PhantomData<&'a T>);
+
+        macro_rules! dyn_ret {
+            ($a:lifetime, $o:ty) => (DynFuture<$a, $o>);
+        }
+    } else {
+        macro_rules! dyn_ret {
+            ($a:lifetime, $o:ty) => (Pin<Box<dyn core::future::Future<Output = $o> + 'a>>)
+        }
+    }
+}
+
 /// An asynchronous stream of values.
 ///
 /// This trait is an async version of [`std::iter::Iterator`].
@@ -227,7 +243,6 @@ pub trait Stream {
     ///
     /// Basic usage:
     ///
-    /// ```
     /// # fn main() { async_std::task::block_on(async {
     /// #
     /// use async_std::prelude::*;
@@ -266,6 +281,48 @@ pub trait Stream {
             f,
         }
     }
+
+    /// Transforms a stream into a collection.
+    ///
+    /// `collect()` can take anything streamable, and turn it into a relevant
+    /// collection. This is one of the more powerful methods in the async
+    /// standard library, used in a variety of contexts.
+    ///
+    /// The most basic pattern in which `collect()` is used is to turn one
+    /// collection into another. You take a collection, call [`stream`] on it,
+    /// do a bunch of transformations, and then `collect()` at the end.
+    ///
+    /// Because `collect()` is so general, it can cause problems with type
+    /// inference. As such, `collect()` is one of the few times you'll see
+    /// the syntax affectionately known as the 'turbofish': `::<>`. This
+    /// helps the inference algorithm understand specifically which collection
+    /// you're trying to collect into.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # fn main() { async_std::task::block_on(async {
+    /// #
+    /// use async_std::prelude::*;
+    /// use async_std::stream;
+    ///
+    /// let s = stream::repeat(9u8).take(3);
+    /// let buf: Vec<u8> = s.collect().await;
+    ///
+    /// assert_eq!(buf, vec![9; 3]);
+    /// #
+    /// # }) }
+    /// ```
+    ///
+    /// [`stream`]: trait.Stream.html#tymethod.next
+    #[must_use = "if you really need to exhaust the iterator, consider `.for_each(drop)` instead (TODO)"]
+    fn collect<'a, B>(self) -> dyn_ret!('a, B)
+    where
+        Self: futures::stream::Stream + Sized + 'a,
+        B: FromStream<<Self as futures::stream::Stream>::Item>,
+    {
+        FromStream::from_stream(self)
+    }
 }
 
 impl<T: futures_core::stream::Stream + Unpin + ?Sized> Stream for T {

src/vec/from_stream.rs

@@ -0,0 +1,25 @@
+use crate::stream::{FromStream, IntoStream, Stream};
+
+use std::pin::Pin;
+
+impl<T> FromStream<T> for Vec<T> {
+    #[inline]
+    fn from_stream<'a, S: IntoStream<Item = T>>(
+        stream: S,
+    ) -> Pin<Box<dyn core::future::Future<Output = Self> + 'a>>
+    where
+        <S as IntoStream>::IntoStream: 'a,
+    {
+        let stream = stream.into_stream();
+
+        Pin::from(Box::new(async move {
+            pin_utils::pin_mut!(stream);
+
+            let mut out = vec![];
+            while let Some(item) = stream.next().await {
+                out.push(item);
+            }
+            out
+        }))
+    }
+}

src/vec/mod.rs

@@ -0,0 +1,9 @@
+//! The Rust core allocation and collections library
+//!
+//! This library provides smart pointers and collections for managing
+//! heap-allocated values.
+
+mod from_stream;
+
+#[doc(inline)]
+pub use std::vec::Vec;