async_std/option/
product.rs

1use std::pin::Pin;
2
3use crate::prelude::*;
4use crate::stream::{Product, Stream};
5use std::convert::identity;
6
7impl<T, U> Product<Option<U>> for Option<T>
8where
9    T: Product<U>,
10{
11    #[doc = r#"
12        Takes each element in the `Stream`: if it is a `None`, no further
13        elements are taken, and the `None` is returned. Should no `None` occur,
14        the product of all elements is returned.
15
16        # Examples
17
18        This multiplies every integer in a vector, rejecting the product if a negative element is
19        encountered:
20
21        ```
22        # fn main() { async_std::task::block_on(async {
23        #
24        use async_std::prelude::*;
25        use async_std::stream;
26
27        let v = stream::from_iter(vec![1, 2, 4]);
28        let prod: Option<i32> = v.map(|x|
29            if x < 0 {
30                None
31            } else {
32                Some(x)
33            }).product().await;
34        assert_eq!(prod, Some(8));
35        #
36        # }) }
37        ```
38    "#]
39    fn product<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Option<T>> + 'a>>
40    where
41        S: Stream<Item = Option<U>> + 'a,
42    {
43        Box::pin(async move {
44            // Using `take_while` here because it is able to stop the stream early
45            // if a failure occurs
46            let mut found_none = false;
47            let out = <T as Product<U>>::product(
48                stream
49                    .take_while(|elem| {
50                        elem.is_some() || {
51                            found_none = true;
52                            // Stop processing the stream on `None`
53                            false
54                        }
55                    })
56                    .filter_map(identity),
57            )
58            .await;
59
60            if found_none { None } else { Some(out) }
61        })
62    }
63}
async_std/option/product.rs

async_std/option/
product.rs
