tokio_stream/stream_ext/
skip_while.rs

1use crate::Stream;
2
3use core::fmt;
4use core::pin::Pin;
5use core::task::{ready, Context, Poll};
6use futures_core::FusedStream;
7use pin_project_lite::pin_project;
8
9pin_project! {
10    /// Stream for the [`skip_while`](super::StreamExt::skip_while) method.
11    #[must_use = "streams do nothing unless polled"]
12    pub struct SkipWhile<St, F> {
13        #[pin]
14        stream: St,
15        predicate: Option<F>,
16    }
17}
18
19impl<St, F> fmt::Debug for SkipWhile<St, F>
20where
21    St: fmt::Debug,
22{
23    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
24        f.debug_struct("SkipWhile")
25            .field("stream", &self.stream)
26            .finish()
27    }
28}
29
30impl<St, F> SkipWhile<St, F> {
31    pub(super) fn new(stream: St, predicate: F) -> Self {
32        Self {
33            stream,
34            predicate: Some(predicate),
35        }
36    }
37}
38
39impl<St, F> Stream for SkipWhile<St, F>
40where
41    St: Stream,
42    F: FnMut(&St::Item) -> bool,
43{
44    type Item = St::Item;
45
46    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
47        let mut this = self.project();
48        if let Some(predicate) = this.predicate {
49            loop {
50                match ready!(this.stream.as_mut().poll_next(cx)) {
51                    Some(item) => {
52                        if !(predicate)(&item) {
53                            *this.predicate = None;
54                            return Poll::Ready(Some(item));
55                        }
56                    }
57                    None => return Poll::Ready(None),
58                }
59            }
60        } else {
61            this.stream.poll_next(cx)
62        }
63    }
64
65    fn size_hint(&self) -> (usize, Option<usize>) {
66        let (lower, upper) = self.stream.size_hint();
67
68        if self.predicate.is_some() {
69            return (0, upper);
70        }
71
72        (lower, upper)
73    }
74}
75
76impl<St, F> FusedStream for SkipWhile<St, F>
77where
78    St: FusedStream,
79    F: FnMut(&St::Item) -> bool,
80{
81    fn is_terminated(&self) -> bool {
82        self.stream.is_terminated()
83    }
84}
tokio_stream/stream_ext/skip_while.rs

tokio_stream/stream_ext/
skip_while.rs
