add crate::stream::Stream bound

src/result/from_stream.rs

@@ -1,47 +1,47 @@
-use crate::stream::{FromStream, IntoStream, Stream};
+// use crate::stream::{FromStream, IntoStream, Stream};
 
-use std::pin::Pin;
+// use std::pin::Pin;
 
-impl<T: Send, E: Send, V> FromStream<Result<T, E>> for Result<V, E>
-where
-    V: FromStream<T>,
-{
-    /// Takes each element in the stream: if it is an `Err`, no further
-    /// elements are taken, and the `Err` is returned. Should no `Err`
-    /// occur, a container with the values of each `Result` is returned.
-    #[inline]
-    fn from_stream<'a, S: IntoStream<Item = Result<T, E>>>(
-        stream: S,
-    ) -> Pin<Box<dyn core::future::Future<Output = Self> + Send + 'a>>
-    where
-        <S as IntoStream>::IntoStream: Send + 'a,
-    {
-        let stream = stream.into_stream();
+// impl<T: Send, E: Send, V> FromStream<Result<T, E>> for Result<V, E>
+// where
+//     V: FromStream<T>,
+// {
+//     /// Takes each element in the stream: if it is an `Err`, no further
+//     /// elements are taken, and the `Err` is returned. Should no `Err`
+//     /// occur, a container with the values of each `Result` is returned.
+//     #[inline]
+//     fn from_stream<'a, S: IntoStream<Item = Result<T, E>>>(
+//         stream: S,
+//     ) -> Pin<Box<dyn core::future::Future<Output = Self> + Send + 'a>>
+//     where
+//         <S as IntoStream>::IntoStream: Send + 'a,
+//     {
+//         let stream = stream.into_stream();
 
-        Pin::from(Box::new(async move {
-            pin_utils::pin_mut!(stream);
+//         Pin::from(Box::new(async move {
+//             pin_utils::pin_mut!(stream);
 
-            // Using `scan` here because it is able to stop the stream early
-            // if a failure occurs
-            let mut found_error = None;
-            let out: V = stream
-                .scan((), |_, elem| {
-                    match elem {
-                        Ok(elem) => Some(elem),
-                        Err(err) => {
-                            found_error = Some(err);
-                            // Stop processing the stream on error
-                            None
-                        }
-                    }
-                })
-                .collect()
-                .await;
+//             // Using `scan` here because it is able to stop the stream early
+//             // if a failure occurs
+//             let mut found_error = None;
+//             let out: V = stream
+//                 .scan((), |_, elem| {
+//                     match elem {
+//                         Ok(elem) => Some(elem),
+//                         Err(err) => {
+//                             found_error = Some(err);
+//                             // Stop processing the stream on error
+//                             None
+//                         }
+//                     }
+//                 })
+//                 .collect()
+//                 .await;
 
-            match found_error {
-                Some(err) => Err(err),
-                None => Ok(out),
-            }
-        }))
-    }
-}
+//             match found_error {
+//                 Some(err) => Err(err),
+//                 None => Ok(out),
+//             }
+//         }))
+//     }
+// }

src/stream/from_stream.rs

@@ -11,7 +11,7 @@ use std::pin::Pin;
 ///
 /// [`IntoStream`]: trait.IntoStream.html
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-pub trait FromStream<T: Send> {
+pub trait FromStream<T: Send + Unpin> {
     /// Creates a value from a stream.
     ///
     /// # Examples
@@ -23,7 +23,7 @@ pub trait FromStream<T: Send> {
     ///
     /// // 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> + Send + Unpin + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn core::future::Future<Output = Self> + Send + 'a>>;
+    ) -> Pin<Box<dyn core::future::Future<Output = Self> + Send + Unpin + 'a>>;
 }

src/stream/into_stream.rs

@@ -1,4 +1,4 @@
-use futures_core::stream::Stream;
+use crate::stream::Stream;
 
 /// Conversion into a `Stream`.
 ///

src/stream/stream/mod.rs

@@ -750,9 +750,9 @@ pub trait Stream {
     #[must_use = "if you really need to exhaust the iterator, consider `.for_each(drop)` instead (TODO)"]
     fn collect<'a, B>(self) -> dyn_ret!('a, B)
     where
-        Self: futures_core::stream::Stream + Sized + Send + 'a,
-        <Self as futures_core::stream::Stream>::Item: Send,
-        B: FromStream<<Self as futures_core::stream::Stream>::Item>,
+        Self: Stream + Sized + Send + Unpin + 'a,
+        <Self as Stream>::Item: Send + Unpin,
+        B: FromStream<<Self as crate::stream::Stream>::Item>,
     {
         FromStream::from_stream(self)
     }

src/vec/from_stream.rs

@@ -1,25 +1,26 @@
-use crate::stream::{FromStream, IntoStream, Stream};
+use crate::stream::{FromStream, IntoStream};
 
 use std::pin::Pin;
 
-impl<T: Send> FromStream<T> for Vec<T> {
+impl<T: Send + Unpin> 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> + Send + Unpin + 'a>>
     where
         <S as IntoStream>::IntoStream: Send + 'a,
     {
-        let stream = stream.into_stream();
+        let _stream = stream.into_stream();
 
-        Pin::from(Box::new(async move {
-            pin_utils::pin_mut!(stream);
+        // Box::pin(async move {
+        //     pin_utils::pin_mut!(stream);
 
-            let mut out = vec![];
-            while let Some(item) = stream.next().await {
-                out.push(item);
-            }
-            out
-        }))
+        //     let mut out = vec![];
+        //     while let Some(item) = stream.next().await {
+        //         out.push(item);
+        //     }
+        //     out
+        // })
+        panic!();
     }
 }

src/vec/into_stream.rs

@@ -7,7 +7,7 @@ pub struct IntoStream<T> {
     iter: std::vec::IntoIter<T>,
 }
 
-impl<T: Send> crate::stream::IntoStream for Vec<T> {
+impl<T: Send + Unpin> crate::stream::IntoStream for Vec<T> {
     type Item = T;
     type IntoStream = IntoStream<T>;
 
@@ -25,30 +25,30 @@ impl<T: Send> crate::stream::IntoStream for Vec<T> {
     /// }
     /// ```
     #[inline]
-    fn into_stream(mut self) -> IntoStream<T> {
+    fn into_stream(self) -> IntoStream<T> {
         let iter = self.into_iter();
         IntoStream { iter }
     }
 }
 
-impl<T: Send> futures_core::stream::Stream for IntoStream<T> {
+impl<T: Send + Unpin> crate::stream::Stream for IntoStream<T> {
     type Item = T;
 
     fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        Poll::Ready(self.iter.next())
+        Poll::Ready(Pin::new(&mut *self).iter.next())
     }
 }
 
 /// Slice stream.
 #[derive(Debug)]
-pub struct Stream<'a, T: 'a> {
+pub struct Stream<'a, T> {
     iter: std::slice::Iter<'a, T>,
 }
 
-impl<'a, T: Sync> futures_core::stream::Stream for Stream<'a, T> {
+impl<'a, T: Sync> crate::stream::Stream for Stream<'a, T> {
     type Item = &'a T;
 
-    fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         Poll::Ready(self.iter.next())
     }
 }
@@ -65,14 +65,14 @@ impl<'a, T: Sync> crate::stream::IntoStream for &'a Vec<T> {
 
 /// Mutable slice stream.
 #[derive(Debug)]
-pub struct StreamMut<'a, T: 'a> {
+pub struct StreamMut<'a, T> {
     iter: std::slice::IterMut<'a, T>,
 }
 
-impl<'a, T: Sync> futures_core::stream::Stream for StreamMut<'a, T> {
+impl<'a, T: Sync> crate::stream::Stream for StreamMut<'a, T> {
     type Item = &'a mut T;
 
-    fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         Poll::Ready(self.iter.next())
     }
 }