tokio_stream/stream_ext/
skip.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`](super::StreamExt::skip) method.
11    #[must_use = "streams do nothing unless polled"]
12    pub struct Skip<St> {
13        #[pin]
14        stream: St,
15        remaining: usize,
16    }
17}
18
19impl<St> fmt::Debug for Skip<St>
20where
21    St: fmt::Debug,
22{
23    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
24        f.debug_struct("Skip")
25            .field("stream", &self.stream)
26            .finish()
27    }
28}
29
30impl<St> Skip<St> {
31    pub(super) fn new(stream: St, remaining: usize) -> Self {
32        Self { stream, remaining }
33    }
34}
35
36impl<St> Stream for Skip<St>
37where
38    St: Stream,
39{
40    type Item = St::Item;
41
42    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
43        loop {
44            match ready!(self.as_mut().project().stream.poll_next(cx)) {
45                Some(e) => {
46                    if self.remaining == 0 {
47                        return Poll::Ready(Some(e));
48                    }
49                    *self.as_mut().project().remaining -= 1;
50                }
51                None => return Poll::Ready(None),
52            }
53        }
54    }
55
56    fn size_hint(&self) -> (usize, Option<usize>) {
57        let (lower, upper) = self.stream.size_hint();
58
59        let lower = lower.saturating_sub(self.remaining);
60        let upper = upper.map(|x| x.saturating_sub(self.remaining));
61
62        (lower, upper)
63    }
64}
65
66impl<St> FusedStream for Skip<St>
67where
68    St: FusedStream,
69{
70    fn is_terminated(&self) -> bool {
71        self.stream.is_terminated()
72    }
73}
tokio_stream/stream_ext/skip.rs

tokio_stream/stream_ext/
skip.rs
