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
89
90
91
92
93
pub use inner::*;
mod inner {
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use async_io::Timer;
use futures_lite::future::Future;
use pin_project::pin_project;
/// same as `after` but return () to make it compatible as previous
pub fn sleep(duration: Duration) -> Sleeper {
Sleeper(after(duration))
}
#[pin_project]
pub struct Sleeper(#[pin] Timer);
impl Future for Sleeper {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
if this.0.poll(cx).is_ready() {
Poll::Ready(())
} else {
Poll::Pending
}
}
}
/// wait for until `duration` has elapsed.
///
/// this effectively give back control to async execution engine until duration is finished
///
/// # Examples
///
/// ```
/// use fluvio_future::timer::after;
/// use std::time::{Duration, Instant};
///
/// fluvio_future::task::run(async {
/// after(Duration::from_secs(1)).await;
/// });
/// ```
pub fn after(duration: Duration) -> Timer {
Timer::after(duration)
}
}
#[cfg(test)]
mod test {
use std::time::Duration;
use std::time::Instant;
use log::debug;
use tokio::select;
use fluvio_future::test_async;
use fluvio_future::timer::sleep;
/// test timer loop
#[test_async]
async fn test_sleep() -> Result<(), ()> {
let mut sleep_count: u16 = 0;
let time_now = Instant::now();
let mut sleep_ft = sleep(Duration::from_millis(10));
for _ in 0u16..10u16 {
select! {
_ = &mut sleep_ft => {
// fire everytime but won't make cause more delay than initial 10 ms
sleep_count += 1;
debug!("timer fired");
}
}
}
let elapsed = time_now.elapsed();
debug!("total time elaspsed: {:#?}", elapsed);
assert!(elapsed < Duration::from_millis(50));
assert!(elapsed > Duration::from_millis(10));
assert_eq!(sleep_count, 10);
Ok(())
}
}