Rewrote `Result`s implementation using `take_while` and `filter_map`

pull/667/head
Oleg Nosov 5 years ago
parent ed248017b4
commit ed7ddacb28
No known key found for this signature in database
GPG Key ID: DE90B83800644E24

@ -17,26 +17,34 @@ where
let stream = stream.into_stream(); let stream = stream.into_stream();
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 is_error = false;
let mut found_error = None; let mut found_error = None;
let out: V = stream let out: V = stream
.scan(&mut found_error, |error, elem| { .take_while(|elem| {
match elem { // Stop processing the stream on `Err`
Ok(elem) => Some(elem), !is_error
&& (elem.is_ok() || {
is_error = true;
// Capture first `Err`
true
})
})
.filter_map(|elem| match elem {
Ok(value) => Some(value),
Err(err) => { Err(err) => {
**error = Some(err); found_error = Some(err);
// Stop processing the stream on error
None None
} }
}
}) })
.collect() .collect()
.await; .await;
match found_error { if is_error {
Some(err) => Err(err), Err(found_error.unwrap())
None => Ok(out), } else {
Ok(out)
} }
}) })
} }

@ -40,24 +40,35 @@ where
S: Stream<Item = Result<U, E>> + 'a, S: Stream<Item = Result<U, E>> + '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 is_error = false;
let mut found_error = None; let mut found_error = None;
let out = <T as Product<U>>::product(stream.scan(&mut found_error, |error, elem| { let out = <T as Product<U>>::product(
match elem { stream
Ok(elem) => Some(elem), .take_while(|elem| {
// Stop processing the stream on `Err`
!is_error
&& (elem.is_ok() || {
is_error = true;
// Capture first `Err`
true
})
})
.filter_map(|elem| match elem {
Ok(value) => Some(value),
Err(err) => { Err(err) => {
**error = Some(err); found_error = Some(err);
// Stop processing the stream on error
None None
} }
} }),
})) )
.await; .await;
match found_error { if is_error {
Some(err) => Err(err), Err(found_error.unwrap())
None => Ok(out), } else {
Ok(out)
} }
}) })
} }

@ -40,24 +40,35 @@ where
S: Stream<Item = Result<U, E>> + 'a, S: Stream<Item = Result<U, E>> + '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 is_error = false;
let mut found_error = None; let mut found_error = None;
let out = <T as Sum<U>>::sum(stream.scan(&mut found_error, |error, elem| { let out = <T as Sum<U>>::sum(
match elem { stream
Ok(elem) => Some(elem), .take_while(|elem| {
// Stop processing the stream on `Err`
!is_error
&& (elem.is_ok() || {
is_error = true;
// Capture first `Err`
true
})
})
.filter_map(|elem| match elem {
Ok(value) => Some(value),
Err(err) => { Err(err) => {
**error = Some(err); found_error = Some(err);
// Stop processing the stream on error
None None
} }
} }),
})) )
.await; .await;
match found_error { if is_error {
Some(err) => Err(err), Err(found_error.unwrap())
None => Ok(out), } else {
Ok(out)
} }
}) })
} }

Loading…
Cancel
Save