From e7ae10ebee69258c144d10793e829f786f2b1f2f Mon Sep 17 00:00:00 2001 From: Fedor Sakharov Date: Wed, 11 Sep 2019 14:44:42 +0300 Subject: [PATCH 1/6] adds stream::filter combinator --- src/stream/stream/filter.rs | 49 +++++++++++++++++++++++++++++++++++++ src/stream/stream/mod.rs | 30 +++++++++++++++++++++++ 2 files changed, 79 insertions(+) create mode 100644 src/stream/stream/filter.rs diff --git a/src/stream/stream/filter.rs b/src/stream/stream/filter.rs new file mode 100644 index 0000000..135d75d --- /dev/null +++ b/src/stream/stream/filter.rs @@ -0,0 +1,49 @@ +use std::marker::PhantomData; +use std::pin::Pin; + +use crate::stream::Stream; +use crate::task::{Context, Poll}; + +#[doc(hidden)] +#[allow(missing_debug_implementations)] +pub struct Filter { + stream: S, + predicate: P, + __t: PhantomData, +} + +impl Filter { + pin_utils::unsafe_pinned!(stream: S); + pin_utils::unsafe_unpinned!(predicate: P); + + pub(super) fn new(stream: S, predicate: P) -> Self { + Filter { + stream, + predicate, + __t: PhantomData, + } + } +} + +impl futures_core::stream::Stream for Filter +where + S: Stream, + P: FnMut(&S::Item) -> bool, +{ + type Item = S::Item; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); + + match next { + Some(v) => match (self.as_mut().predicate())(&v) { + true => Poll::Ready(Some(v)), + false => { + cx.waker().wake_by_ref(); + Poll::Pending + } + }, + None => Poll::Ready(None), + } + } +} diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index 8ddab21..195f2eb 100644 --- a/src/stream/stream/mod.rs +++ b/src/stream/stream/mod.rs @@ -24,6 +24,7 @@ mod all; mod any; mod enumerate; +mod filter; mod filter_map; mod find; mod find_map; @@ -44,6 +45,7 @@ pub use zip::Zip; use all::AllFuture; use any::AnyFuture; use enumerate::Enumerate; +use filter::Filter; use filter_map::FilterMap; use find::FindFuture; use find_map::FindMapFuture; @@ -282,6 +284,34 @@ pub trait Stream { done: false, } } + /// Creates a stream that uses a predicate to determine if an element + /// should be yeilded. + /// + /// # Examples + /// + /// Basic usage: + /// + ///``` + /// # fn main() { async_std::task::block_on(async { + /// # + /// use std::collections::VecDeque; + /// use async_std::stream::Stream; + /// + /// let s: VecDeque = vec![1, 2, 3, 4].into_iter().collect(); + /// let mut s = s.filter(|i| i % 2 == 0); + /// + /// assert_eq!(s.next().await, Some(2)); + /// assert_eq!(s.next().await, Some(4)); + /// assert_eq!(s.next().await, None); + /// # + /// # }) } + fn filter

(self, predicate: P) -> Filter + where + Self: Sized, + P: FnMut(&Self::Item) -> bool, + { + Filter::new(self, predicate) + } /// Both filters and maps a stream. /// From 570329b17648727a2800b3ecb1bf21dfd13bb320 Mon Sep 17 00:00:00 2001 From: Fedor Sakharov Date: Sat, 21 Sep 2019 14:40:25 +0300 Subject: [PATCH 2/6] adds stream::skip combinator --- src/stream/stream/mod.rs | 27 ++++++++++++++++++++++++++ src/stream/stream/skip.rs | 41 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+) create mode 100644 src/stream/stream/skip.rs diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index 8ddab21..5f4010c 100644 --- a/src/stream/stream/mod.rs +++ b/src/stream/stream/mod.rs @@ -33,6 +33,7 @@ mod min_by; mod next; mod nth; mod scan; +mod skip; mod take; mod zip; @@ -51,6 +52,7 @@ use fold::FoldFuture; use min_by::MinByFuture; use next::NextFuture; use nth::NthFuture; +use skip::Skip; use std::cmp::Ordering; use std::marker::PhantomData; @@ -661,6 +663,31 @@ pub trait Stream { Scan::new(self, initial_state, f) } + /// Creates a combinator that skips the first `n` elements. + /// + /// ## Examples + /// + /// ``` + /// # fn main() { async_std::task::block_on(async { + /// # + /// use std::collections::VecDeque; + /// use async_std::stream::Stream; + /// + /// let s: VecDeque = vec![1, 2, 3].into_iter().collect(); + /// let mut skipped = s.skip(2); + /// + /// assert_eq!(skipped.next().await, Some(3)); + /// assert_eq!(skipped.next().await, None); + /// # + /// # }) } + /// ``` + fn skip(self, n: usize) -> Skip + where + Self: Sized, + { + Skip::new(self, n) + } + /// 'Zips up' two streams into a single stream of pairs. /// /// `zip()` returns a new stream that will iterate over two other streams, returning a tuple diff --git a/src/stream/stream/skip.rs b/src/stream/stream/skip.rs new file mode 100644 index 0000000..09f5cab --- /dev/null +++ b/src/stream/stream/skip.rs @@ -0,0 +1,41 @@ +use std::pin::Pin; +use std::task::{Context, Poll}; + +use crate::stream::Stream; + +#[doc(hidden)] +#[allow(missing_debug_implementations)] +pub struct Skip { + stream: S, + n: usize, +} + +impl Skip { + pin_utils::unsafe_pinned!(stream: S); + pin_utils::unsafe_unpinned!(n: usize); + + pub(crate) fn new(stream: S, n: usize) -> Self { + Skip { stream, n } + } +} + +impl Stream for Skip +where + S: Stream, +{ + type Item = S::Item; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + loop { + let next = futures_core::ready!(self.as_mut().stream().poll_next(cx)); + + match next { + Some(v) => match self.n { + 0 => return Poll::Ready(Some(v)), + _ => *self.as_mut().n() -= 1, + }, + None => return Poll::Ready(None), + } + } + } +} From 53ce30ae6655c1dc107a171a4364cc4055493f26 Mon Sep 17 00:00:00 2001 From: Stjepan Glavina Date: Sat, 21 Sep 2019 15:17:49 +0200 Subject: [PATCH 3/6] Fix async_std imports in metadata.rs --- src/fs/metadata.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/fs/metadata.rs b/src/fs/metadata.rs index 6bb9993..2c9e41e 100644 --- a/src/fs/metadata.rs +++ b/src/fs/metadata.rs @@ -70,7 +70,7 @@ cfg_if! { /// ```no_run /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// # - /// use std::fs; + /// use async_std::fs; /// /// let metadata = fs::metadata("a.txt").await?; /// println!("{:?}", metadata.file_type()); @@ -90,7 +90,7 @@ cfg_if! { /// ```no_run /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// # - /// use std::fs; + /// use async_std::fs; /// /// let metadata = fs::metadata(".").await?; /// println!("{:?}", metadata.is_dir()); @@ -110,7 +110,7 @@ cfg_if! { /// ```no_run /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// # - /// use std::fs; + /// use async_std::fs; /// /// let metadata = fs::metadata("a.txt").await?; /// println!("{:?}", metadata.is_file()); @@ -128,7 +128,7 @@ cfg_if! { /// ```no_run /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// # - /// use std::fs; + /// use async_std::fs; /// /// let metadata = fs::metadata("a.txt").await?; /// println!("{}", metadata.len()); @@ -146,7 +146,7 @@ cfg_if! { /// ```no_run /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// # - /// use std::fs; + /// use async_std::fs; /// /// let metadata = fs::metadata("a.txt").await?; /// println!("{:?}", metadata.permissions()); @@ -169,7 +169,7 @@ cfg_if! { /// ```no_run /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// # - /// use std::fs; + /// use async_std::fs; /// /// let metadata = fs::metadata("a.txt").await?; /// println!("{:?}", metadata.modified()); @@ -192,7 +192,7 @@ cfg_if! { /// ```no_run /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// # - /// use std::fs; + /// use async_std::fs; /// /// let metadata = fs::metadata("a.txt").await?; /// println!("{:?}", metadata.accessed()); @@ -215,7 +215,7 @@ cfg_if! { /// ```no_run /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async { /// # - /// use std::fs; + /// use async_std::fs; /// /// let metadata = fs::metadata("a.txt").await?; /// println!("{:?}", metadata.created()); From 75da138696b495b65775aab356ad8146f1c2b8e8 Mon Sep 17 00:00:00 2001 From: Fedor Sakharov Date: Sat, 21 Sep 2019 16:37:30 +0300 Subject: [PATCH 4/6] export Skip type --- src/stream/stream/mod.rs | 2 +- src/stream/stream/skip.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index 5f4010c..6444651 100644 --- a/src/stream/stream/mod.rs +++ b/src/stream/stream/mod.rs @@ -39,6 +39,7 @@ mod zip; pub use fuse::Fuse; pub use scan::Scan; +use skip::Skip; pub use take::Take; pub use zip::Zip; @@ -52,7 +53,6 @@ use fold::FoldFuture; use min_by::MinByFuture; use next::NextFuture; use nth::NthFuture; -use skip::Skip; use std::cmp::Ordering; use std::marker::PhantomData; diff --git a/src/stream/stream/skip.rs b/src/stream/stream/skip.rs index 09f5cab..8a2d966 100644 --- a/src/stream/stream/skip.rs +++ b/src/stream/stream/skip.rs @@ -3,8 +3,8 @@ use std::task::{Context, Poll}; use crate::stream::Stream; -#[doc(hidden)] -#[allow(missing_debug_implementations)] +/// A stream to skip first n elements of another stream. +#[derive(Debug)] pub struct Skip { stream: S, n: usize, From e430851bc400f6bf3c5dc4562c44ae0a012e23f2 Mon Sep 17 00:00:00 2001 From: Fedor Sakharov Date: Sat, 21 Sep 2019 16:40:01 +0300 Subject: [PATCH 5/6] export Filter type --- src/stream/stream/filter.rs | 4 ++-- src/stream/stream/mod.rs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/stream/stream/filter.rs b/src/stream/stream/filter.rs index 135d75d..68211f7 100644 --- a/src/stream/stream/filter.rs +++ b/src/stream/stream/filter.rs @@ -4,8 +4,8 @@ use std::pin::Pin; use crate::stream::Stream; use crate::task::{Context, Poll}; -#[doc(hidden)] -#[allow(missing_debug_implementations)] +/// A stream to filter elements of another stream with a predicate. +#[derive(Debug)] pub struct Filter { stream: S, predicate: P, diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index 195f2eb..66645ce 100644 --- a/src/stream/stream/mod.rs +++ b/src/stream/stream/mod.rs @@ -37,6 +37,7 @@ mod scan; mod take; mod zip; +pub use filter::Filter; pub use fuse::Fuse; pub use scan::Scan; pub use take::Take; @@ -45,7 +46,6 @@ pub use zip::Zip; use all::AllFuture; use any::AnyFuture; use enumerate::Enumerate; -use filter::Filter; use filter_map::FilterMap; use find::FindFuture; use find_map::FindMapFuture; From fdd81e1b2a04d174fb8f4a0f1f7e170dcffa501d Mon Sep 17 00:00:00 2001 From: Fedor Sakharov Date: Sat, 21 Sep 2019 16:40:58 +0300 Subject: [PATCH 6/6] Actually export Skip --- src/stream/stream/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index 6444651..6284062 100644 --- a/src/stream/stream/mod.rs +++ b/src/stream/stream/mod.rs @@ -39,7 +39,7 @@ mod zip; pub use fuse::Fuse; pub use scan::Scan; -use skip::Skip; +pub use skip::Skip; pub use take::Take; pub use zip::Zip;