From b15f60d0d74227c01860193ca9a8518ab35475ce Mon Sep 17 00:00:00 2001
From: Yuval Adam <_@yuv.al>
Date: Tue, 5 Jul 2022 19:18:45 +0300
Subject: Implement basic stream chaining

---
 src/main.rs | 27 ++++++++++++++++++---------
 1 file changed, 18 insertions(+), 9 deletions(-)

(limited to 'src/main.rs')

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 {
-- 
cgit v1.3.1