async_std/stream/stream/
position.rs

1use core::future::Future;
2use core::pin::Pin;
3
4use crate::stream::Stream;
5use crate::task::{Context, Poll};
6
7#[doc(hidden)]
8#[allow(missing_debug_implementations)]
9pub struct PositionFuture<'a, S, P> {
10    stream: &'a mut S,
11    predicate: P,
12    index: usize,
13}
14
15impl<'a, S, P> Unpin for PositionFuture<'a, S, P> {}
16
17impl<'a, S, P> PositionFuture<'a, S, P> {
18    pub(super) fn new(stream: &'a mut S, predicate: P) -> Self {
19        Self {
20            stream,
21            predicate,
22            index: 0,
23        }
24    }
25}
26
27impl<'a, S, P> Future for PositionFuture<'a, S, P>
28where
29    S: Stream + Unpin,
30    P: FnMut(S::Item) -> bool,
31{
32    type Output = Option<usize>;
33
34    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
35        let next = futures_core::ready!(Pin::new(&mut self.stream).poll_next(cx));
36
37        match next {
38            Some(v) => {
39                if (&mut self.predicate)(v) {
40                    Poll::Ready(Some(self.index))
41                } else {
42                    cx.waker().wake_by_ref();
43                    self.index += 1;
44                    Poll::Pending
45                }
46            }
47            None => Poll::Ready(None),
48        }
49    }
50}
async_std/stream/stream/position.rs

async_std/stream/stream/
position.rs
