#![deny(unsafe_code)]
use std::cmp::{self, Ord};
use std::collections::BinaryHeap;
use std::time::{Duration, Instant};
use crossbeam_channel::{Receiver, after, never};
use malloc_size_of_derive::MallocSizeOf;
pub type BoxedTimerCallback = Box<dyn Fn() + Send + 'static>;
#[derive(MallocSizeOf)]
pub struct TimerEventRequest {
#[ignore_malloc_size_of = "Size of a boxed function"]
pub callback: BoxedTimerCallback,
pub duration: Duration,
}
impl TimerEventRequest {
fn dispatch(self) {
(self.callback)()
}
}
#[derive(MallocSizeOf)]
struct ScheduledEvent {
id: TimerId,
request: TimerEventRequest,
for_time: Instant,
}
impl Ord for ScheduledEvent {
fn cmp(&self, other: &ScheduledEvent) -> cmp::Ordering {
self.for_time.cmp(&other.for_time).reverse()
}
}
impl PartialOrd for ScheduledEvent {
fn partial_cmp(&self, other: &ScheduledEvent) -> Option<cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Eq for ScheduledEvent {}
impl PartialEq for ScheduledEvent {
fn eq(&self, other: &ScheduledEvent) -> bool {
std::ptr::eq(self, other)
}
}
#[derive(Clone, Copy, MallocSizeOf, PartialEq)]
pub struct TimerId(usize);
#[derive(Default, MallocSizeOf)]
pub struct TimerScheduler {
queue: BinaryHeap<ScheduledEvent>,
current_id: usize,
}
impl TimerScheduler {
pub fn schedule_timer(&mut self, request: TimerEventRequest) -> TimerId {
let for_time = Instant::now() + request.duration;
let id = TimerId(self.current_id);
self.current_id += 1;
self.queue.push(ScheduledEvent {
id,
request,
for_time,
});
id
}
pub fn cancel_timer(&mut self, id: TimerId) {
self.queue.retain(|event| event.id != id);
}
pub fn wait_channel(&self) -> Receiver<Instant> {
self.queue
.peek()
.map(|event| {
let now = Instant::now();
if event.for_time < now {
after(Duration::ZERO)
} else {
after(event.for_time - now)
}
})
.unwrap_or_else(never)
}
pub fn dispatch_completed_timers(&mut self) {
let now = Instant::now();
loop {
match self.queue.peek() {
Some(event) if event.for_time <= now => {},
_ => break,
}
self.queue
.pop()
.expect("Expected request")
.request
.dispatch();
}
}
}