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