1use core::pin::Pin;
2
3use pin_project_lite::pin_project;
4
5use crate::stream::Stream;
6use crate::task::{Context, Poll};
7
8pin_project! {
9 /// A stream to filter elements of another stream with a predicate.
10 ///
11 /// This `struct` is created by the [`filter`] method on [`Stream`]. See its
12 /// documentation for more.
13 ///
14 /// [`filter`]: trait.Stream.html#method.filter
15 /// [`Stream`]: trait.Stream.html
16 #[derive(Debug)]
17 pub struct Filter<S, P> {
18 #[pin]
19 stream: S,
20 predicate: P,
21 }
22}
23
24impl<S, P> Filter<S, P> {
25 pub(super) fn new(stream: S, predicate: P) -> Self {
26 Self {
27 stream,
28 predicate,
29 }
30 }
31}
32
33impl<S, P> Stream for Filter<S, P>
34where
35 S: Stream,
36 P: FnMut(&S::Item) -> bool,
37{
38 type Item = S::Item;
39
40 fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
41 let this: Projection<'_, S, P> = self.project();
42 let next: Option<::Item> = futures_core::ready!(this.stream.poll_next(cx));
43
44 match next {
45 Some(v: ::Item) if (this.predicate)(&v) => Poll::Ready(Some(v)),
46 Some(_) => {
47 cx.waker().wake_by_ref();
48 Poll::Pending
49 }
50 None => Poll::Ready(None),
51 }
52 }
53}
54