Merge pull request #265 from sunjay/fromstream-option-vec

FromStream impl for Option<T> + Revised impl for Vec<T>

Author: yoshuawuyts

src/lib.rs

@@ -67,6 +67,7 @@ cfg_if! {
 
         mod vec;
         mod result;
+        mod option;
     }
 }
 

src/option/from_stream.rs

@@ -0,0 +1,45 @@
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::{FromStream, IntoStream};
+
+impl<T, V> FromStream<Option<T>> for Option<V>
+where
+    V: FromStream<T>,
+{
+    /// Takes each element in the stream: if it is `None`, no further
+    /// elements are taken, and `None` is returned. Should no `None`
+    /// occur, a container with the values of each `Option` is returned.
+    #[inline]
+    fn from_stream<'a, S: IntoStream<Item = Option<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);
+
+            // Using `scan` here because it is able to stop the stream early
+            // if a failure occurs
+            let mut found_error = false;
+            let out: V = stream
+                .scan((), |_, elem| {
+                    match elem {
+                        Some(elem) => Some(elem),
+                        None => {
+                            found_error = true;
+                            // Stop processing the stream on error
+                            None
+                        }
+                    }
+                })
+                .collect()
+                .await;
+
+            if found_error { None } else { Some(out) }
+        })
+    }
+}

src/option/mod.rs

@@ -0,0 +1,9 @@
+//! The Rust core optional value type
+//!
+//! This module provides the `Option<T>` type for returning and
+//! propagating optional values.
+
+mod from_stream;
+
+#[doc(inline)]
+pub use std::option::Option;

src/result/from_stream.rs

@@ -19,7 +19,7 @@ where
     {
         let stream = stream.into_stream();
 
-        Pin::from(Box::new(async move {
+        Box::pin(async move {
             pin_utils::pin_mut!(stream);
 
             // Using `scan` here because it is able to stop the stream early
@@ -43,6 +43,6 @@ where
                 Some(err) => Err(err),
                 None => Ok(out),
             }
-        }))
+        })
     }
 }

src/vec/from_stream.rs

@@ -1,7 +1,6 @@
 use std::pin::Pin;
 
-use crate::prelude::*;
-use crate::stream::{FromStream, IntoStream};
+use crate::stream::{Extend, FromStream, IntoStream};
 
 impl<T> FromStream<T> for Vec<T> {
     #[inline]
@@ -13,14 +12,12 @@ impl<T> FromStream<T> for Vec<T> {
     {
         let stream = stream.into_stream();
 
-        Pin::from(Box::new(async move {
+        Box::pin(async move {
             pin_utils::pin_mut!(stream);
 
             let mut out = vec![];
-            while let Some(item) = stream.next().await {
-                out.push(item);
-            }
+            out.stream_extend(stream).await;
             out
-        }))
+        })
     }
 }