Skip to content

Commit a8e48b1

Browse files
bors[bot]montekki
andauthored
Merge #226
226: adds stream::step_by combinator r=stjepang a=montekki Ref: #129 Stdlib: https://doc.rust-lang.org/std/iter/trait.Iterator.html#method.step_by Co-authored-by: Fedor Sakharov <fedor.sakharov@gmail.com>
2 parents 6470130 + 376049b commit a8e48b1

File tree

2 files changed

+86
-0
lines changed

2 files changed

+86
-0
lines changed

src/stream/stream/mod.rs

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ mod nth;
3737
mod scan;
3838
mod skip;
3939
mod skip_while;
40+
mod step_by;
4041
mod take;
4142
mod zip;
4243

@@ -46,6 +47,7 @@ pub use inspect::Inspect;
4647
pub use scan::Scan;
4748
pub use skip::Skip;
4849
pub use skip_while::SkipWhile;
50+
pub use step_by::StepBy;
4951
pub use take::Take;
5052
pub use zip::Zip;
5153

@@ -232,6 +234,40 @@ pub trait Stream {
232234
}
233235
}
234236

237+
/// Creates a stream that yields each `step`th element.
238+
///
239+
/// # Panics
240+
///
241+
/// This method will panic if the given step is `0`.
242+
///
243+
/// # Examples
244+
///
245+
/// Basic usage:
246+
///
247+
/// ```
248+
/// # fn main() { async_std::task::block_on(async {
249+
/// #
250+
/// use async_std::prelude::*;
251+
/// use std::collections::VecDeque;
252+
///
253+
/// let s: VecDeque<_> = vec![0u8, 1, 2, 3, 4].into_iter().collect();
254+
/// let mut stepped = s.step_by(2);
255+
///
256+
/// assert_eq!(stepped.next().await, Some(0));
257+
/// assert_eq!(stepped.next().await, Some(2));
258+
/// assert_eq!(stepped.next().await, Some(4));
259+
/// assert_eq!(stepped.next().await, None);
260+
///
261+
/// #
262+
/// # }) }
263+
/// ```
264+
fn step_by(self, step: usize) -> StepBy<Self>
265+
where
266+
Self: Sized,
267+
{
268+
StepBy::new(self, step)
269+
}
270+
235271
/// Creates a stream that gives the current element's count as well as the next value.
236272
///
237273
/// # Overflow behaviour.

src/stream/stream/step_by.rs

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
use std::pin::Pin;
2+
3+
use crate::stream::Stream;
4+
use crate::task::{Context, Poll};
5+
6+
/// A stream that steps a given amount of elements of another stream.
7+
#[derive(Debug)]
8+
pub struct StepBy<S> {
9+
stream: S,
10+
step: usize,
11+
i: usize,
12+
}
13+
14+
impl<S> StepBy<S> {
15+
pin_utils::unsafe_pinned!(stream: S);
16+
pin_utils::unsafe_unpinned!(step: usize);
17+
pin_utils::unsafe_unpinned!(i: usize);
18+
19+
pub(crate) fn new(stream: S, step: usize) -> Self {
20+
StepBy {
21+
stream,
22+
step: step.checked_sub(1).unwrap(),
23+
i: 0,
24+
}
25+
}
26+
}
27+
28+
impl<S> Stream for StepBy<S>
29+
where
30+
S: Stream,
31+
{
32+
type Item = S::Item;
33+
34+
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
35+
loop {
36+
let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
37+
38+
match next {
39+
Some(v) => match self.i {
40+
0 => {
41+
*self.as_mut().i() = self.step;
42+
return Poll::Ready(Some(v));
43+
}
44+
_ => *self.as_mut().i() -= 1,
45+
},
46+
None => return Poll::Ready(None),
47+
}
48+
}
49+
}
50+
}

0 commit comments

Comments
 (0)