FromStream for Option<T>

async-rs · yoshuawuyts · Oct 1, 2019 · Sep 29, 2019 · Sep 29, 2019 · Sep 29, 2019
commit 76b10c4784fc109e1574fa7874975e2a9ed80888
diff --git a/src/lib.rs b/src/lib.rs
@@ -67,6 +67,7 @@ cfg_if! {
 
         mod vec;
         mod result;
+        mod option;
     }
 }
 

diff --git a/src/option/from_stream.rs b/src/option/from_stream.rs
@@ -0,0 +1,49 @@
+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();
+
+        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 = 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)
+            }
+        }))
+    }
+}
diff --git a/src/option/mod.rs b/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;
-Original file line number
+Diff line change
@@ Expand Up / @@ -67,6 +67,7 @@ cfg_if! { @@
             mod vec;
             mod result;
+            mod option;
         }
     }
@@ Expand Down @@