|
|
@ -1,7 +1,7 @@
|
|
|
|
use std::pin::Pin;
|
|
|
|
use std::pin::Pin;
|
|
|
|
|
|
|
|
|
|
|
|
use crate::prelude::*;
|
|
|
|
use crate::prelude::*;
|
|
|
|
use crate::stream::{Stream, Product};
|
|
|
|
use crate::stream::{Product, Stream};
|
|
|
|
|
|
|
|
|
|
|
|
impl<T, U> Product<Option<U>> for Option<T>
|
|
|
|
impl<T, U> Product<Option<U>> for Option<T>
|
|
|
|
where
|
|
|
|
where
|
|
|
@ -36,23 +36,24 @@ where
|
|
|
|
```
|
|
|
|
```
|
|
|
|
"#]
|
|
|
|
"#]
|
|
|
|
fn product<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Option<T>> + 'a>>
|
|
|
|
fn product<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Option<T>> + 'a>>
|
|
|
|
where S: Stream<Item = Option<U>> + 'a
|
|
|
|
where
|
|
|
|
|
|
|
|
S: Stream<Item = Option<U>> + 'a,
|
|
|
|
{
|
|
|
|
{
|
|
|
|
Box::pin(async move {
|
|
|
|
Box::pin(async move {
|
|
|
|
// Using `scan` here because it is able to stop the stream early
|
|
|
|
// Using `take_while` here because it is able to stop the stream early
|
|
|
|
// if a failure occurs
|
|
|
|
// if a failure occurs
|
|
|
|
let mut found_none = false;
|
|
|
|
let mut found_none = false;
|
|
|
|
let out = <T as Product<U>>::product(stream
|
|
|
|
let out = <T as Product<U>>::product(
|
|
|
|
.scan((), |_, elem| {
|
|
|
|
stream
|
|
|
|
match elem {
|
|
|
|
.take_while(|elem| {
|
|
|
|
Some(elem) => Some(elem),
|
|
|
|
elem.is_some() || {
|
|
|
|
None => {
|
|
|
|
|
|
|
|
found_none = true;
|
|
|
|
found_none = true;
|
|
|
|
// Stop processing the stream on error
|
|
|
|
false
|
|
|
|
None
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
})).await;
|
|
|
|
})
|
|
|
|
|
|
|
|
.map(Option::unwrap),
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
.await;
|
|
|
|
|
|
|
|
|
|
|
|
if found_none {
|
|
|
|
if found_none {
|
|
|
|
None
|
|
|
|
None
|
|
|
|