use pin_utils::unsafe_pinned;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use futures::prelude::*;
use super::delay;
use super::{Delay, Instant, TimerHandle};
#[derive(Debug)]
pub struct Interval {
delay: Delay,
interval: Duration,
}
impl Interval {
unsafe_pinned!(delay: Delay);
pub fn new(dur: Duration) -> Interval {
Interval::new_at(Instant::now() + dur, dur)
}
pub fn new_at(at: Instant, dur: Duration) -> Interval {
Interval {
delay: Delay::new_at(at),
interval: dur,
}
}
pub fn new_handle(at: Instant, dur: Duration, handle: TimerHandle) -> Interval {
Interval {
delay: Delay::new_handle(at, handle),
interval: dur,
}
}
}
impl Stream for Interval {
type Item = ();
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if Pin::new(&mut *self).delay().poll(cx).is_pending() {
return Poll::Pending;
}
let next = next_interval(delay::fires_at(&self.delay), Instant::now(), self.interval);
self.delay.reset_at(next);
Poll::Ready(Some(()))
}
}
fn duration_to_nanos(dur: Duration) -> Option<u64> {
dur.as_secs()
.checked_mul(1_000_000_000)
.and_then(|v| v.checked_add(dur.subsec_nanos() as u64))
}
fn next_interval(prev: Instant, now: Instant, interval: Duration) -> Instant {
let new = prev + interval;
if new > now {
return new;
} else {
#[allow(clippy::expect_used)]
let spent_ns =
duration_to_nanos(now.duration_since(prev)).expect("interval should be expired");
#[allow(clippy::expect_used)]
let interval_ns =
duration_to_nanos(interval).expect("interval is less that 427 thousand years");
let mult = spent_ns / interval_ns + 1;
assert!(
mult < (1 << 32),
"can't skip more than 4 billion intervals of {:?} \
(trying to skip {})",
interval,
mult
);
return prev + interval * (mult as u32);
}
}
#[cfg(test)]
mod test {
use super::next_interval;
use std::time::{Duration, Instant};
struct Timeline(Instant);
impl Timeline {
fn new() -> Timeline {
Timeline(Instant::now())
}
fn at(&self, millis: u64) -> Instant {
self.0 + Duration::from_millis(millis)
}
fn at_ns(&self, sec: u64, nanos: u32) -> Instant {
self.0 + Duration::new(sec, nanos)
}
}
fn dur(millis: u64) -> Duration {
Duration::from_millis(millis)
}
fn almost_eq(a: Instant, b: Instant) -> bool {
if a == b {
true
} else if a > b {
a - b < Duration::from_millis(1)
} else {
b - a < Duration::from_millis(1)
}
}
#[test]
fn norm_next() {
let tm = Timeline::new();
assert!(almost_eq(
next_interval(tm.at(1), tm.at(2), dur(10)),
tm.at(11)
));
assert!(almost_eq(
next_interval(tm.at(7777), tm.at(7788), dur(100)),
tm.at(7877)
));
assert!(almost_eq(
next_interval(tm.at(1), tm.at(1000), dur(2100)),
tm.at(2101)
));
}
#[test]
fn fast_forward() {
let tm = Timeline::new();
assert!(almost_eq(
next_interval(tm.at(1), tm.at(1000), dur(10)),
tm.at(1001)
));
assert!(almost_eq(
next_interval(tm.at(7777), tm.at(8888), dur(100)),
tm.at(8977)
));
assert!(almost_eq(
next_interval(tm.at(1), tm.at(10000), dur(2100)),
tm.at(10501)
));
}
#[test]
#[should_panic(expected = "can't skip more than 4 billion intervals")]
fn large_skip() {
let tm = Timeline::new();
assert_eq!(
next_interval(tm.at_ns(0, 1), tm.at_ns(25, 0), Duration::new(0, 2)),
tm.at_ns(25, 1)
);
}
}