diff options
author | Artem Vorotnikov <artem@vorotnikov.me> | 2019-12-21 07:17:05 +0300 |
---|---|---|
committer | Carl Lerche <me@carllerche.com> | 2019-12-20 20:17:05 -0800 |
commit | 3b9c7b1715777b6db69d420c1530aa845e6306c3 (patch) | |
tree | ef8f7016c075d49e5d9b894bb400f30ad45288e7 /tokio/src/stream/filter_map.rs | |
parent | 3bff5a3ffe68c7bbf94fc91a58633f5a91f4266c (diff) |
stream: filtering utilities (#2001)
Adds `StreamExt::filter` and `StreamExt::filter_map`.
Diffstat (limited to 'tokio/src/stream/filter_map.rs')
-rw-r--r-- | tokio/src/stream/filter_map.rs | 62 |
1 files changed, 62 insertions, 0 deletions
diff --git a/tokio/src/stream/filter_map.rs b/tokio/src/stream/filter_map.rs new file mode 100644 index 00000000..3aeb036f --- /dev/null +++ b/tokio/src/stream/filter_map.rs @@ -0,0 +1,62 @@ +use crate::stream::Stream; + +use core::fmt; +use core::pin::Pin; +use core::task::{Context, Poll}; +use pin_project_lite::pin_project; + +pin_project! { + /// Stream returned by the [`filter_map`](super::StreamExt::filter_map) method. + #[must_use = "streams do nothing unless polled"] + pub struct FilterMap<St, F> { + #[pin] + stream: St, + f: F, + } +} + +impl<St, F> fmt::Debug for FilterMap<St, F> +where + St: fmt::Debug, +{ + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("FilterMap") + .field("stream", &self.stream) + .finish() + } +} + +impl<St, F, T> FilterMap<St, F> +where + St: Stream, + F: FnMut(St::Item) -> Option<T>, +{ + pub(super) fn new(stream: St, f: F) -> Self { + Self { stream, f } + } +} + +impl<St, F, T> Stream for FilterMap<St, F> +where + St: Stream, + F: FnMut(St::Item) -> Option<T>, +{ + type Item = T; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> { + loop { + match ready!(self.as_mut().project().stream.poll_next(cx)) { + Some(e) => { + if let Some(e) = (self.as_mut().project().f)(e) { + return Poll::Ready(Some(e)); + } + } + None => return Poll::Ready(None), + } + } + } + + fn size_hint(&self) -> (usize, Option<usize>) { + (0, self.stream.size_hint().1) // can't know a lower bound, due to the predicate + } +} |