stream: iter() should yield every so often. (#2343)

tokio/src/stream/iter.rs

@@ -44,7 +44,8 @@ where
 {
     type Item = I::Item;
 
-    fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<I::Item>> {
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<I::Item>> {
+        ready!(crate::coop::poll_proceed(cx));
         Poll::Ready(self.iter.next())
     }
 

tokio/tests/stream_iter.rs

@@ -0,0 +1,18 @@
+use tokio::stream;
+use tokio_test::task;
+
+use std::iter;
+
+#[tokio::test]
+async fn coop() {
+    let mut stream = task::spawn(stream::iter(iter::repeat(1)));
+
+    for _ in 0..10_000 {
+        if stream.poll_next().is_pending() {
+            assert!(stream.is_woken());
+            return;
+        }
+    }
+
+    panic!("did not yield");
+}