From ac5e6af1f3c271ce3ae798ccd483c3aa82aa2c58 Mon Sep 17 00:00:00 2001
From: Yuval Adam <_@yuv.al>
Date: Tue, 5 Jul 2022 13:37:52 +0300
Subject: Tokio async stream hello world

---
 Cargo.lock  | 84 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 Cargo.toml  |  4 +++
 README.md   |  5 ++++
 src/main.rs | 45 ++++++++++-----------------------
 4 files changed, 106 insertions(+), 32 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 12e2899..65f4be3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2,6 +2,27 @@
 # It is not intended for manual editing.
 version = 3
 
+[[package]]
+name = "async-stream"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dad5c83079eae9969be7fadefe640a1c566901f05ff91ab221de4b6f68d9507e"
+dependencies = [
+ "async-stream-impl",
+ "futures-core",
+]
+
+[[package]]
+name = "async-stream-impl"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "10f203db73a71dfa2fb6dd22763990fa26f3d2625a6da2da900d23b87d26be27"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "cfg-if"
 version = "1.0.0"
@@ -108,6 +129,15 @@ dependencies = [
  "wasi",
 ]
 
+[[package]]
+name = "hermit-abi"
+version = "0.1.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "62b467343b94ba476dcb2500d242dadbb39557df889310ac77c5d99100aaac33"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "libc"
 version = "0.2.126"
@@ -120,6 +150,22 @@ version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d"
 
+[[package]]
+name = "num_cpus"
+version = "1.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "19e64526ebdee182341572e50e9ad03965aa510cd94427a4549448f285e957a1"
+dependencies = [
+ "hermit-abi",
+ "libc",
+]
+
+[[package]]
+name = "once_cell"
+version = "1.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "18a6dbe30758c9f83eb00cbea4ac95966305f5a7772f3f42ebfc7fc7eddbd8e1"
+
 [[package]]
 name = "pin-project-lite"
 version = "0.2.9"
@@ -190,8 +236,12 @@ dependencies = [
 name = "rust-async-playground"
 version = "0.1.0"
 dependencies = [
+ "async-stream",
  "futures",
+ "futures-util",
  "rand",
+ "tokio",
+ "tokio-stream",
 ]
 
 [[package]]
@@ -211,6 +261,40 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "tokio"
+version = "1.19.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c51a52ed6686dd62c320f9b89299e9dfb46f730c7a48e635c19f21d116cb1439"
+dependencies = [
+ "num_cpus",
+ "once_cell",
+ "pin-project-lite",
+ "tokio-macros",
+]
+
+[[package]]
+name = "tokio-macros"
+version = "1.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9724f9a975fb987ef7a3cd9be0350edcbe130698af5b8f7a631e23d42d052484"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "tokio-stream"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df54d54117d6fdc4e4fea40fe1e4e566b3505700e148a6827e59b34b0d2600d9"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.1"
diff --git a/Cargo.toml b/Cargo.toml
index 4ab1e0e..fa1525e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -4,5 +4,9 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+async-stream = "0.3.3"
 futures = "0.3"
+futures-util = "0.3.21"
 rand = "0.8"
+tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros"] }
+tokio-stream = "0.1.9"
diff --git a/README.md b/README.md
index e863d02..c039fc1 100644
--- a/README.md
+++ b/README.md
@@ -12,3 +12,8 @@ https://docs.rs/futures/0.3.6/futures/stream/trait.Stream.html
 
 This is the relevant chapter in the Rust Async Book - https://rust-lang.github.io/async-book/05_streams/02_iteration_and_concurrency.html
 
+Meanwhile it seems some features aren't fully stable in Rust just yet. Some crates provide relevant functionality
+
+https://github.com/taiki-e/futures-async-stream
+
+https://github.com/tokio-rs/async-stream
diff --git a/src/main.rs b/src/main.rs
index 642901b..55c435d 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,38 +1,19 @@
-use futures::executor::block_on;
-use rand::prelude::*;
+use async_stream::stream;
 
-trait SimpleFuture {
-    type Output;
-    fn poll(&mut self, wake: fn()) -> Poll<Self::Output>;
-}
-
-enum Poll<T> {
-    Ready(T),
-    Pending,
-}
-
-pub struct RandomFuture {}
+use futures_util::pin_mut;
+use futures_util::stream::StreamExt;
 
-impl SimpleFuture for RandomFuture {
-    type Output = f64;
-    fn poll(&mut self, _wake: fn()) -> Poll<Self::Output> {
-        let mut rng = rand::thread_rng();
-        let y: f64 = rng.gen();
-        println!("y value {}", y);
-        if y > 0.9 {
-            Poll::Ready(y)
-        } else {
-            // probably need to utilize wake() here
-            Poll::Pending
+#[tokio::main]
+async fn main() {
+    let s = stream! {
+        for i in 0..3 {
+            yield i;
         }
-    }
-}
+    };
 
-async fn hello_world() {
-    println!("hello, world!");
-}
+    pin_mut!(s); // needed for iteration
 
-fn main() {
-    let future = hello_world();
-    block_on(future);
+    while let Some(value) = s.next().await {
+        println!("got {}", value);
+    }
 }
-- 
cgit v1.3.1