diff options
| -rw-r--r-- | README.md | 7 | ||||
| -rw-r--r-- | src/main.rs | 27 |
2 files changed, 25 insertions, 9 deletions
@@ -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 { |
