summaryrefslogtreecommitdiff
path: root/src/main.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/main.rs')
-rw-r--r--src/main.rs27
1 files changed, 18 insertions, 9 deletions
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 {