blob: c82d35d9d9195f6aee3b6a8a32e57fb1e57ddcee (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
use async_stream::stream;
use futures::stream::Stream;
use futures_util::pin_mut;
use futures_util::stream::StreamExt;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
struct Delay {
when: Instant,
}
impl Future for Delay {
type Output = &'static str;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
if Instant::now() >= self.when {
Poll::Ready("done")
} else {
// Ignore this line for now. (this possibly busy waits?)
cx.waker().wake_by_ref();
Poll::Pending
}
}
}
fn number_source() -> impl Stream<Item = u8> {
stream! {
for i in 0..10 {
yield i
}
}
}
struct NumberSource {
i: u32,
}
impl Stream for NumberSource {
type Item = u32;
fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.i += 1;
Poll::Ready(Some(self.i))
}
}
struct RandomSource {
seed: u32,
modulus: u32,
a: u32,
c: u32,
}
impl Stream for RandomSource {
type Item = u32;
fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.seed = (self.a * self.seed + self.c) % self.modulus;
Poll::Ready(Some(self.seed))
}
}
fn double<S: Stream<Item = u32>>(input: S) -> impl Stream<Item = u32> {
stream! {
for await val in input {
yield val * 2
}
}
}
fn print_sink<S: Stream<Item = u32>>(input: S) -> impl Stream<Item = ()> {
stream! {
for await val in input {
print!("{}", val);
}
}
}
#[tokio::main]
async fn main() {
let number_source = NumberSource { i: 0 };
let s = print_sink(double(number_source));
pin_mut!(s); // needed for iteration
s.next().await;
}
|
