Implement basic stream chaining

2 files changed, 25 insertions, 9 deletions

diff --git a/README.md b/README.md
index c039fc1..6d1d8d4 100644
--- a/README.md
+++ b/README.md
@@ -17,3 +17,10 @@ Meanwhile it seems some features aren't fully stable in Rust just yet. Some crat
 https://github.com/taiki-e/futures-async-stream
 
 https://github.com/tokio-rs/async-stream
+
+This page has a good review https://blog.yoshuawuyts.com/rust-streams/
+
+
+## Design
+
+Each block processes input and out asynchronously.
diff --git a/src/main.rs b/src/main.rs
index 52dc625..54e8529 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,5 +1,7 @@
 use async_stream::stream;
 
+use futures::stream::Stream;
+
 use futures_util::pin_mut;
 use futures_util::stream::StreamExt;
 
@@ -26,18 +28,25 @@ impl Future for Delay {
     }
 }
 
-#[tokio::main]
-async fn main() {
-    let s = stream! {
-        let mut when = Instant::now();
+fn number_source() -> impl Stream<Item = u8> {
+    stream! {
         for i in 0..10 {
-            let delay = Delay { when };
-            delay.await;
-            yield i;
-            when += Duration::from_millis(1000);
+            yield i
         }
-    };
+    }
+}
 
+fn double<S: Stream<Item = u8>>(input: S) -> impl Stream<Item = u8> {
+    stream! {
+        for await val in input {
+            yield val * 2
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    let s = double(number_source());
     pin_mut!(s); // needed for iteration
 
     while let Some(value) = s.next().await {