mod api;
#[path = "virtual.rs"]
mod virtual_clock;
#[cfg(test)]
mod tests;
use std::{any::Any, collections::BTreeMap, fmt::Debug, time::Duration};
use ahash::AHashMap;
pub use api::ClockApi; use jiff::Timestamp;
use nautilus_core::{
DurationNanos, UnixNanos,
correctness::{check_positive_u64, check_valid_string_utf8},
datetime::try_datetime_to_unix_nanos,
};
use ustr::Ustr;
pub use virtual_clock::VirtualClock;
use crate::timer::{TimeEvent, TimeEventCallback, TimeEventHandler, Timer};
pub trait Clock: Debug + Any {
fn utc_now(&self) -> Timestamp {
self.timestamp_ns().to_datetime_utc()
}
fn timestamp_ns(&self) -> UnixNanos;
fn timestamp_us(&self) -> u64;
fn timestamp_ms(&self) -> u64;
fn timestamp(&self) -> f64;
fn timer_names(&self) -> Vec<&str>;
fn timer_count(&self) -> usize;
fn timer_exists(&self, name: &Ustr) -> bool;
fn register_default_handler(&mut self, callback: TimeEventCallback);
fn cancel_default_handler(&mut self);
fn cancel_callbacks(&mut self);
fn set_time_alert(
&mut self,
name: &str,
alert_time: Timestamp,
callback: Option<TimeEventCallback>,
allow_past: Option<bool>,
) -> anyhow::Result<()> {
self.set_time_alert_ns(
name,
try_datetime_to_unix_nanos(alert_time)?,
callback,
allow_past,
)
}
fn set_time_alert_ns(
&mut self,
name: &str,
alert_time_ns: UnixNanos,
callback: Option<TimeEventCallback>,
allow_past: Option<bool>,
) -> anyhow::Result<()>;
#[expect(clippy::too_many_arguments)]
fn set_timer(
&mut self,
name: &str,
interval: Duration,
start_time: Option<Timestamp>,
stop_time: Option<Timestamp>,
callback: Option<TimeEventCallback>,
allow_past: Option<bool>,
fire_immediately: Option<bool>,
) -> anyhow::Result<()> {
self.set_timer_ns(
name,
duration_to_nanos(interval)?,
start_time.map(try_datetime_to_unix_nanos).transpose()?,
stop_time.map(try_datetime_to_unix_nanos).transpose()?,
callback,
allow_past,
fire_immediately,
)
}
#[expect(clippy::too_many_arguments)]
fn set_timer_ns(
&mut self,
name: &str,
interval_ns: DurationNanos,
start_time_ns: Option<UnixNanos>,
stop_time_ns: Option<UnixNanos>,
callback: Option<TimeEventCallback>,
allow_past: Option<bool>,
fire_immediately: Option<bool>,
) -> anyhow::Result<()>;
fn next_time_ns(&self, name: &str) -> Option<UnixNanos>;
fn cancel_timer(&mut self, name: &str);
fn cancel_timers(&mut self);
fn reset(&mut self);
}
impl dyn Clock {
pub fn as_any(&self) -> &dyn std::any::Any {
self
}
pub fn as_any_mut(&mut self) -> &mut dyn std::any::Any {
self
}
}
fn duration_to_nanos(duration: Duration) -> anyhow::Result<DurationNanos> {
DurationNanos::try_from(duration)
.map_err(|_| anyhow::anyhow!("Interval exceeds u64 nanoseconds"))
}
#[derive(Debug, Default)]
pub struct CallbackRegistry {
default_callback: Option<TimeEventCallback>,
callbacks: AHashMap<Ustr, TimeEventCallback>,
}
impl CallbackRegistry {
#[must_use]
pub fn new() -> Self {
Self::default()
}
pub fn register_default_handler(&mut self, callback: TimeEventCallback) {
self.default_callback = Some(callback);
}
pub fn cancel_default_handler(&mut self) {
self.default_callback = None;
}
pub fn register_callback(&mut self, name: Ustr, callback: TimeEventCallback) {
self.callbacks.insert(name, callback);
}
#[must_use]
pub fn has_any_callback(&self, name: &Ustr) -> bool {
self.callbacks.contains_key(name) || self.default_callback.is_some()
}
#[must_use]
pub fn get_callback(&self, name: &Ustr) -> Option<TimeEventCallback> {
self.callbacks
.get(name)
.cloned()
.or_else(|| self.default_callback.clone())
}
#[must_use]
pub fn get_handler(&self, event: TimeEvent) -> TimeEventHandler {
let callback = self
.get_callback(&event.name)
.unwrap_or_else(|| panic!("Event '{}' should have associated handler", event.name));
TimeEventHandler::new(event, callback)
}
pub fn clear(&mut self) {
self.callbacks.clear();
}
}
pub fn validate_and_prepare_time_alert(
name: &str,
mut alert_time_ns: UnixNanos,
allow_past: Option<bool>,
ts_now: UnixNanos,
) -> anyhow::Result<(Ustr, UnixNanos)> {
check_valid_string_utf8(name, stringify!(name))?;
let name = Ustr::from(name);
let allow_past = allow_past.unwrap_or(true);
if alert_time_ns < ts_now {
if allow_past {
log::warn!(
"Timer '{name}' alert time {} was in the past, adjusted to current time for immediate firing",
alert_time_ns.to_rfc3339(),
);
alert_time_ns = ts_now;
} else {
anyhow::bail!(
"Timer '{name}' alert time {} was in the past (current time is {ts_now})",
alert_time_ns.to_rfc3339(),
);
}
}
Ok((name, alert_time_ns))
}
pub fn validate_and_prepare_timer(
name: &str,
interval_ns: DurationNanos,
start_time_ns: Option<UnixNanos>,
stop_time_ns: Option<UnixNanos>,
allow_past: Option<bool>,
fire_immediately: Option<bool>,
ts_now: UnixNanos,
) -> anyhow::Result<(Ustr, UnixNanos, Option<UnixNanos>, bool, bool)> {
check_valid_string_utf8(name, stringify!(name))?;
check_positive_u64(interval_ns.as_u64(), stringify!(interval_ns))?;
let name = Ustr::from(name);
let allow_past = allow_past.unwrap_or(true);
let fire_immediately = fire_immediately.unwrap_or(false);
let start_time_ns = start_time_ns
.filter(|start_time_ns| *start_time_ns != 0)
.unwrap_or(ts_now);
let next_event_time = if fire_immediately {
start_time_ns
} else {
start_time_ns.checked_add(interval_ns).ok_or_else(|| {
anyhow::anyhow!("Timer '{name}' first event time exceeds UnixNanos range")
})?
};
if !allow_past && next_event_time < ts_now {
anyhow::bail!(
"Timer '{name}' next event time {} would be in the past (current time is {ts_now})",
next_event_time.to_rfc3339(),
);
}
if let Some(stop_time) = stop_time_ns {
if stop_time <= start_time_ns {
anyhow::bail!(
"Timer '{name}' stop time {} must be after start time {}",
stop_time.to_rfc3339(),
start_time_ns.to_rfc3339(),
);
}
if !allow_past && stop_time <= ts_now {
anyhow::bail!(
"Timer '{name}' stop time {} is in the past (current time is {ts_now})",
stop_time.to_rfc3339(),
);
}
}
Ok((
name,
start_time_ns,
stop_time_ns,
allow_past,
fire_immediately,
))
}
pub(crate) fn replace_existing_timer<T: Timer>(timers: &mut BTreeMap<Ustr, T>, name: &Ustr) {
let Some(mut timer) = timers.remove(name) else {
return;
};
if timer.is_expired() {
return;
}
timer.cancel();
log::warn!("Timer '{name}' replaced");
}