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>

Author: bors[bot]

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);
+        }))
+    }
+}