]>
Commit | Line | Data |
---|---|---|
5099ac24 FG |
1 | use core::fmt; |
2 | use core::pin::Pin; | |
3 | use futures_core::future::Future; | |
4 | use futures_core::ready; | |
5 | use futures_core::stream::{FusedStream, Stream, TryStream}; | |
6 | use futures_core::task::{Context, Poll}; | |
7 | #[cfg(feature = "sink")] | |
8 | use futures_sink::Sink; | |
9 | use pin_project_lite::pin_project; | |
10 | ||
11 | pin_project! { | |
12 | /// Stream for the [`try_filter`](super::TryStreamExt::try_filter) | |
13 | /// method. | |
14 | #[must_use = "streams do nothing unless polled"] | |
15 | pub struct TryFilter<St, Fut, F> | |
16 | where St: TryStream | |
17 | { | |
18 | #[pin] | |
19 | stream: St, | |
20 | f: F, | |
21 | #[pin] | |
22 | pending_fut: Option<Fut>, | |
23 | pending_item: Option<St::Ok>, | |
24 | } | |
25 | } | |
26 | ||
27 | impl<St, Fut, F> fmt::Debug for TryFilter<St, Fut, F> | |
28 | where | |
29 | St: TryStream + fmt::Debug, | |
30 | St::Ok: fmt::Debug, | |
31 | Fut: fmt::Debug, | |
32 | { | |
33 | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { | |
34 | f.debug_struct("TryFilter") | |
35 | .field("stream", &self.stream) | |
36 | .field("pending_fut", &self.pending_fut) | |
37 | .field("pending_item", &self.pending_item) | |
38 | .finish() | |
39 | } | |
40 | } | |
41 | ||
42 | impl<St, Fut, F> TryFilter<St, Fut, F> | |
43 | where | |
44 | St: TryStream, | |
45 | { | |
46 | pub(super) fn new(stream: St, f: F) -> Self { | |
47 | Self { stream, f, pending_fut: None, pending_item: None } | |
48 | } | |
49 | ||
50 | delegate_access_inner!(stream, St, ()); | |
51 | } | |
52 | ||
53 | impl<St, Fut, F> FusedStream for TryFilter<St, Fut, F> | |
54 | where | |
55 | St: TryStream + FusedStream, | |
56 | F: FnMut(&St::Ok) -> Fut, | |
57 | Fut: Future<Output = bool>, | |
58 | { | |
59 | fn is_terminated(&self) -> bool { | |
60 | self.pending_fut.is_none() && self.stream.is_terminated() | |
61 | } | |
62 | } | |
63 | ||
64 | impl<St, Fut, F> Stream for TryFilter<St, Fut, F> | |
65 | where | |
66 | St: TryStream, | |
67 | Fut: Future<Output = bool>, | |
68 | F: FnMut(&St::Ok) -> Fut, | |
69 | { | |
70 | type Item = Result<St::Ok, St::Error>; | |
71 | ||
72 | fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { | |
73 | let mut this = self.project(); | |
74 | ||
75 | Poll::Ready(loop { | |
76 | if let Some(fut) = this.pending_fut.as_mut().as_pin_mut() { | |
77 | let res = ready!(fut.poll(cx)); | |
78 | this.pending_fut.set(None); | |
79 | if res { | |
80 | break this.pending_item.take().map(Ok); | |
81 | } | |
82 | *this.pending_item = None; | |
83 | } else if let Some(item) = ready!(this.stream.as_mut().try_poll_next(cx)?) { | |
84 | this.pending_fut.set(Some((this.f)(&item))); | |
85 | *this.pending_item = Some(item); | |
86 | } else { | |
87 | break None; | |
88 | } | |
89 | }) | |
90 | } | |
91 | ||
92 | fn size_hint(&self) -> (usize, Option<usize>) { | |
353b0b11 | 93 | let pending_len = usize::from(self.pending_fut.is_some()); |
5099ac24 FG |
94 | let (_, upper) = self.stream.size_hint(); |
95 | let upper = match upper { | |
96 | Some(x) => x.checked_add(pending_len), | |
97 | None => None, | |
98 | }; | |
99 | (0, upper) // can't know a lower bound, due to the predicate | |
100 | } | |
101 | } | |
102 | ||
103 | // Forwarding impl of Sink from the underlying stream | |
104 | #[cfg(feature = "sink")] | |
105 | impl<S, Fut, F, Item, E> Sink<Item> for TryFilter<S, Fut, F> | |
106 | where | |
107 | S: TryStream + Sink<Item, Error = E>, | |
108 | { | |
109 | type Error = E; | |
110 | ||
111 | delegate_sink!(stream, Item); | |
112 | } |