add crate::stream::Stream bound

yoshuawuyts · yoshuawuyts · commit d9a6b39f37a9 · 2019-09-19T13:11:08.000+02:00
Signed-off-by: Yoshua Wuyts &lt;yoshuawuyts@gmail.com&gt;
diff --git a/src/result/from_stream.rs b/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),
+//             }
+//         }))
+//     }
+// }
diff --git a/src/stream/from_stream.rs b/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>>;
 }
diff --git a/src/stream/into_stream.rs b/src/stream/into_stream.rs
@@ -1,4 +1,4 @@
-use futures_core::stream::Stream;
+use crate::stream::Stream;
 
 /// Conversion into a `Stream`.
 ///
diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
@@ -750,18 +750,18 @@ 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)
     }
 }
 
-impl<T: futures_core::stream::Stream + ?Sized> Stream for T {
-    type Item = <Self as futures_core::stream::Stream>::Item;
+// impl<T: futures_core::stream::Stream + ?Sized> Stream for T {
+//     type Item = <Self as futures_core::stream::Stream>::Item;
 
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        futures_core::stream::Stream::poll_next(self, cx)
-    }
-}
+//     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+//         futures_core::stream::Stream::poll_next(self, cx)
+//     }
+// }
diff --git a/src/vec/from_stream.rs b/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!();
     }
 }
diff --git a/src/vec/into_stream.rs b/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())
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-use futures_core::stream::Stream;`
	`1`	`+use crate::stream::Stream;`
`2`	`2`
`3`	`3`	/// Conversion into a `Stream`.
`4`	`4`	`///`
Original file line number	Diff line number	Diff line change
`@@ -7,7 +7,7 @@ pub struct IntoStream<T> {`
`7`	`7`	`iter: std::vec::IntoIter<T>,`
`8`	`8`	`}`
`9`	`9`
`10`		`-impl<T: Send> crate::stream::IntoStream for Vec<T> {`
	`10`	`+impl<T: Send + Unpin> crate::stream::IntoStream for Vec<T> {`
`11`	`11`	`type Item = T;`
`12`	`12`	`type IntoStream = IntoStream<T>;`
`13`	`13`
`@@ -25,30 +25,30 @@ impl<T: Send> crate::stream::IntoStream for Vec<T> {`
`25`	`25`	`/// }`
`26`	`26`	/// ```
`27`	`27`	`#[inline]`
`28`		`- fn into_stream(mut self) -> IntoStream<T> {`
	`28`	`+ fn into_stream(self) -> IntoStream<T> {`
`29`	`29`	`let iter = self.into_iter();`
`30`	`30`	`IntoStream { iter }`
`31`	`31`	`}`
`32`	`32`	`}`
`33`	`33`
`34`		`-impl<T: Send> futures_core::stream::Stream for IntoStream<T> {`
	`34`	`+impl<T: Send + Unpin> crate::stream::Stream for IntoStream<T> {`
`35`	`35`	`type Item = T;`
`36`	`36`
`37`	`37`	`fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {`
`38`		`- Poll::Ready(self.iter.next())`
	`38`	`+ Poll::Ready(Pin::new(&mut *self).iter.next())`
`39`	`39`	`}`
`40`	`40`	`}`
`41`	`41`
`42`	`42`	`/// Slice stream.`
`43`	`43`	`#[derive(Debug)]`
`44`		`-pub struct Stream<'a, T: 'a> {`
	`44`	`+pub struct Stream<'a, T> {`
`45`	`45`	`iter: std::slice::Iter<'a, T>,`
`46`	`46`	`}`
`47`	`47`
`48`		`-impl<'a, T: Sync> futures_core::stream::Stream for Stream<'a, T> {`
	`48`	`+impl<'a, T: Sync> crate::stream::Stream for Stream<'a, T> {`
`49`	`49`	`type Item = &'a T;`
`50`	`50`
`51`		`- fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {`
	`51`	`+ fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {`
`52`	`52`	`Poll::Ready(self.iter.next())`
`53`	`53`	`}`
`54`	`54`	`}`
`@@ -65,14 +65,14 @@ impl<'a, T: Sync> crate::stream::IntoStream for &'a Vec<T> {`
`65`	`65`
`66`	`66`	`/// Mutable slice stream.`
`67`	`67`	`#[derive(Debug)]`
`68`		`-pub struct StreamMut<'a, T: 'a> {`
	`68`	`+pub struct StreamMut<'a, T> {`
`69`	`69`	`iter: std::slice::IterMut<'a, T>,`
`70`	`70`	`}`
`71`	`71`
`72`		`-impl<'a, T: Sync> futures_core::stream::Stream for StreamMut<'a, T> {`
	`72`	`+impl<'a, T: Sync> crate::stream::Stream for StreamMut<'a, T> {`
`73`	`73`	`type Item = &'a mut T;`
`74`	`74`
`75`		`- fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {`
	`75`	`+ fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {`
`76`	`76`	`Poll::Ready(self.iter.next())`
`77`	`77`	`}`
`78`	`78`	`}`