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
|
pub enum Poll<T> {
// something happened that we want to report
Event(T),
// underlying future/stream returned NotReady, so it's safe for us to also
// return NotReady
NotReady,
// didn't do any work, so we want to return NotReady assuming at least one
// other poll method returned NotReady (if every poll method returns
// NothingToDo, something is broken)
NothingToDo,
// did some work, so we want to loop
DidWork,
// the stream has ended
Done,
}
pub fn poll_future<T, Item, Error>(
future: &mut T,
poll_fns: &'static [&'static dyn for<'a> Fn(
&'a mut T,
) -> Result<
Poll<Item>,
Error,
>],
) -> futures::Poll<Item, Error>
where
T: futures::future::Future<Item = Item, Error = Error>,
{
loop {
let mut not_ready = false;
let mut did_work = false;
for f in poll_fns {
match f(future)? {
Poll::Event(e) => return Ok(futures::Async::Ready(e)),
Poll::NotReady => not_ready = true,
Poll::NothingToDo => {}
Poll::DidWork => did_work = true,
Poll::Done => unreachable!(),
}
}
if !did_work {
if not_ready {
return Ok(futures::Async::NotReady);
} else {
unreachable!()
}
}
}
}
pub fn poll_stream<T, Item, Error>(
stream: &mut T,
poll_fns: &'static [&'static dyn for<'a> Fn(
&'a mut T,
) -> Result<
Poll<Item>,
Error,
>],
) -> futures::Poll<Option<Item>, Error>
where
T: futures::stream::Stream<Item = Item, Error = Error>,
{
loop {
let mut not_ready = false;
let mut did_work = false;
for f in poll_fns {
match f(stream)? {
Poll::Event(e) => return Ok(futures::Async::Ready(Some(e))),
Poll::NotReady => not_ready = true,
Poll::NothingToDo => {}
Poll::DidWork => did_work = true,
Poll::Done => return Ok(futures::Async::Ready(None)),
}
}
if !did_work {
if not_ready {
return Ok(futures::Async::NotReady);
} else {
unreachable!()
}
}
}
}
|
