Add stream::Extend

Author: tirr-c

src/stream/extend.rs

@@ -0,0 +1,50 @@
+use std::pin::Pin;
+
+use crate::future::Future;
+use crate::stream::{IntoStream, Stream};
+
+/// Extend a collection with the contents of a stream.
+///
+/// Streams produce a series of values asynchronously, and collections can also be thought of as a
+/// series of values. The `Extend` trait bridges this gap, allowing you to extend a collection
+/// asynchronously by including the contents of that stream. When extending a collection with an
+/// already existing key, that entry is updated or, in the case of collections that permit multiple
+/// entries with equal keys, that entry is inserted.
+///
+/// ## Examples
+///
+/// ```
+/// # fn main() { async_std::task::block_on(async {
+/// #
+/// use async_std::prelude::*;
+/// use async_std::stream::{self, Extend};
+///
+/// let mut v: Vec<usize> = vec![1, 2];
+/// let s = stream::repeat(3usize).take(3);
+/// v.extend_with_stream(s).await;
+///
+/// assert_eq!(v, vec![1, 2, 3, 3, 3]);
+/// #
+/// # }) }
+/// ```
+#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+pub trait Extend<A> {
+    /// Extends a collection with the contents of a stream.
+    fn extend_with_stream<'a, T: IntoStream<Item = A> + 'a>(
+        &'a mut self,
+        stream: T,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>>;
+}
+
+impl Extend<()> for () {
+    fn extend_with_stream<'a, T: IntoStream<Item = ()> + 'a>(
+        &'a mut self,
+        stream: T,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+        let stream = stream.into_stream();
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+            while let Some(_) = stream.next().await {}
+        })
+    }
+}

src/stream/mod.rs

@@ -23,6 +23,7 @@
 
 pub use double_ended_stream::DoubleEndedStream;
 pub use empty::{empty, Empty};
+pub use extend::Extend;
 pub use from_stream::FromStream;
 pub use into_stream::IntoStream;
 pub use once::{once, Once};
@@ -31,6 +32,7 @@ pub use stream::{Fuse, Scan, Stream, Take, Zip};
 
 mod double_ended_stream;
 mod empty;
+mod extend;
 mod from_stream;
 mod into_stream;
 mod once;

src/vec/extend.rs

@@ -0,0 +1,19 @@
+use std::pin::Pin;
+
+use crate::future::Future;
+use crate::stream::{Extend, IntoStream, Stream};
+
+impl<T> Extend<T> for Vec<T> {
+    fn extend_with_stream<'a, S: IntoStream<Item = T> + 'a>(
+        &'a mut self,
+        stream: S,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+        let stream = stream.into_stream();
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+            while let Some(item) = stream.next().await {
+                self.push(item);
+            }
+        })
+    }
+}

src/vec/mod.rs

@@ -3,6 +3,7 @@
 //! This library provides smart pointers and collections for managing
 //! heap-allocated values.
 
+mod extend;
 mod from_stream;
 
 #[doc(inline)]