From 2aa4a82499d4becd2284cdb482213d541b8804dd Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 28 Apr 2024 16:29:10 +0200 Subject: Adding upstream version 86.0.1. Signed-off-by: Daniel Baumann --- third_party/rust/tokio/src/stream/fold.rs | 51 +++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 third_party/rust/tokio/src/stream/fold.rs (limited to 'third_party/rust/tokio/src/stream/fold.rs') diff --git a/third_party/rust/tokio/src/stream/fold.rs b/third_party/rust/tokio/src/stream/fold.rs new file mode 100644 index 0000000000..7b9fead3db --- /dev/null +++ b/third_party/rust/tokio/src/stream/fold.rs @@ -0,0 +1,51 @@ +use crate::stream::Stream; + +use core::future::Future; +use core::pin::Pin; +use core::task::{Context, Poll}; +use pin_project_lite::pin_project; + +pin_project! { + /// Future returned by the [`fold`](super::StreamExt::fold) method. + #[derive(Debug)] + pub struct FoldFuture { + #[pin] + stream: St, + acc: Option, + f: F, + } +} + +impl FoldFuture { + pub(super) fn new(stream: St, init: B, f: F) -> Self { + Self { + stream, + acc: Some(init), + f, + } + } +} + +impl Future for FoldFuture +where + St: Stream, + F: FnMut(B, St::Item) -> B, +{ + type Output = B; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let mut me = self.project(); + loop { + let next = ready!(me.stream.as_mut().poll_next(cx)); + + match next { + Some(v) => { + let old = me.acc.take().unwrap(); + let new = (me.f)(old, v); + *me.acc = Some(new); + } + None => return Poll::Ready(me.acc.take().unwrap()), + } + } + } +} -- cgit v1.2.3