Merge pull request #1 from stjepang/into_stream_trait

Remove some Unpin and Send bounds

Author: yoshuawuyts

src/stream/from_stream.rs

@@ -10,7 +10,7 @@ use std::pin::Pin;
 /// See also: [`IntoStream`].
 ///
 /// [`IntoStream`]: trait.IntoStream.html
-pub trait FromStream<T>: Sized + Unpin {
+pub trait FromStream<T> {
     /// Creates a value from a stream.
     ///
     /// # Examples
@@ -22,7 +22,7 @@ pub trait FromStream<T>: Sized + Unpin {
     ///
     /// // let _five_fives = async_std::stream::repeat(5).take(5);
     /// ```
-    fn from_stream<'a, S: IntoStream<Item = T> + Send + 'a>(
+    fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn core::future::Future<Output = Self> + Send + 'a>>;
+    ) -> Pin<Box<dyn core::future::Future<Output = Self> + 'a>>;
 }

src/stream/into_stream.rs

@@ -18,13 +18,13 @@ pub trait IntoStream {
     type Item;
 
     /// Which kind of stream are we turning this into?
-    type IntoStream: Stream<Item = Self::Item> + Unpin + Send;
+    type IntoStream: Stream<Item = Self::Item>;
 
     /// Creates a stream from a value.
     fn into_stream(self) -> Self::IntoStream;
 }
 
-impl<I: Stream + Unpin + Send> IntoStream for I {
+impl<I: Stream> IntoStream for I {
     type Item = I::Item;
     type IntoStream = I;
 

src/stream/stream.rs

@@ -54,7 +54,7 @@ cfg_if! {
         }
     } else {
         macro_rules! dyn_ret {
-            ($a:lifetime, $o:ty) => (Pin<Box<dyn core::future::Future<Output = $o> + Send + 'a>>)
+            ($a:lifetime, $o:ty) => (Pin<Box<dyn core::future::Future<Output = $o> + 'a>>)
         }
     }
 }
@@ -162,9 +162,9 @@ pub trait Stream {
     ///
     /// [`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: FromStream<<Self as Stream>::Item>>(self) -> dyn_ret!('a, B)
+    fn collect<'a, B>(self) -> dyn_ret!('a, B)
     where
-        Self: futures::stream::Stream + Sized + Send + Unpin + 'a,
+        Self: futures::stream::Stream + Sized + 'a,
         B: FromStream<<Self as futures::stream::Stream>::Item>,
     {
         FromStream::from_stream(self)

src/vec/from_stream.rs

@@ -2,16 +2,19 @@ use crate::stream::{FromStream, IntoStream, Stream};
 
 use std::pin::Pin;
 
-impl<T: Unpin + Sized + Send> FromStream<T> for Vec<T> {
+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> + Send + 'a>>
+    ) -> Pin<Box<dyn core::future::Future<Output = Self> + 'a>>
     where
-        <S as IntoStream>::IntoStream: Send + 'a,
+        <S as IntoStream>::IntoStream: 'a,
     {
-        let mut stream = stream.into_stream();
+        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);