diff options
Diffstat (limited to 'src/main.rs')
| -rw-r--r-- | src/main.rs | 27 |
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 { |
