fix Stream::throttle hot loop (#584)

Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com>

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))
+            let s = stream::interval(Duration::from_millis(5)).take(2);
-                .enumerate()
-                .take(3);
 
             // throttle for 10 milliseconds
             let mut s = s.throttle(Duration::from_millis(10));
 
-            assert_eq!(s.next().await, Some((0, ())));
+            s.next().await;
-            let duration_ms = start.elapsed().as_millis();
+            assert!(start.elapsed().as_millis() >= 5);
-            assert!(duration_ms >= 5);
 
-            assert_eq!(s.next().await, Some((1, ())));
+            s.next().await;
-            let duration_ms = start.elapsed().as_millis();
+            assert!(start.elapsed().as_millis() >= 15);
-            assert!(duration_ms >= 15);
 
-            assert_eq!(s.next().await, Some((2, ())));
+            s.next().await;
-            let duration_ms = start.elapsed().as_millis();
+            assert!(start.elapsed().as_millis() >= 35);
-            assert!(duration_ms >= 25);
-
-            assert_eq!(s.next().await, None);
-            let duration_ms = start.elapsed().as_millis();
-            assert!(duration_ms >= 35);
             #
             # }) }
             ```

src/stream/stream/throttle.rs

@@ -55,10 +55,7 @@ impl<S: Stream> Stream for Throttle<S> {
         }
 
         match this.stream.poll_next(cx) {
-            Poll::Pending => {
+            Poll::Pending => Poll::Pending,
-                cx.waker().wake_by_ref(); // Continue driving even though emitting Pending
-                Poll::Pending
-            }
             Poll::Ready(None) => Poll::Ready(None),
             Poll::Ready(Some(v)) => {
                 *this.blocked = true;