fix(stream): add send guards on collect

yoshuawuyts · dignifiedquire · web-flow · commit c82b1efb6927 · 2020-06-27T16:46:14.000+02:00
Closes #639 Co-authored-by: dignifiedquire <me@dignifiedquire.com>
diff --git a/src/collections/binary_heap/extend.rs b/src/collections/binary_heap/extend.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<T: Ord> stream::Extend<T> for BinaryHeap<T> {
+impl<T: Ord + Send> stream::Extend<T> for BinaryHeap<T> {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         self.reserve(stream.size_hint().0);
diff --git a/src/collections/binary_heap/from_stream.rs b/src/collections/binary_heap/from_stream.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<T: Ord> FromStream<T> for BinaryHeap<T> {
+impl<T: Ord + Send> FromStream<T> for BinaryHeap<T> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/collections/btree_map/extend.rs b/src/collections/btree_map/extend.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<K: Ord, V> stream::Extend<(K, V)> for BTreeMap<K, V> {
+impl<K: Ord + Send, V: Send> stream::Extend<(K, V)> for BTreeMap<K, V> {
     fn extend<'a, S: IntoStream<Item = (K, V)> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         Box::pin(stream.into_stream().for_each(move |(k, v)| {
             self.insert(k, v);
         }))
diff --git a/src/collections/btree_map/from_stream.rs b/src/collections/btree_map/from_stream.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<K: Ord, V> FromStream<(K, V)> for BTreeMap<K, V> {
+impl<K: Ord + Send, V: Send> FromStream<(K, V)> for BTreeMap<K, V> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = (K, V)> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/collections/btree_set/extend.rs b/src/collections/btree_set/extend.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<T: Ord> stream::Extend<T> for BTreeSet<T> {
+impl<T: Ord + Send> stream::Extend<T> for BTreeSet<T> {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         Box::pin(stream.into_stream().for_each(move |item| {
             self.insert(item);
         }))
diff --git a/src/collections/btree_set/from_stream.rs b/src/collections/btree_set/from_stream.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<T: Ord> FromStream<T> for BTreeSet<T> {
+impl<T: Ord + Send> FromStream<T> for BTreeSet<T> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/collections/hash_map/extend.rs b/src/collections/hash_map/extend.rs
@@ -7,13 +7,17 @@ use crate::stream::{self, IntoStream};
 
 impl<K, V, H> stream::Extend<(K, V)> for HashMap<K, V, H>
 where
-    K: Eq + Hash,
-    H: BuildHasher + Default,
+    K: Eq + Hash + Send,
+    V: Send,
+    H: BuildHasher + Default + Send,
 {
     fn extend<'a, S: IntoStream<Item = (K, V)> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         // The following is adapted from the hashbrown source code:
diff --git a/src/collections/hash_map/from_stream.rs b/src/collections/hash_map/from_stream.rs
@@ -7,13 +7,17 @@ use crate::stream::{self, FromStream, IntoStream};
 
 impl<K, V, H> FromStream<(K, V)> for HashMap<K, V, H>
 where
-    K: Eq + Hash,
-    H: BuildHasher + Default,
+    K: Eq + Hash + Send,
+    H: BuildHasher + Default + Send,
+    V: Send,
 {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = (K, V)> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/collections/hash_set/extend.rs b/src/collections/hash_set/extend.rs
@@ -7,13 +7,16 @@ use crate::stream::{self, IntoStream};
 
 impl<T, H> stream::Extend<T> for HashSet<T, H>
 where
-    T: Eq + Hash,
-    H: BuildHasher + Default,
+    T: Eq + Hash + Send,
+    H: BuildHasher + Default + Send,
 {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         // 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.
 
diff --git a/src/collections/hash_set/from_stream.rs b/src/collections/hash_set/from_stream.rs
@@ -7,13 +7,16 @@ use crate::stream::{self, FromStream, IntoStream};
 
 impl<T, H> FromStream<T> for HashSet<T, H>
 where
-    T: Eq + Hash,
-    H: BuildHasher + Default,
+    T: Eq + Hash + Send,
+    H: BuildHasher + Default + Send,
 {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/collections/linked_list/extend.rs b/src/collections/linked_list/extend.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<T> stream::Extend<T> for LinkedList<T> {
+impl<T: Send> stream::Extend<T> for LinkedList<T> {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
         Box::pin(stream.for_each(move |item| self.push_back(item)))
     }
diff --git a/src/collections/linked_list/from_stream.rs b/src/collections/linked_list/from_stream.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<T> FromStream<T> for LinkedList<T> {
+impl<T: Send> FromStream<T> for LinkedList<T> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/collections/vec_deque/extend.rs b/src/collections/vec_deque/extend.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<T> stream::Extend<T> for VecDeque<T> {
+impl<T: Send> stream::Extend<T> for VecDeque<T> {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         self.reserve(stream.size_hint().0);
diff --git a/src/collections/vec_deque/from_stream.rs b/src/collections/vec_deque/from_stream.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<T> FromStream<T> for VecDeque<T> {
+impl<T: Send> FromStream<T> for VecDeque<T> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/option/from_stream.rs b/src/option/from_stream.rs
@@ -4,7 +4,7 @@ use crate::prelude::*;
 use crate::stream::{FromStream, IntoStream};
 use std::convert::identity;
 
-impl<T, V> FromStream<Option<T>> for Option<V>
+impl<T: Send, V> FromStream<Option<T>> for Option<V>
 where
     V: FromStream<T>,
 {
@@ -14,7 +14,10 @@ where
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = Option<T>> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/path/pathbuf.rs b/src/path/pathbuf.rs
@@ -323,7 +323,10 @@ impl<P: AsRef<Path>> stream::Extend<P> for PathBuf {
     fn extend<'a, S: IntoStream<Item = P> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -337,11 +340,14 @@ impl<P: AsRef<Path>> stream::Extend<P> for PathBuf {
 }
 
 #[cfg(feature = "unstable")]
-impl<'b, P: AsRef<Path> + 'b> FromStream<P> for PathBuf {
+impl<'b, P: AsRef<Path> + 'b + Send> FromStream<P> for PathBuf {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = P> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/result/from_stream.rs b/src/result/from_stream.rs
@@ -5,6 +5,8 @@ use crate::stream::{FromStream, IntoStream};
 
 impl<T, E, V> FromStream<Result<T, E>> for Result<V, E>
 where
+    T: Send,
+    E: Send,
     V: FromStream<T>,
 {
     /// Takes each element in the stream: if it is an `Err`, no further
@@ -30,7 +32,10 @@ where
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = Result<T, E>> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
diff --git a/src/stream/extend.rs b/src/stream/extend.rs
@@ -34,7 +34,9 @@ pub trait Extend<A> {
     fn extend<'a, T: IntoStream<Item = A> + 'a>(
         &'a mut self,
         stream: T,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>>;
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <T as IntoStream>::IntoStream: Send;
 }
 
 /// Extends a collection with the contents of a stream.
@@ -69,6 +71,7 @@ pub async fn extend<'a, C, T, S>(collection: &mut C, stream: S)
 where
     C: Extend<T>,
     S: IntoStream<Item = T> + 'a,
+    <S as IntoStream>::IntoStream: Send,
 {
     Extend::extend(collection, stream).await
 }
diff --git a/src/stream/from_stream.rs b/src/stream/from_stream.rs
@@ -72,7 +72,10 @@ use crate::stream::IntoStream;
 /// impl FromStream<i32> for MyCollection {
 ///     fn from_stream<'a, S: IntoStream<Item = i32> + 'a>(
 ///         stream: S,
-///     ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+///     ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
+///    where
+///        <S as IntoStream>::IntoStream: Send,
+///    {
 ///         let stream = stream.into_stream();
 ///
 ///         Box::pin(async move {
@@ -107,12 +110,12 @@ use crate::stream::IntoStream;
 /// assert_eq!(c.0, vec![5, 5, 5, 5, 5]);
 /// #
 /// # Ok(()) }) }
-///```
+/// ```
 ///
 /// [`IntoStream`]: trait.IntoStream.html
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-pub trait FromStream<T> {
+pub trait FromStream<T: Send> {
     /// Creates a value from a stream.
     ///
     /// # Examples
@@ -135,5 +138,7 @@ pub trait FromStream<T> {
     /// ```
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>>;
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send;
 }
diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
@@ -1888,10 +1888,11 @@ extension_trait! {
         #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
         fn collect<'a, B>(
             self,
-        ) -> impl Future<Output = B> + 'a [Pin<Box<dyn Future<Output = B> + 'a>>]
+        ) -> impl Future<Output = B> + 'a [Pin<Box<dyn Future<Output = B> + 'a + Send>>]
         where
-            Self: Sized + 'a,
+            Self: Sized + 'a + Send,
             B: FromStream<Self::Item>,
+            Self::Item: Send,
         {
             FromStream::from_stream(self)
         }
diff --git a/src/string/extend.rs b/src/string/extend.rs
diff --git a/src/string/from_stream.rs b/src/string/from_stream.rs
diff --git a/src/unit/extend.rs b/src/unit/extend.rs
diff --git a/src/unit/from_stream.rs b/src/unit/from_stream.rs
diff --git a/src/vec/extend.rs b/src/vec/extend.rs
diff --git a/src/vec/from_stream.rs b/src/vec/from_stream.rs
diff --git a/tests/collect.rs b/tests/collect.rs
