Merge #271

271: FromStream impls for collections (and more!) r=yoshuawuyts a=sunjay

Just opening this to have some visibility on my work as I finish it off. Hopefully will be done in the next day or two, but if not, this is here for someone else to finish it off.

I'm currently in the process of adding the `FromStream` impls for all the collections. This is generally a very easy and repetitive process:

1. Look up the impl of `FromIterator` for the given collection, it probably uses the `Extend` trait which is also implemented for that collection
2. Copy and paste the directory for the collection that is closest to the collection you're currently doing (closest in terms of the type parameters needed)
3. Update the `Extend` impl to be for the collection you're implementing, being careful to use the `reserve` method if the collection has one to avoid allocating too many times
4. Update the `FromStream` impl to be for the collection you're implementing
5. Make sure you update the docs in the copied `mod.rs` and that you've updated `collections/mod.rs`
6. Test with `--features unstable` or your code will not be compiled

The majority of this work is just looking at what `std` does and adapting it to streams. Honestly it's kind of relaxing after a long day... (maybe I'm weird!) 😄

Co-authored-by: Sunjay Varma <varma.sunjay@gmail.com>

src/collections/binary_heap/extend.rs

@@ -0,0 +1,18 @@
+use std::collections::BinaryHeap;
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::{Extend, IntoStream};
+
+impl<T: Ord> Extend<T> for BinaryHeap<T> {
+    fn stream_extend<'a, S: IntoStream<Item = T> + 'a>(
+        &'a mut self,
+        stream: S,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+        let stream = stream.into_stream();
+        //TODO: Add this back in when size_hint is added to Stream/StreamExt
+        //let (lower_bound, _) = stream.size_hint();
+        //self.reserve(lower_bound);
+        Box::pin(stream.for_each(move |item| self.push(item)))
+    }
+}

src/collections/binary_heap/from_stream.rs

@@ -0,0 +1,24 @@
+use std::collections::BinaryHeap;
+use std::pin::Pin;
+
+use crate::stream::{Extend, FromStream, IntoStream};
+
+impl<T: Ord> FromStream<T> for BinaryHeap<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();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            let mut out = BinaryHeap::new();
+            out.stream_extend(stream).await;
+            out
+        })
+    }
+}

src/collections/binary_heap/mod.rs

@@ -0,0 +1,7 @@
+//! The Rust priority queue implemented with a binary heap
+
+mod extend;
+mod from_stream;
+
+#[doc(inline)]
+pub use std::collections::BinaryHeap;

src/collections/btree_map/extend.rs

@@ -0,0 +1,16 @@
+use std::collections::BTreeMap;
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::{Extend, IntoStream};
+
+impl<K: Ord, V> Extend<(K, V)> for BTreeMap<K, V> {
+    fn stream_extend<'a, S: IntoStream<Item = (K, V)> + 'a>(
+        &'a mut self,
+        stream: S,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+        Box::pin(stream.into_stream().for_each(move |(k, v)| {
+            self.insert(k, v);
+        }))
+    }
+}

src/collections/btree_map/from_stream.rs

@@ -0,0 +1,24 @@
+use std::collections::BTreeMap;
+use std::pin::Pin;
+
+use crate::stream::{Extend, FromStream, IntoStream};
+
+impl<K: Ord, V> FromStream<(K, V)> for BTreeMap<K, V> {
+    #[inline]
+    fn from_stream<'a, S: IntoStream<Item = (K, V)>>(
+        stream: S,
+    ) -> Pin<Box<dyn core::future::Future<Output = Self> + 'a>>
+    where
+        <S as IntoStream>::IntoStream: 'a,
+    {
+        let stream = stream.into_stream();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            let mut out = BTreeMap::new();
+            out.stream_extend(stream).await;
+            out
+        })
+    }
+}

src/collections/btree_map/mod.rs

@@ -0,0 +1,7 @@
+//! The Rust B-Tree Map
+
+mod extend;
+mod from_stream;
+
+#[doc(inline)]
+pub use std::collections::BTreeMap;

src/collections/btree_set/extend.rs

@@ -0,0 +1,16 @@
+use std::collections::BTreeSet;
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::{Extend, IntoStream};
+
+impl<T: Ord> Extend<T> for BTreeSet<T> {
+    fn stream_extend<'a, S: IntoStream<Item = T> + 'a>(
+        &'a mut self,
+        stream: S,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+        Box::pin(stream.into_stream().for_each(move |item| {
+            self.insert(item);
+        }))
+    }
+}

src/collections/btree_set/from_stream.rs

@@ -0,0 +1,24 @@
+use std::collections::BTreeSet;
+use std::pin::Pin;
+
+use crate::stream::{Extend, FromStream, IntoStream};
+
+impl<T: Ord> FromStream<T> for BTreeSet<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();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            let mut out = BTreeSet::new();
+            out.stream_extend(stream).await;
+            out
+        })
+    }
+}

src/collections/btree_set/mod.rs

@@ -0,0 +1,7 @@
+//! The Rust B-Tree Set
+
+mod extend;
+mod from_stream;
+
+#[doc(inline)]
+pub use std::collections::BTreeSet;

src/collections/hash_map/extend.rs

@@ -0,0 +1,38 @@
+use std::collections::HashMap;
+use std::hash::{BuildHasher, Hash};
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::{Extend, IntoStream};
+
+impl<K, V, H> Extend<(K, V)> for HashMap<K, V, H>
+where
+    K: Eq + Hash,
+    H: BuildHasher + Default,
+{
+    fn stream_extend<'a, S: IntoStream<Item = (K, V)> + 'a>(
+        &'a mut self,
+        stream: S,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+        let stream = stream.into_stream();
+
+        // The following is adapted from the hashbrown source code:
+        // https://github.com/rust-lang/hashbrown/blob/d1ad4fc3aae2ade446738eea512e50b9e863dd0c/src/map.rs#L2470-L2491
+        //
+        // Keys may be already present or show multiple times in the stream. Reserve the entire
+        // hint lower bound if the map is empty. Otherwise reserve half the hint (rounded up), so
+        // the map will only resize twice in the worst case.
+
+        //TODO: Add this back in when size_hint is added to Stream/StreamExt
+        //let reserve = if self.is_empty() {
+        //    stream.size_hint().0
+        //} else {
+        //    (stream.size_hint().0 + 1) / 2
+        //};
+        //self.reserve(reserve);
+
+        Box::pin(stream.for_each(move |(k, v)| {
+            self.insert(k, v);
+        }))
+    }
+}

src/collections/hash_map/from_stream.rs

@@ -0,0 +1,29 @@
+use std::collections::HashMap;
+use std::hash::{BuildHasher, Hash};
+use std::pin::Pin;
+
+use crate::stream::{Extend, FromStream, IntoStream};
+
+impl<K, V, H> FromStream<(K, V)> for HashMap<K, V, H>
+where
+    K: Eq + Hash,
+    H: BuildHasher + Default,
+{
+    #[inline]
+    fn from_stream<'a, S: IntoStream<Item = (K, V)>>(
+        stream: S,
+    ) -> Pin<Box<dyn core::future::Future<Output = Self> + 'a>>
+    where
+        <S as IntoStream>::IntoStream: 'a,
+    {
+        let stream = stream.into_stream();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            let mut out = HashMap::with_hasher(Default::default());
+            out.stream_extend(stream).await;
+            out
+        })
+    }
+}

src/collections/hash_map/mod.rs

@@ -0,0 +1,7 @@
+//! The Rust hash map, implemented with quadratic probing and SIMD lookup.
+
+mod extend;
+mod from_stream;
+
+#[doc(inline)]
+pub use std::collections::HashMap;

src/collections/hash_set/extend.rs

@@ -0,0 +1,41 @@
+use std::collections::HashSet;
+use std::hash::{BuildHasher, Hash};
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::{Extend, IntoStream};
+
+impl<T, H> Extend<T> for HashSet<T, H>
+where
+    T: Eq + Hash,
+    H: BuildHasher + Default,
+{
+    fn stream_extend<'a, S: IntoStream<Item = T> + 'a>(
+        &'a mut self,
+        stream: S,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+        // The Extend impl for HashSet in the standard library delegates to the internal HashMap.
+        // Thus, this impl is just a copy of the async Extend impl for HashMap in this crate.
+
+        let stream = stream.into_stream();
+
+        // The following is adapted from the hashbrown source code:
+        // https://github.com/rust-lang/hashbrown/blob/d1ad4fc3aae2ade446738eea512e50b9e863dd0c/src/map.rs#L2470-L2491
+        //
+        // Keys may be already present or show multiple times in the stream. Reserve the entire
+        // hint lower bound if the map is empty. Otherwise reserve half the hint (rounded up), so
+        // the map will only resize twice in the worst case.
+
+        //TODO: Add this back in when size_hint is added to Stream/StreamExt
+        //let reserve = if self.is_empty() {
+        //    stream.size_hint().0
+        //} else {
+        //    (stream.size_hint().0 + 1) / 2
+        //};
+        //self.reserve(reserve);
+
+        Box::pin(stream.for_each(move |item| {
+            self.insert(item);
+        }))
+    }
+}

src/collections/hash_set/from_stream.rs

@@ -0,0 +1,29 @@
+use std::collections::HashSet;
+use std::hash::{BuildHasher, Hash};
+use std::pin::Pin;
+
+use crate::stream::{Extend, FromStream, IntoStream};
+
+impl<T, H> FromStream<T> for HashSet<T, H>
+where
+    T: Eq + Hash,
+    H: BuildHasher + Default,
+{
+    #[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();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            let mut out = HashSet::with_hasher(Default::default());
+            out.stream_extend(stream).await;
+            out
+        })
+    }
+}

src/collections/hash_set/mod.rs

@@ -0,0 +1,7 @@
+//! The Rust hash set, implemented as a `HashMap` where the value is `()`.
+
+mod extend;
+mod from_stream;
+
+#[doc(inline)]
+pub use std::collections::HashSet;

src/collections/linked_list/extend.rs

@@ -0,0 +1,18 @@
+use std::collections::LinkedList;
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::{Extend, IntoStream};
+
+impl<T> Extend<T> for LinkedList<T> {
+    fn stream_extend<'a, S: IntoStream<Item = T> + 'a>(
+        &'a mut self,
+        stream: S,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+        let stream = stream.into_stream();
+        //TODO: Add this back in when size_hint is added to Stream/StreamExt
+        //let (lower_bound, _) = stream.size_hint();
+        //self.reserve(lower_bound);
+        Box::pin(stream.for_each(move |item| self.push_back(item)))
+    }
+}

src/collections/linked_list/from_stream.rs

@@ -0,0 +1,24 @@
+use std::collections::LinkedList;
+use std::pin::Pin;
+
+use crate::stream::{Extend, FromStream, IntoStream};
+
+impl<T> FromStream<T> for LinkedList<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();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            let mut out = LinkedList::new();
+            out.stream_extend(stream).await;
+            out
+        })
+    }
+}

src/collections/linked_list/mod.rs

@@ -0,0 +1,7 @@
+//! The Rust doubly-linked list with owned nodes
+
+mod extend;
+mod from_stream;
+
+#[doc(inline)]
+pub use std::collections::LinkedList;

src/collections/mod.rs

@@ -0,0 +1,20 @@
+//! The Rust standard collections
+//!
+//! This library provides efficient implementations of the most common general purpose programming
+//! data structures.
+
+pub mod binary_heap;
+pub mod btree_map;
+pub mod btree_set;
+pub mod hash_map;
+pub mod hash_set;
+pub mod linked_list;
+pub mod vec_deque;
+
+pub use binary_heap::BinaryHeap;
+pub use btree_map::BTreeMap;
+pub use btree_set::BTreeSet;
+pub use hash_map::HashMap;
+pub use hash_set::HashSet;
+pub use linked_list::LinkedList;
+pub use vec_deque::VecDeque;

src/collections/vec_deque/extend.rs

@@ -0,0 +1,18 @@
+use std::collections::VecDeque;
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::{Extend, IntoStream};
+
+impl<T> Extend<T> for VecDeque<T> {
+    fn stream_extend<'a, S: IntoStream<Item = T> + 'a>(
+        &'a mut self,
+        stream: S,
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+        let stream = stream.into_stream();
+        //TODO: Add this back in when size_hint is added to Stream/StreamExt
+        //let (lower_bound, _) = stream.size_hint();
+        //self.reserve(lower_bound);
+        Box::pin(stream.for_each(move |item| self.push_back(item)))
+    }
+}

src/collections/vec_deque/from_stream.rs

@@ -0,0 +1,24 @@
+use std::collections::VecDeque;
+use std::pin::Pin;
+
+use crate::stream::{Extend, FromStream, IntoStream};
+
+impl<T> FromStream<T> for VecDeque<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();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            let mut out = VecDeque::new();
+            out.stream_extend(stream).await;
+            out
+        })
+    }
+}

src/collections/vec_deque/mod.rs

@@ -0,0 +1,7 @@
+//! The Rust double-ended queue, implemented with a growable ring buffer.
+
+mod extend;
+mod from_stream;
+
+#[doc(inline)]
+pub use std::collections::VecDeque;

src/lib.rs

@@ -65,10 +65,12 @@ cfg_if! {
         #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
         pub mod pin;
 
+        mod unit;
         mod vec;
         mod result;
         mod option;
         mod string;
+        mod collections;
     }
 }
 

src/unit/from_stream.rs

@@ -0,0 +1,16 @@
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::{FromStream, IntoStream};
+
+impl FromStream<()> for () {
+    #[inline]
+    fn from_stream<'a, S: IntoStream<Item = ()>>(
+        stream: S,
+    ) -> Pin<Box<dyn core::future::Future<Output = Self> + 'a>>
+    where
+        <S as IntoStream>::IntoStream: 'a,
+    {
+        Box::pin(stream.into_stream().for_each(|_| ()))
+    }
+}

src/unit/mod.rs

@@ -0,0 +1,6 @@
+//! The Rust primitive `()` type, sometimes called "unit" or "nil".
+//!
+//! This module provides types and implementations for working
+//! asynchronously with values of type `()`.
+
+mod from_stream;

src/vec/extend.rs

@@ -9,11 +9,9 @@ impl<T> Extend<T> for Vec<T> {
         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);
-            }
-        })
+        //TODO: Add this back in when size_hint is added to Stream/StreamExt
+        //let (lower_bound, _) = stream.size_hint();
+        //self.reserve(lower_bound);
+        Box::pin(stream.for_each(move |item| self.push(item)))
     }
 }

src/vec/from_stream.rs

@@ -1,4 +1,7 @@
+use std::borrow::Cow;
 use std::pin::Pin;
+use std::rc::Rc;
+use std::sync::Arc;
 
 use crate::stream::{Extend, FromStream, IntoStream};
 
@@ -21,3 +24,75 @@ impl<T> FromStream<T> for Vec<T> {
         })
     }
 }
+
+impl<'b, T: Clone> FromStream<T> for Cow<'b, [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();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            Cow::Owned(FromStream::from_stream(stream).await)
+        })
+    }
+}
+
+impl<T> FromStream<T> for Box<[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();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            Vec::from_stream(stream).await.into_boxed_slice()
+        })
+    }
+}
+
+impl<T> FromStream<T> for Rc<[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();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            Vec::from_stream(stream).await.into()
+        })
+    }
+}
+
+impl<T> FromStream<T> for Arc<[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();
+
+        Box::pin(async move {
+            pin_utils::pin_mut!(stream);
+
+            Vec::from_stream(stream).await.into()
+        })
+    }
+}