forked from mirror/async-std
Add try_rfold
This commit is contained in:
parent
c4b9a7f680
commit
55194edbf7
2 changed files with 99 additions and 0 deletions
|
@ -1,10 +1,12 @@
|
|||
mod nth_back;
|
||||
mod rfind;
|
||||
mod rfold;
|
||||
mod try_rfold;
|
||||
|
||||
use nth_back::NthBackFuture;
|
||||
use rfind::RFindFuture;
|
||||
use rfold::RFoldFuture;
|
||||
use try_rfold::TryRFoldFuture;
|
||||
|
||||
extension_trait! {
|
||||
use crate::stream::Stream;
|
||||
|
@ -121,5 +123,46 @@ extension_trait! {
|
|||
{
|
||||
RFoldFuture::new(self, accum, f)
|
||||
}
|
||||
|
||||
#[doc = r#"
|
||||
A combinator that applies a function as long as it returns successfully, producing a single, final value.
|
||||
Immediately returns the error when the function returns unsuccessfully.
|
||||
|
||||
# Examples
|
||||
|
||||
Basic usage:
|
||||
|
||||
```
|
||||
# fn main() { async_std::task::block_on(async {
|
||||
#
|
||||
use async_std::stream::Sample;
|
||||
use async_std::stream::double_ended::DoubleEndedStreamExt;
|
||||
|
||||
let s = Sample::from(vec![1, 2, 3, 4, 5]);
|
||||
let sum = s.try_rfold(0, |acc, v| {
|
||||
if (acc+v) % 2 == 1 {
|
||||
Ok(v+3)
|
||||
} else {
|
||||
Err("fail")
|
||||
}
|
||||
}).await;
|
||||
|
||||
assert_eq!(sum, Err("fail"));
|
||||
#
|
||||
# }) }
|
||||
```
|
||||
"#]
|
||||
fn try_rfold<B, F, E>(
|
||||
self,
|
||||
accum: B,
|
||||
f: F,
|
||||
) -> impl Future<Output = Option<B>> [TryRFoldFuture<Self, F, B>]
|
||||
where
|
||||
Self: Sized,
|
||||
F: FnMut(B, Self::Item) -> Result<B, E>,
|
||||
{
|
||||
TryRFoldFuture::new(self, accum, f)
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
56
src/stream/double_ended/try_rfold.rs
Normal file
56
src/stream/double_ended/try_rfold.rs
Normal file
|
@ -0,0 +1,56 @@
|
|||
use crate::future::Future;
|
||||
use std::pin::Pin;
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::stream::DoubleEndedStream;
|
||||
|
||||
pin_project! {
|
||||
#[doc(hidden)]
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct TryRFoldFuture<S, F, T> {
|
||||
#[pin]
|
||||
stream: S,
|
||||
f: F,
|
||||
acc: Option<T>,
|
||||
}
|
||||
}
|
||||
|
||||
impl<S, F, T> TryRFoldFuture<S, F, T> {
|
||||
pub(super) fn new(stream: S, init: T, f: F) -> Self {
|
||||
TryRFoldFuture {
|
||||
stream,
|
||||
f,
|
||||
acc: Some(init),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<S, F, T, E> Future for TryRFoldFuture<S, F, T>
|
||||
where
|
||||
S: DoubleEndedStream + Unpin,
|
||||
F: FnMut(T, S::Item) -> Result<T, E>,
|
||||
{
|
||||
type Output = Result<T, E>;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let mut this = self.project();
|
||||
loop {
|
||||
let next = futures_core::ready!(this.stream.as_mut().poll_next_back(cx));
|
||||
|
||||
match next {
|
||||
Some(v) => {
|
||||
let old = this.acc.take().unwrap();
|
||||
let new = (this.f)(old, v);
|
||||
|
||||
match new {
|
||||
Ok(o) => *this.acc = Some(o),
|
||||
Err(e) => return Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
None => return Poll::Ready(Ok(this.acc.take().unwrap())),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
Loading…
Reference in a new issue