diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index e68e6ac..f7a1ba2 100644 --- a/src/stream/stream/mod.rs +++ b/src/stream/stream/mod.rs @@ -336,28 +336,19 @@ extension_trait! { let start = Instant::now(); // emit value every 5 milliseconds - let s = stream::interval(Duration::from_millis(5)) - .enumerate() - .take(3); + let s = stream::interval(Duration::from_millis(5)).take(2); // throttle for 10 milliseconds let mut s = s.throttle(Duration::from_millis(10)); - assert_eq!(s.next().await, Some((0, ()))); - let duration_ms = start.elapsed().as_millis(); - assert!(duration_ms >= 5); + s.next().await; + assert!(start.elapsed().as_millis() >= 5); - assert_eq!(s.next().await, Some((1, ()))); - let duration_ms = start.elapsed().as_millis(); - assert!(duration_ms >= 15); + s.next().await; + assert!(start.elapsed().as_millis() >= 15); - assert_eq!(s.next().await, Some((2, ()))); - let duration_ms = start.elapsed().as_millis(); - assert!(duration_ms >= 25); - - assert_eq!(s.next().await, None); - let duration_ms = start.elapsed().as_millis(); - assert!(duration_ms >= 35); + s.next().await; + assert!(start.elapsed().as_millis() >= 35); # # }) } ``` diff --git a/src/stream/stream/throttle.rs b/src/stream/stream/throttle.rs index b2480bb..ce8c13b 100644 --- a/src/stream/stream/throttle.rs +++ b/src/stream/stream/throttle.rs @@ -55,10 +55,7 @@ impl Stream for Throttle { } match this.stream.poll_next(cx) { - Poll::Pending => { - cx.waker().wake_by_ref(); // Continue driving even though emitting Pending - Poll::Pending - } + Poll::Pending => Poll::Pending, Poll::Ready(None) => Poll::Ready(None), Poll::Ready(Some(v)) => { *this.blocked = true;