async_std/stream/double_ended_stream/
try_rfold.rs

1use crate::future::Future;
2use core::pin::Pin;
3use crate::task::{Context, Poll};
4
5use pin_project_lite::pin_project;
6
7use crate::stream::DoubleEndedStream;
8
9pin_project! {
10    #[doc(hidden)]
11    #[allow(missing_debug_implementations)]
12    pub struct TryRFoldFuture<S, F, T> {
13        #[pin]
14        stream: S,
15        f: F,
16        acc: Option<T>,
17    }
18}
19
20impl<S, F, T> TryRFoldFuture<S, F, T> {
21    pub(super) fn new(stream: S, init: T, f: F) -> Self {
22        TryRFoldFuture {
23            stream,
24            f,
25            acc: Some(init),
26        }
27    }
28}
29
30impl<S, F, T, E> Future for TryRFoldFuture<S, F, T>
31where
32    S: DoubleEndedStream + Unpin,
33    F: FnMut(T, S::Item) -> Result<T, E>,
34{
35    type Output = Result<T, E>;
36
37    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
38        let mut this = self.project();
39        loop {
40            let next = futures_core::ready!(this.stream.as_mut().poll_next_back(cx));
41
42            match next {
43                Some(v) => {
44                    let old = this.acc.take().unwrap();
45                    let new = (this.f)(old, v);
46
47                    match new {
48                        Ok(o) => *this.acc = Some(o),
49                        Err(e) => return Poll::Ready(Err(e)),
50                    }
51                }
52                None => return Poll::Ready(Ok(this.acc.take().unwrap())),
53            }
54        }
55    }
56}
async_std/stream/double_ended_stream/try_rfold.rs

async_std/stream/double_ended_stream/
try_rfold.rs
