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}; 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 [`skip_while`](super::StreamExt::skip_while) method. 13 #[must_use = "streams do nothing unless polled"] 14 pub struct SkipWhile<St, Fut, F> where St: Stream { 15 #[pin] 16 stream: St, 17 f: F, 18 #[pin] 19 pending_fut: Option<Fut>, 20 pending_item: Option<St::Item>, 21 done_skipping: bool, 22 } 23 } 24 25 impl<St, Fut, F> fmt::Debug for SkipWhile<St, Fut, F> 26 where 27 St: Stream + fmt::Debug, 28 St::Item: fmt::Debug, 29 Fut: fmt::Debug, 30 { fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result31 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 32 f.debug_struct("SkipWhile") 33 .field("stream", &self.stream) 34 .field("pending_fut", &self.pending_fut) 35 .field("pending_item", &self.pending_item) 36 .field("done_skipping", &self.done_skipping) 37 .finish() 38 } 39 } 40 41 impl<St, Fut, F> SkipWhile<St, Fut, F> 42 where 43 St: Stream, 44 F: FnMut(&St::Item) -> Fut, 45 Fut: Future<Output = bool>, 46 { new(stream: St, f: F) -> Self47 pub(super) fn new(stream: St, f: F) -> Self { 48 Self { stream, f, pending_fut: None, pending_item: None, done_skipping: false } 49 } 50 51 delegate_access_inner!(stream, St, ()); 52 } 53 54 impl<St, Fut, F> FusedStream for SkipWhile<St, Fut, F> 55 where 56 St: FusedStream, 57 F: FnMut(&St::Item) -> Fut, 58 Fut: Future<Output = bool>, 59 { is_terminated(&self) -> bool60 fn is_terminated(&self) -> bool { 61 self.pending_item.is_none() && self.stream.is_terminated() 62 } 63 } 64 65 impl<St, Fut, F> Stream for SkipWhile<St, Fut, F> 66 where 67 St: Stream, 68 F: FnMut(&St::Item) -> Fut, 69 Fut: Future<Output = bool>, 70 { 71 type Item = St::Item; 72 poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<St::Item>>73 fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<St::Item>> { 74 let mut this = self.project(); 75 76 if *this.done_skipping { 77 return this.stream.poll_next(cx); 78 } 79 80 Poll::Ready(loop { 81 if let Some(fut) = this.pending_fut.as_mut().as_pin_mut() { 82 let skipped = ready!(fut.poll(cx)); 83 let item = this.pending_item.take(); 84 this.pending_fut.set(None); 85 if !skipped { 86 *this.done_skipping = true; 87 break item; 88 } 89 } else if let Some(item) = ready!(this.stream.as_mut().poll_next(cx)) { 90 this.pending_fut.set(Some((this.f)(&item))); 91 *this.pending_item = Some(item); 92 } else { 93 break None; 94 } 95 }) 96 } 97 size_hint(&self) -> (usize, Option<usize>)98 fn size_hint(&self) -> (usize, Option<usize>) { 99 if self.done_skipping { 100 self.stream.size_hint() 101 } else { 102 let pending_len = usize::from(self.pending_item.is_some()); 103 let (_, upper) = self.stream.size_hint(); 104 let upper = match upper { 105 Some(x) => x.checked_add(pending_len), 106 None => None, 107 }; 108 (0, upper) // can't know a lower bound, due to the predicate 109 } 110 } 111 } 112 113 // Forwarding impl of Sink from the underlying stream 114 #[cfg(feature = "sink")] 115 impl<S, Fut, F, Item> Sink<Item> for SkipWhile<S, Fut, F> 116 where 117 S: Stream + Sink<Item>, 118 F: FnMut(&S::Item) -> Fut, 119 Fut: Future<Output = bool>, 120 { 121 type Error = S::Error; 122 123 delegate_sink!(stream, Item); 124 } 125