1 use core::fmt;
2 use core::pin::Pin;
3 use futures_core::future::{FusedFuture, Future};
4 use futures_core::stream::Stream;
5 use futures_core::task::{Context, Poll};
6 use pin_project::{pin_project, project};
7 
8 /// Future for the [`fold`](super::StreamExt::fold) method.
9 #[pin_project]
10 #[must_use = "futures do nothing unless you `.await` or poll them"]
11 pub struct Fold<St, Fut, T, F> {
12     #[pin]
13     stream: St,
14     f: F,
15     accum: Option<T>,
16     #[pin]
17     future: Option<Fut>,
18 }
19 
20 impl<St, Fut, T, F> fmt::Debug for Fold<St, Fut, T, F>
21 where
22     St: fmt::Debug,
23     Fut: fmt::Debug,
24     T: fmt::Debug,
25 {
fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result26     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
27         f.debug_struct("Fold")
28             .field("stream", &self.stream)
29             .field("accum", &self.accum)
30             .field("future", &self.future)
31             .finish()
32     }
33 }
34 
35 impl<St, Fut, T, F> Fold<St, Fut, T, F>
36 where St: Stream,
37       F: FnMut(T, St::Item) -> Fut,
38       Fut: Future<Output = T>,
39 {
new(stream: St, f: F, t: T) -> Fold<St, Fut, T, F>40     pub(super) fn new(stream: St, f: F, t: T) -> Fold<St, Fut, T, F> {
41         Fold {
42             stream,
43             f,
44             accum: Some(t),
45             future: None,
46         }
47     }
48 }
49 
50 impl<St, Fut, T, F> FusedFuture for Fold<St, Fut, T, F>
51     where St: Stream,
52           F: FnMut(T, St::Item) -> Fut,
53           Fut: Future<Output = T>,
54 {
is_terminated(&self) -> bool55     fn is_terminated(&self) -> bool {
56         self.accum.is_none() && self.future.is_none()
57     }
58 }
59 
60 impl<St, Fut, T, F> Future for Fold<St, Fut, T, F>
61     where St: Stream,
62           F: FnMut(T, St::Item) -> Fut,
63           Fut: Future<Output = T>,
64 {
65     type Output = T;
66 
67     #[project]
poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T>68     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
69         #[project]
70         let Fold { mut stream, f, accum, mut future } = self.project();
71         Poll::Ready(loop {
72             if let Some(fut) = future.as_mut().as_pin_mut() {
73                 // we're currently processing a future to produce a new accum value
74                 *accum = Some(ready!(fut.poll(cx)));
75                 future.set(None);
76             } else if accum.is_some() {
77                 // we're waiting on a new item from the stream
78                 let res = ready!(stream.as_mut().poll_next(cx));
79                 let a = accum.take().unwrap();
80                 if let Some(item) = res {
81                     future.set(Some(f(a, item)));
82                 } else {
83                     break a;
84                 }
85             } else {
86                 panic!("Fold polled after completion")
87             }
88         })
89     }
90 }
91