1#![warn(clippy::clone_on_ref_ptr)]
19
20use std::{
21 any::Any,
22 cell::RefCell,
23 collections::{BTreeMap, BinaryHeap},
24 fmt::Debug,
25 ops::Deref,
26 time::Duration,
27};
28
29use ahash::AHashMap;
30use chrono::{DateTime, Utc};
31use nautilus_core::{
32 AtomicTime, UUID4, UnixNanos,
33 correctness::{check_positive_u64, check_predicate_true, check_valid_string_utf8},
34 datetime::NANOSECONDS_IN_SECOND,
35 string::formatting::Separable,
36};
37use ustr::Ustr;
38
39use crate::timer::{
40 ScheduledTimeEvent, TestTimer, TimeEvent, TimeEventCallback, TimeEventHandler, Timer,
41 create_valid_interval,
42};
43
44pub trait Clock: Debug + Any {
50 fn utc_now(&self) -> DateTime<Utc> {
52 DateTime::from_timestamp_nanos(self.timestamp_ns().as_i64())
53 }
54
55 fn timestamp_ns(&self) -> UnixNanos;
57
58 fn timestamp_us(&self) -> u64;
60
61 fn timestamp_ms(&self) -> u64;
63
64 fn timestamp(&self) -> f64;
66
67 fn timer_names(&self) -> Vec<&str>;
69
70 fn timer_count(&self) -> usize;
72
73 fn timer_exists(&self, name: &Ustr) -> bool;
75
76 fn register_default_handler(&mut self, callback: TimeEventCallback);
79
80 fn cancel_default_handler(&mut self);
86
87 fn cancel_callbacks(&mut self);
93
94 fn get_handler(&self, event: TimeEvent) -> TimeEventHandler;
98
99 fn set_time_alert(
113 &mut self,
114 name: &str,
115 alert_time: DateTime<Utc>,
116 callback: Option<TimeEventCallback>,
117 allow_past: Option<bool>,
118 ) -> anyhow::Result<()> {
119 self.set_time_alert_ns(name, alert_time.into(), callback, allow_past)
120 }
121
122 fn set_time_alert_ns(
143 &mut self,
144 name: &str,
145 alert_time_ns: UnixNanos,
146 callback: Option<TimeEventCallback>,
147 allow_past: Option<bool>,
148 ) -> anyhow::Result<()>;
149
150 #[expect(clippy::too_many_arguments)]
166 fn set_timer(
167 &mut self,
168 name: &str,
169 interval: Duration,
170 start_time: Option<DateTime<Utc>>,
171 stop_time: Option<DateTime<Utc>>,
172 callback: Option<TimeEventCallback>,
173 allow_past: Option<bool>,
174 fire_immediately: Option<bool>,
175 ) -> anyhow::Result<()> {
176 self.set_timer_ns(
177 name,
178 interval.as_nanos() as u64,
179 start_time.map(UnixNanos::from),
180 stop_time.map(UnixNanos::from),
181 callback,
182 allow_past,
183 fire_immediately,
184 )
185 }
186
187 #[expect(clippy::too_many_arguments)]
215 fn set_timer_ns(
216 &mut self,
217 name: &str,
218 interval_ns: u64,
219 start_time_ns: Option<UnixNanos>,
220 stop_time_ns: Option<UnixNanos>,
221 callback: Option<TimeEventCallback>,
222 allow_past: Option<bool>,
223 fire_immediately: Option<bool>,
224 ) -> anyhow::Result<()>;
225
226 fn next_time_ns(&self, name: &str) -> Option<UnixNanos>;
230
231 fn cancel_timer(&mut self, name: &str);
233
234 fn cancel_timers(&mut self);
236
237 fn reset(&mut self);
239}
240
241impl dyn Clock {
242 pub fn as_any(&self) -> &dyn std::any::Any {
244 self
245 }
246 pub fn as_any_mut(&mut self) -> &mut dyn std::any::Any {
248 self
249 }
250}
251
252#[derive(Debug)]
254pub struct ClockApi<'a> {
255 backing: ClockApiBacking<'a>,
256}
257
258enum ClockApiBacking<'a> {
259 Native(&'a RefCell<dyn Clock>),
260 Handlers(ClockApiHandlers<'a>),
261}
262
263struct ClockApiHandlers<'a> {
264 timestamp_ns: Box<dyn Fn() -> UnixNanos + 'a>,
265 set_time_alert_ns: Box<SetTimeAlertNsHandler<'a>>,
266 set_timer_ns: Box<SetTimerNsHandler<'a>>,
267 timer_names: Box<dyn Fn() -> Vec<String> + 'a>,
268 timer_count: Box<dyn Fn() -> usize + 'a>,
269 timer_exists: Box<dyn Fn(&str) -> bool + 'a>,
270 next_time_ns: Box<NextTimeNsHandler<'a>>,
271 cancel_timer: Box<dyn Fn(&str) + 'a>,
272 cancel_timers: Box<dyn Fn() + 'a>,
273}
274
275impl Debug for ClockApiBacking<'_> {
276 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
277 match self {
278 Self::Native(_) => f.write_str("Native"),
279 Self::Handlers(_) => f.write_str("Handlers"),
280 }
281 }
282}
283
284type SetTimeAlertNsHandler<'a> =
285 dyn Fn(&str, UnixNanos, Option<TimeEventCallback>, Option<bool>) -> anyhow::Result<()> + 'a;
286type NextTimeNsHandler<'a> = dyn Fn(&str) -> Option<UnixNanos> + 'a;
287type SetTimerNsHandler<'a> = dyn Fn(
288 &str,
289 u64,
290 Option<UnixNanos>,
291 Option<UnixNanos>,
292 Option<TimeEventCallback>,
293 Option<bool>,
294 Option<bool>,
295 ) -> anyhow::Result<()>
296 + 'a;
297
298impl<'a> ClockApi<'a> {
299 pub(crate) fn new(clock: &'a RefCell<dyn Clock>) -> Self {
300 Self {
301 backing: ClockApiBacking::Native(clock),
302 }
303 }
304
305 #[doc(hidden)]
307 #[must_use]
308 #[expect(
309 clippy::too_many_arguments,
310 reason = "clock API backing mirrors the full ClockApi surface"
311 )]
312 pub fn from_handlers<
313 TimestampNs,
314 SetTimeAlertNs,
315 SetTimerNs,
316 TimerNames,
317 TimerCount,
318 TimerExists,
319 NextTimeNs,
320 CancelTimer,
321 CancelTimers,
322 >(
323 timestamp_ns: TimestampNs,
324 set_time_alert_ns: SetTimeAlertNs,
325 set_timer_ns: SetTimerNs,
326 timer_names: TimerNames,
327 timer_count: TimerCount,
328 timer_exists: TimerExists,
329 next_time_ns: NextTimeNs,
330 cancel_timer: CancelTimer,
331 cancel_timers: CancelTimers,
332 ) -> Self
333 where
334 TimestampNs: Fn() -> UnixNanos + 'a,
335 SetTimeAlertNs:
336 Fn(&str, UnixNanos, Option<TimeEventCallback>, Option<bool>) -> anyhow::Result<()> + 'a,
337 SetTimerNs: Fn(
338 &str,
339 u64,
340 Option<UnixNanos>,
341 Option<UnixNanos>,
342 Option<TimeEventCallback>,
343 Option<bool>,
344 Option<bool>,
345 ) -> anyhow::Result<()>
346 + 'a,
347 TimerNames: Fn() -> Vec<String> + 'a,
348 TimerCount: Fn() -> usize + 'a,
349 TimerExists: Fn(&str) -> bool + 'a,
350 NextTimeNs: Fn(&str) -> Option<UnixNanos> + 'a,
351 CancelTimer: Fn(&str) + 'a,
352 CancelTimers: Fn() + 'a,
353 {
354 Self {
355 backing: ClockApiBacking::Handlers(ClockApiHandlers {
356 timestamp_ns: Box::new(timestamp_ns),
357 set_time_alert_ns: Box::new(set_time_alert_ns),
358 set_timer_ns: Box::new(set_timer_ns),
359 timer_names: Box::new(timer_names),
360 timer_count: Box::new(timer_count),
361 timer_exists: Box::new(timer_exists),
362 next_time_ns: Box::new(next_time_ns),
363 cancel_timer: Box::new(cancel_timer),
364 cancel_timers: Box::new(cancel_timers),
365 }),
366 }
367 }
368
369 #[must_use]
375 pub fn timestamp_ns(&self) -> UnixNanos {
376 match &self.backing {
377 ClockApiBacking::Native(clock) => clock.borrow().timestamp_ns(),
378 ClockApiBacking::Handlers(handlers) => (handlers.timestamp_ns)(),
379 }
380 }
381
382 #[must_use]
388 pub fn timestamp_us(&self) -> u64 {
389 match &self.backing {
390 ClockApiBacking::Native(clock) => clock.borrow().timestamp_us(),
391 ClockApiBacking::Handlers(handlers) => (handlers.timestamp_ns)().as_micros(),
392 }
393 }
394
395 #[must_use]
401 pub fn timestamp_ms(&self) -> u64 {
402 match &self.backing {
403 ClockApiBacking::Native(clock) => clock.borrow().timestamp_ms(),
404 ClockApiBacking::Handlers(handlers) => (handlers.timestamp_ns)().as_millis(),
405 }
406 }
407
408 #[must_use]
414 pub fn timestamp(&self) -> f64 {
415 match &self.backing {
416 ClockApiBacking::Native(clock) => clock.borrow().timestamp(),
417 ClockApiBacking::Handlers(handlers) => {
418 (handlers.timestamp_ns)().as_f64() / (NANOSECONDS_IN_SECOND as f64)
419 }
420 }
421 }
422
423 #[must_use]
429 pub fn utc_now(&self) -> DateTime<Utc> {
430 match &self.backing {
431 ClockApiBacking::Native(clock) => clock.borrow().utc_now(),
432 ClockApiBacking::Handlers(handlers) => {
433 DateTime::from_timestamp_nanos((handlers.timestamp_ns)().as_i64())
434 }
435 }
436 }
437
438 pub fn set_time_alert(
449 &self,
450 name: &str,
451 alert_time: DateTime<Utc>,
452 callback: Option<TimeEventCallback>,
453 allow_past: Option<bool>,
454 ) -> anyhow::Result<()> {
455 match &self.backing {
456 ClockApiBacking::Native(clock) => clock
457 .borrow_mut()
458 .set_time_alert(name, alert_time, callback, allow_past),
459 ClockApiBacking::Handlers(handlers) => {
460 (handlers.set_time_alert_ns)(name, alert_time.into(), callback, allow_past)
461 }
462 }
463 }
464
465 pub fn set_time_alert_ns(
476 &self,
477 name: &str,
478 alert_time_ns: UnixNanos,
479 callback: Option<TimeEventCallback>,
480 allow_past: Option<bool>,
481 ) -> anyhow::Result<()> {
482 match &self.backing {
483 ClockApiBacking::Native(clock) => {
484 clock
485 .borrow_mut()
486 .set_time_alert_ns(name, alert_time_ns, callback, allow_past)
487 }
488 ClockApiBacking::Handlers(handlers) => {
489 (handlers.set_time_alert_ns)(name, alert_time_ns, callback, allow_past)
490 }
491 }
492 }
493
494 #[expect(clippy::too_many_arguments, reason = "timer scheduling mirrors Clock")]
505 pub fn set_timer(
506 &self,
507 name: &str,
508 interval: Duration,
509 start_time: Option<DateTime<Utc>>,
510 stop_time: Option<DateTime<Utc>>,
511 callback: Option<TimeEventCallback>,
512 allow_past: Option<bool>,
513 fire_immediately: Option<bool>,
514 ) -> anyhow::Result<()> {
515 match &self.backing {
516 ClockApiBacking::Native(clock) => clock.borrow_mut().set_timer(
517 name,
518 interval,
519 start_time,
520 stop_time,
521 callback,
522 allow_past,
523 fire_immediately,
524 ),
525 ClockApiBacking::Handlers(handlers) => (handlers.set_timer_ns)(
526 name,
527 interval.as_nanos() as u64,
528 start_time.map(UnixNanos::from),
529 stop_time.map(UnixNanos::from),
530 callback,
531 allow_past,
532 fire_immediately,
533 ),
534 }
535 }
536
537 #[expect(clippy::too_many_arguments, reason = "timer scheduling mirrors Clock")]
548 pub fn set_timer_ns(
549 &self,
550 name: &str,
551 interval_ns: u64,
552 start_time_ns: Option<UnixNanos>,
553 stop_time_ns: Option<UnixNanos>,
554 callback: Option<TimeEventCallback>,
555 allow_past: Option<bool>,
556 fire_immediately: Option<bool>,
557 ) -> anyhow::Result<()> {
558 match &self.backing {
559 ClockApiBacking::Native(clock) => clock.borrow_mut().set_timer_ns(
560 name,
561 interval_ns,
562 start_time_ns,
563 stop_time_ns,
564 callback,
565 allow_past,
566 fire_immediately,
567 ),
568 ClockApiBacking::Handlers(handlers) => (handlers.set_timer_ns)(
569 name,
570 interval_ns,
571 start_time_ns,
572 stop_time_ns,
573 callback,
574 allow_past,
575 fire_immediately,
576 ),
577 }
578 }
579
580 #[must_use]
586 pub fn timer_names(&self) -> Vec<String> {
587 match &self.backing {
588 ClockApiBacking::Native(clock) => clock
589 .borrow()
590 .timer_names()
591 .into_iter()
592 .map(str::to_string)
593 .collect(),
594 ClockApiBacking::Handlers(handlers) => (handlers.timer_names)(),
595 }
596 }
597
598 #[must_use]
604 pub fn timer_count(&self) -> usize {
605 match &self.backing {
606 ClockApiBacking::Native(clock) => clock.borrow().timer_count(),
607 ClockApiBacking::Handlers(handlers) => (handlers.timer_count)(),
608 }
609 }
610
611 #[must_use]
617 pub fn timer_exists(&self, name: &str) -> bool {
618 match &self.backing {
619 ClockApiBacking::Native(clock) => clock.borrow().timer_exists(&Ustr::from(name)),
620 ClockApiBacking::Handlers(handlers) => (handlers.timer_exists)(name),
621 }
622 }
623
624 #[must_use]
630 pub fn next_time_ns(&self, name: &str) -> Option<UnixNanos> {
631 match &self.backing {
632 ClockApiBacking::Native(clock) => clock.borrow().next_time_ns(name),
633 ClockApiBacking::Handlers(handlers) => (handlers.next_time_ns)(name),
634 }
635 }
636
637 pub fn cancel_timer(&self, name: &str) {
643 match &self.backing {
644 ClockApiBacking::Native(clock) => clock.borrow_mut().cancel_timer(name),
645 ClockApiBacking::Handlers(handlers) => (handlers.cancel_timer)(name),
646 }
647 }
648
649 pub fn cancel_timers(&self) {
655 match &self.backing {
656 ClockApiBacking::Native(clock) => clock.borrow_mut().cancel_timers(),
657 ClockApiBacking::Handlers(handlers) => (handlers.cancel_timers)(),
658 }
659 }
660}
661
662#[derive(Debug, Default)]
667pub struct CallbackRegistry {
668 default_callback: Option<TimeEventCallback>,
669 callbacks: AHashMap<Ustr, TimeEventCallback>,
670}
671
672impl CallbackRegistry {
673 #[must_use]
675 pub fn new() -> Self {
676 Self {
677 default_callback: None,
678 callbacks: AHashMap::new(),
679 }
680 }
681
682 pub fn register_default_handler(&mut self, callback: TimeEventCallback) {
684 self.default_callback = Some(callback);
685 }
686
687 pub fn cancel_default_handler(&mut self) {
689 self.default_callback = None;
690 }
691
692 pub fn register_callback(&mut self, name: Ustr, callback: TimeEventCallback) {
694 self.callbacks.insert(name, callback);
695 }
696
697 #[must_use]
699 pub fn has_any_callback(&self, name: &Ustr) -> bool {
700 self.callbacks.contains_key(name) || self.default_callback.is_some()
701 }
702
703 #[must_use]
705 pub fn get_callback(&self, name: &Ustr) -> Option<TimeEventCallback> {
706 self.callbacks
707 .get(name)
708 .cloned()
709 .or_else(|| self.default_callback.clone())
710 }
711
712 #[must_use]
718 pub fn get_handler(&self, event: TimeEvent) -> TimeEventHandler {
719 let callback = self
720 .get_callback(&event.name)
721 .unwrap_or_else(|| panic!("Event '{}' should have associated handler", event.name));
722
723 TimeEventHandler::new(event, callback)
724 }
725
726 pub fn clear(&mut self) {
728 self.callbacks.clear();
729 }
730}
731
732pub fn validate_and_prepare_time_alert(
740 name: &str,
741 mut alert_time_ns: UnixNanos,
742 allow_past: Option<bool>,
743 ts_now: UnixNanos,
744) -> anyhow::Result<(Ustr, UnixNanos)> {
745 check_valid_string_utf8(name, stringify!(name))?;
746
747 let name = Ustr::from(name);
748 let allow_past = allow_past.unwrap_or(true);
749
750 if alert_time_ns < ts_now {
751 if allow_past {
752 alert_time_ns = ts_now;
753 log::warn!(
754 "Timer '{name}' alert time {} was in the past, adjusted to current time for immediate firing",
755 alert_time_ns.to_rfc3339(),
756 );
757 } else {
758 anyhow::bail!(
759 "Timer '{name}' alert time {} was in the past (current time is {ts_now})",
760 alert_time_ns.to_rfc3339(),
761 );
762 }
763 }
764
765 Ok((name, alert_time_ns))
766}
767
768pub fn validate_and_prepare_timer(
777 name: &str,
778 interval_ns: u64,
779 start_time_ns: Option<UnixNanos>,
780 stop_time_ns: Option<UnixNanos>,
781 allow_past: Option<bool>,
782 fire_immediately: Option<bool>,
783 ts_now: UnixNanos,
784) -> anyhow::Result<(Ustr, UnixNanos, Option<UnixNanos>, bool, bool)> {
785 check_valid_string_utf8(name, stringify!(name))?;
786 check_positive_u64(interval_ns, stringify!(interval_ns))?;
787
788 let name = Ustr::from(name);
789 let allow_past = allow_past.unwrap_or(true);
790 let fire_immediately = fire_immediately.unwrap_or(false);
791
792 let mut start_time_ns = start_time_ns.unwrap_or_default();
793
794 if start_time_ns == 0 {
795 start_time_ns = ts_now;
797 } else if !allow_past {
798 let next_event_time = if fire_immediately {
799 start_time_ns
800 } else {
801 start_time_ns + interval_ns
802 };
803
804 if next_event_time < ts_now {
805 anyhow::bail!(
806 "Timer '{name}' next event time {} would be in the past (current time is {ts_now})",
807 next_event_time.to_rfc3339(),
808 );
809 }
810 }
811
812 if let Some(stop_time) = stop_time_ns {
813 if stop_time <= start_time_ns {
814 anyhow::bail!(
815 "Timer '{name}' stop time {} must be after start time {}",
816 stop_time.to_rfc3339(),
817 start_time_ns.to_rfc3339(),
818 );
819 }
820
821 if !allow_past && stop_time <= ts_now {
822 anyhow::bail!(
823 "Timer '{name}' stop time {} is in the past (current time is {ts_now})",
824 stop_time.to_rfc3339(),
825 );
826 }
827 }
828
829 Ok((
830 name,
831 start_time_ns,
832 stop_time_ns,
833 allow_past,
834 fire_immediately,
835 ))
836}
837
838#[derive(Debug)]
846pub struct TestClock {
847 time: AtomicTime,
848 timers: BTreeMap<Ustr, TestTimer>,
849 timer_queue: BinaryHeap<ScheduledTimeEvent>,
850 callbacks: CallbackRegistry,
851}
852
853impl TestClock {
854 #[must_use]
856 pub fn new() -> Self {
857 Self {
858 time: AtomicTime::new(false, UnixNanos::default()),
859 timers: BTreeMap::new(),
860 timer_queue: BinaryHeap::new(),
861 callbacks: CallbackRegistry::new(),
862 }
863 }
864
865 #[must_use]
867 pub const fn get_timers(&self) -> &BTreeMap<Ustr, TestTimer> {
868 &self.timers
869 }
870
871 pub fn advance_time(&mut self, to_time_ns: UnixNanos, set_time: bool) -> Vec<TimeEvent> {
888 const WARN_TIME_EVENTS_THRESHOLD: usize = 1_000_000;
889
890 let from_time_ns = self.time.get_time_ns();
891
892 assert!(
893 to_time_ns >= from_time_ns,
894 "Invariant: time must be non-decreasing, `to_time_ns` {to_time_ns} < `from_time_ns` {from_time_ns}"
895 );
896
897 if set_time {
898 self.time.set_time(to_time_ns);
899 }
900
901 let mut events: Vec<TimeEvent> = Vec::new();
902
903 while self
904 .timer_queue
905 .peek()
906 .is_some_and(|entry| entry.0.ts_event <= to_time_ns)
907 {
908 let entry = self
909 .timer_queue
910 .pop()
911 .expect("timer queue peeked Some but pop returned None");
912
913 let Some((event, next_event)) = self.advance_timer_from_entry(&entry.0) else {
914 continue;
915 };
916
917 events.push(event);
918 if let Some(next_event) = next_event {
919 self.timer_queue.push(next_event);
920 }
921 }
922
923 self.compact_timer_queue_if_needed();
924
925 if events.len() >= WARN_TIME_EVENTS_THRESHOLD {
926 log::warn!(
927 "Allocated {} time events during clock advancement from {} to {}, \
928 consider stopping the timer between large time ranges with no data points",
929 events.len().separate_with_commas(),
930 from_time_ns,
931 to_time_ns
932 );
933 }
934
935 events.sort_by(|a, b| {
936 a.ts_event
937 .cmp(&b.ts_event)
938 .then_with(|| a.name.cmp(&b.name))
939 });
940 events
941 }
942
943 #[must_use]
953 pub fn match_handlers(&self, events: Vec<TimeEvent>) -> Vec<TimeEventHandler> {
954 events
955 .into_iter()
956 .map(|event| self.callbacks.get_handler(event))
957 .collect()
958 }
959
960 fn replace_existing_timer_if_needed(&mut self, name: &Ustr) {
961 replace_existing_timer(&mut self.timers, name);
962 self.compact_timer_queue_if_needed();
963 }
964
965 fn insert_timer(&mut self, timer: TestTimer) {
966 self.timer_queue.push(Self::scheduled_event(&timer));
967 self.timers.insert(timer.name, timer);
968 self.compact_timer_queue_if_needed();
969 }
970
971 fn advance_timer_from_entry(
972 &mut self,
973 entry: &TimeEvent,
974 ) -> Option<(TimeEvent, Option<ScheduledTimeEvent>)> {
975 let timer = self.timers.get_mut(&entry.name)?;
976 if timer.next_time_ns() != entry.ts_event {
977 return None;
978 }
979
980 let Some((event, _)) = timer.next() else {
981 self.timers.remove(&entry.name);
982 return None;
983 };
984
985 let next_entry = if timer.is_expired() {
986 self.timers.remove(&entry.name);
987 None
988 } else {
989 Some(Self::scheduled_event(timer))
990 };
991
992 Some((event, next_entry))
993 }
994
995 fn compact_timer_queue_if_needed(&mut self) {
996 if self.timer_queue.len() > self.timers.len().saturating_mul(2) {
997 self.compact_timer_queue();
998 }
999 }
1000
1001 fn compact_timer_queue(&mut self) {
1002 self.timer_queue = self.timers.values().map(Self::scheduled_event).collect();
1003 }
1004
1005 fn scheduled_event(timer: &TestTimer) -> ScheduledTimeEvent {
1006 ScheduledTimeEvent::new(TimeEvent::new(
1007 timer.name,
1008 UUID4::new(),
1009 timer.next_time_ns(),
1010 timer.next_time_ns(),
1011 ))
1012 }
1013}
1014
1015impl Default for TestClock {
1016 fn default() -> Self {
1018 Self::new()
1019 }
1020}
1021
1022impl Deref for TestClock {
1023 type Target = AtomicTime;
1024
1025 fn deref(&self) -> &Self::Target {
1026 &self.time
1027 }
1028}
1029
1030impl Clock for TestClock {
1031 fn timestamp_ns(&self) -> UnixNanos {
1032 self.time.get_time_ns()
1033 }
1034
1035 fn timestamp_us(&self) -> u64 {
1036 self.time.get_time_us()
1037 }
1038
1039 fn timestamp_ms(&self) -> u64 {
1040 self.time.get_time_ms()
1041 }
1042
1043 fn timestamp(&self) -> f64 {
1044 self.time.get_time()
1045 }
1046
1047 fn timer_names(&self) -> Vec<&str> {
1048 self.timers
1049 .iter()
1050 .filter(|(_, timer)| !timer.is_expired())
1051 .map(|(k, _)| k.as_str())
1052 .collect()
1053 }
1054
1055 fn timer_count(&self) -> usize {
1056 self.timers
1057 .iter()
1058 .filter(|(_, timer)| !timer.is_expired())
1059 .count()
1060 }
1061
1062 fn timer_exists(&self, name: &Ustr) -> bool {
1063 self.timers.contains_key(name)
1064 }
1065
1066 fn register_default_handler(&mut self, callback: TimeEventCallback) {
1067 self.callbacks.register_default_handler(callback);
1068 }
1069
1070 fn cancel_default_handler(&mut self) {
1071 self.callbacks.cancel_default_handler();
1072 }
1073
1074 fn cancel_callbacks(&mut self) {
1075 self.callbacks.clear();
1076 }
1077
1078 fn get_handler(&self, event: TimeEvent) -> TimeEventHandler {
1084 self.callbacks.get_handler(event)
1085 }
1086
1087 fn set_time_alert_ns(
1088 &mut self,
1089 name: &str,
1090 alert_time_ns: UnixNanos,
1091 callback: Option<TimeEventCallback>,
1092 allow_past: Option<bool>,
1093 ) -> anyhow::Result<()> {
1094 let ts_now = self.get_time_ns();
1095 let (name, alert_time_ns) =
1096 validate_and_prepare_time_alert(name, alert_time_ns, allow_past, ts_now)?;
1097
1098 self.replace_existing_timer_if_needed(&name);
1099
1100 check_predicate_true(
1101 callback.is_some() | self.callbacks.has_any_callback(&name),
1102 "No callbacks provided",
1103 )?;
1104
1105 if let Some(callback) = callback {
1106 self.callbacks.register_callback(name, callback);
1107 }
1108
1109 let interval_ns = create_valid_interval((alert_time_ns - ts_now).into());
1111 let fire_immediately = alert_time_ns == ts_now;
1112
1113 let timer = TestTimer::new(
1114 name,
1115 interval_ns,
1116 ts_now,
1117 Some(alert_time_ns),
1118 fire_immediately,
1119 );
1120 self.insert_timer(timer);
1121
1122 Ok(())
1123 }
1124
1125 fn set_timer_ns(
1126 &mut self,
1127 name: &str,
1128 interval_ns: u64,
1129 start_time_ns: Option<UnixNanos>,
1130 stop_time_ns: Option<UnixNanos>,
1131 callback: Option<TimeEventCallback>,
1132 allow_past: Option<bool>,
1133 fire_immediately: Option<bool>,
1134 ) -> anyhow::Result<()> {
1135 let ts_now = self.get_time_ns();
1136 let (name, start_time_ns, stop_time_ns, _allow_past, fire_immediately) =
1137 validate_and_prepare_timer(
1138 name,
1139 interval_ns,
1140 start_time_ns,
1141 stop_time_ns,
1142 allow_past,
1143 fire_immediately,
1144 ts_now,
1145 )?;
1146
1147 check_predicate_true(
1148 callback.is_some() | self.callbacks.has_any_callback(&name),
1149 "No callbacks provided",
1150 )?;
1151
1152 self.replace_existing_timer_if_needed(&name);
1153
1154 if let Some(callback) = callback {
1155 self.callbacks.register_callback(name, callback);
1156 }
1157
1158 let interval_ns = create_valid_interval(interval_ns);
1159
1160 let timer = TestTimer::new(
1161 name,
1162 interval_ns,
1163 start_time_ns,
1164 stop_time_ns,
1165 fire_immediately,
1166 );
1167 self.insert_timer(timer);
1168
1169 Ok(())
1170 }
1171
1172 fn next_time_ns(&self, name: &str) -> Option<UnixNanos> {
1173 self.timers
1174 .get(&Ustr::from(name))
1175 .map(TestTimer::next_time_ns)
1176 }
1177
1178 fn cancel_timer(&mut self, name: &str) {
1179 let timer = self.timers.remove(&Ustr::from(name));
1180 if let Some(mut timer) = timer {
1181 timer.cancel();
1182 }
1183 self.compact_timer_queue_if_needed();
1184 }
1185
1186 fn cancel_timers(&mut self) {
1187 for timer in &mut self.timers.values_mut() {
1188 timer.cancel();
1189 }
1190
1191 self.timers.clear();
1192 self.timer_queue.clear();
1193 }
1194
1195 fn reset(&mut self) {
1196 self.time = AtomicTime::new(false, UnixNanos::default());
1197 self.timers = BTreeMap::new();
1198 self.timer_queue = BinaryHeap::new();
1199 self.callbacks.clear();
1200 }
1201}
1202
1203pub(crate) fn replace_existing_timer<T: Timer>(timers: &mut BTreeMap<Ustr, T>, name: &Ustr) {
1204 let is_expired = timers.get(name).map(T::is_expired);
1205 match is_expired {
1206 Some(true) => {
1207 timers.remove(name);
1208 }
1209 Some(false) => {
1210 if let Some(mut timer) = timers.remove(name) {
1211 timer.cancel();
1212 }
1213 log::warn!("Timer '{name}' replaced");
1214 }
1215 None => {}
1216 }
1217}
1218
1219#[cfg(test)]
1220mod tests {
1221 use std::{
1222 cell::RefCell,
1223 sync::{Arc, Mutex},
1224 time::Duration,
1225 };
1226
1227 use nautilus_core::{MUTEX_POISONED, UnixNanos};
1228 use rstest::{fixture, rstest};
1229 use ustr::Ustr;
1230
1231 use super::*;
1232 use crate::timer::{TimeEvent, TimeEventCallback};
1233
1234 #[derive(Debug, Default)]
1235 struct TestCallback {
1236 called: Arc<Mutex<bool>>,
1238 }
1239
1240 impl TestCallback {
1241 fn new(called: Arc<Mutex<bool>>) -> Self {
1242 Self { called }
1243 }
1244 }
1245
1246 impl From<TestCallback> for TimeEventCallback {
1247 fn from(callback: TestCallback) -> Self {
1248 Self::from(move |_event: TimeEvent| {
1249 if let Ok(mut called) = callback.called.lock() {
1250 *called = true;
1251 }
1252 })
1253 }
1254 }
1255
1256 #[fixture]
1257 pub fn test_clock() -> TestClock {
1258 let mut clock = TestClock::new();
1259 clock.register_default_handler(TestCallback::default().into());
1260 clock
1261 }
1262
1263 #[rstest]
1264 fn test_time_monotonicity(mut test_clock: TestClock) {
1265 let initial_time = test_clock.timestamp_ns();
1266 test_clock.advance_time(UnixNanos::from(*initial_time + 1000), true);
1267 assert!(test_clock.timestamp_ns() > initial_time);
1268 }
1269
1270 #[rstest]
1271 fn test_timer_registration(mut test_clock: TestClock) {
1272 test_clock
1273 .set_time_alert_ns(
1274 "test_timer",
1275 (*test_clock.timestamp_ns() + 1000).into(),
1276 None,
1277 None,
1278 )
1279 .unwrap();
1280 assert_eq!(test_clock.timer_count(), 1);
1281 assert_eq!(test_clock.timer_names(), vec!["test_timer"]);
1282 }
1283
1284 #[rstest]
1285 fn test_timer_expiration(mut test_clock: TestClock) {
1286 let alert_time = (*test_clock.timestamp_ns() + 1000).into();
1287 test_clock
1288 .set_time_alert_ns("test_timer", alert_time, None, None)
1289 .unwrap();
1290 let events = test_clock.advance_time(alert_time, true);
1291 assert_eq!(events.len(), 1);
1292 assert_eq!(events[0].name.as_str(), "test_timer");
1293 }
1294
1295 #[rstest]
1296 fn test_timer_cancellation(mut test_clock: TestClock) {
1297 test_clock
1298 .set_time_alert_ns(
1299 "test_timer",
1300 (*test_clock.timestamp_ns() + 1000).into(),
1301 None,
1302 None,
1303 )
1304 .unwrap();
1305 assert_eq!(test_clock.timer_count(), 1);
1306 test_clock.cancel_timer("test_timer");
1307 assert_eq!(test_clock.timer_count(), 0);
1308 }
1309
1310 #[rstest]
1311 fn test_time_advancement(mut test_clock: TestClock) {
1312 let start_time = test_clock.timestamp_ns();
1313 test_clock
1314 .set_timer_ns("test_timer", 1000, Some(start_time), None, None, None, None)
1315 .unwrap();
1316 let events = test_clock.advance_time(UnixNanos::from(*start_time + 2500), true);
1317 assert_eq!(events.len(), 2);
1318 assert_eq!(*events[0].ts_event, *start_time + 1000);
1319 assert_eq!(*events[1].ts_event, *start_time + 2000);
1320 }
1321
1322 #[rstest]
1323 fn test_default_and_custom_callbacks() {
1324 let mut clock = TestClock::new();
1325 let default_called = Arc::new(Mutex::new(false));
1326 let custom_called = Arc::new(Mutex::new(false));
1327
1328 let default_callback = TestCallback::new(Arc::clone(&default_called));
1329 let custom_callback = TestCallback::new(Arc::clone(&custom_called));
1330
1331 clock.register_default_handler(TimeEventCallback::from(default_callback));
1332 clock
1333 .set_time_alert_ns(
1334 "default_timer",
1335 (*clock.timestamp_ns() + 1000).into(),
1336 None,
1337 None,
1338 )
1339 .unwrap();
1340 clock
1341 .set_time_alert_ns(
1342 "custom_timer",
1343 (*clock.timestamp_ns() + 1000).into(),
1344 Some(TimeEventCallback::from(custom_callback)),
1345 None,
1346 )
1347 .unwrap();
1348
1349 let events = clock.advance_time(UnixNanos::from(*clock.timestamp_ns() + 1000), true);
1350 let handlers = clock.match_handlers(events);
1351
1352 for handler in handlers {
1353 handler.callback.call(handler.event);
1354 }
1355
1356 assert!(*default_called.lock().expect(MUTEX_POISONED));
1357 assert!(*custom_called.lock().expect(MUTEX_POISONED));
1358 }
1359
1360 #[rstest]
1361 fn test_timer_with_rust_local_callback() {
1362 use std::{cell::RefCell, rc::Rc};
1363
1364 let mut clock = TestClock::new();
1365 let call_count = Rc::new(RefCell::new(0_u32));
1366 let call_count_clone = Rc::clone(&call_count);
1367
1368 let callback: Rc<dyn Fn(TimeEvent)> = Rc::new(move |_event: TimeEvent| {
1370 *call_count_clone.borrow_mut() += 1;
1371 });
1372
1373 clock
1374 .set_time_alert_ns(
1375 "local_timer",
1376 (*clock.timestamp_ns() + 1000).into(),
1377 Some(TimeEventCallback::from(callback)),
1378 None,
1379 )
1380 .unwrap();
1381
1382 let events = clock.advance_time(UnixNanos::from(*clock.timestamp_ns() + 1000), true);
1383 let handlers = clock.match_handlers(events);
1384
1385 for handler in handlers {
1386 handler.callback.call(handler.event);
1387 }
1388
1389 assert_eq!(*call_count.borrow(), 1);
1390 }
1391
1392 #[rstest]
1393 fn test_multiple_timers(mut test_clock: TestClock) {
1394 let start_time = test_clock.timestamp_ns();
1395 test_clock
1396 .set_timer_ns("timer1", 1000, Some(start_time), None, None, None, None)
1397 .unwrap();
1398 test_clock
1399 .set_timer_ns("timer2", 2000, Some(start_time), None, None, None, None)
1400 .unwrap();
1401 let events = test_clock.advance_time(UnixNanos::from(*start_time + 2000), true);
1402 assert_eq!(events.len(), 3);
1403 assert_eq!(events[0].name.as_str(), "timer1");
1404 assert_eq!(events[1].name.as_str(), "timer1");
1405 assert_eq!(events[2].name.as_str(), "timer2");
1406 }
1407
1408 #[rstest]
1409 fn test_allow_past_parameter_true(mut test_clock: TestClock) {
1410 test_clock.set_time(UnixNanos::from(2000));
1411 let current_time = test_clock.timestamp_ns();
1412 let past_time = UnixNanos::from(current_time.as_u64() - 1000);
1413
1414 test_clock
1416 .set_time_alert_ns("past_timer", past_time, None, Some(true))
1417 .unwrap();
1418
1419 assert_eq!(test_clock.timer_count(), 1);
1421 assert_eq!(test_clock.timer_names(), vec!["past_timer"]);
1422
1423 let next_time = test_clock.next_time_ns("past_timer").unwrap();
1425 assert!(next_time >= current_time);
1426 }
1427
1428 #[rstest]
1429 fn test_allow_past_parameter_false(mut test_clock: TestClock) {
1430 test_clock.set_time(UnixNanos::from(2000));
1431 let current_time = test_clock.timestamp_ns();
1432 let past_time = current_time - 1000;
1433
1434 let result = test_clock.set_time_alert_ns("past_timer", past_time, None, Some(false));
1436
1437 assert!(result.is_err());
1439 assert!(format!("{}", result.unwrap_err()).contains("was in the past"));
1440
1441 assert_eq!(test_clock.timer_count(), 0);
1443 assert!(test_clock.timer_names().is_empty());
1444 }
1445
1446 #[rstest]
1447 fn test_invalid_stop_time_validation(mut test_clock: TestClock) {
1448 test_clock.set_time(UnixNanos::from(2000));
1449 let current_time = test_clock.timestamp_ns();
1450 let start_time = current_time + 1000;
1451 let stop_time = current_time + 500; let result = test_clock.set_timer_ns(
1455 "invalid_timer",
1456 100,
1457 Some(start_time),
1458 Some(stop_time),
1459 None,
1460 None,
1461 None,
1462 );
1463
1464 assert!(result.is_err());
1466 assert!(format!("{}", result.unwrap_err()).contains("must be after start time"));
1467
1468 assert_eq!(test_clock.timer_count(), 0);
1470 }
1471
1472 #[rstest]
1473 fn test_set_timer_ns_fire_immediately_true(mut test_clock: TestClock) {
1474 let start_time = test_clock.timestamp_ns();
1475 let interval_ns = 1000;
1476
1477 test_clock
1478 .set_timer_ns(
1479 "fire_immediately_timer",
1480 interval_ns,
1481 Some(start_time),
1482 None,
1483 None,
1484 None,
1485 Some(true),
1486 )
1487 .unwrap();
1488
1489 let events = test_clock.advance_time(start_time + 2500, true);
1491
1492 assert_eq!(events.len(), 3);
1494 assert_eq!(*events[0].ts_event, *start_time); assert_eq!(*events[1].ts_event, *start_time + 1000); assert_eq!(*events[2].ts_event, *start_time + 2000); }
1498
1499 #[rstest]
1500 fn test_set_timer_ns_fire_immediately_false(mut test_clock: TestClock) {
1501 let start_time = test_clock.timestamp_ns();
1502 let interval_ns = 1000;
1503
1504 test_clock
1505 .set_timer_ns(
1506 "normal_timer",
1507 interval_ns,
1508 Some(start_time),
1509 None,
1510 None,
1511 None,
1512 Some(false),
1513 )
1514 .unwrap();
1515
1516 let events = test_clock.advance_time(start_time + 2500, true);
1518
1519 assert_eq!(events.len(), 2);
1521 assert_eq!(*events[0].ts_event, *start_time + 1000); assert_eq!(*events[1].ts_event, *start_time + 2000); }
1524
1525 #[rstest]
1526 fn test_set_timer_ns_fire_immediately_default_is_false(mut test_clock: TestClock) {
1527 let start_time = test_clock.timestamp_ns();
1528 let interval_ns = 1000;
1529
1530 test_clock
1532 .set_timer_ns(
1533 "default_timer",
1534 interval_ns,
1535 Some(start_time),
1536 None,
1537 None,
1538 None,
1539 None,
1540 )
1541 .unwrap();
1542
1543 let events = test_clock.advance_time(start_time + 1500, true);
1544
1545 assert_eq!(events.len(), 1);
1547 assert_eq!(*events[0].ts_event, *start_time + 1000); }
1549
1550 #[rstest]
1551 fn test_set_timer_ns_fire_immediately_with_zero_start_time(mut test_clock: TestClock) {
1552 test_clock.set_time(5000.into());
1553 let interval_ns = 1000;
1554
1555 test_clock
1556 .set_timer_ns(
1557 "zero_start_timer",
1558 interval_ns,
1559 None,
1560 None,
1561 None,
1562 None,
1563 Some(true),
1564 )
1565 .unwrap();
1566
1567 let events = test_clock.advance_time(UnixNanos::from(7000), true);
1568
1569 assert_eq!(events.len(), 3);
1572 assert_eq!(*events[0].ts_event, 5000); assert_eq!(*events[1].ts_event, 6000);
1574 assert_eq!(*events[2].ts_event, 7000);
1575 }
1576
1577 #[rstest]
1578 fn test_multiple_timers_different_fire_immediately_settings(mut test_clock: TestClock) {
1579 let start_time = test_clock.timestamp_ns();
1580 let interval_ns = 1000;
1581
1582 test_clock
1584 .set_timer_ns(
1585 "immediate_timer",
1586 interval_ns,
1587 Some(start_time),
1588 None,
1589 None,
1590 None,
1591 Some(true),
1592 )
1593 .unwrap();
1594
1595 test_clock
1597 .set_timer_ns(
1598 "normal_timer",
1599 interval_ns,
1600 Some(start_time),
1601 None,
1602 None,
1603 None,
1604 Some(false),
1605 )
1606 .unwrap();
1607
1608 let events = test_clock.advance_time(start_time + 1500, true);
1609
1610 assert_eq!(events.len(), 3);
1612
1613 let mut event_times: Vec<u64> = events.iter().map(|e| e.ts_event.as_u64()).collect();
1615 event_times.sort_unstable();
1616
1617 assert_eq!(event_times[0], start_time.as_u64()); assert_eq!(event_times[1], start_time.as_u64() + 1000); assert_eq!(event_times[2], start_time.as_u64() + 1000); }
1621
1622 #[rstest]
1623 fn test_timer_name_collision_overwrites(mut test_clock: TestClock) {
1624 let start_time = test_clock.timestamp_ns();
1625
1626 test_clock
1628 .set_timer_ns(
1629 "collision_timer",
1630 1000,
1631 Some(start_time),
1632 None,
1633 None,
1634 None,
1635 None,
1636 )
1637 .unwrap();
1638
1639 let result = test_clock.set_timer_ns(
1641 "collision_timer",
1642 2000,
1643 Some(start_time),
1644 None,
1645 None,
1646 None,
1647 None,
1648 );
1649
1650 assert!(result.is_ok());
1651 assert_eq!(test_clock.timer_count(), 1);
1653
1654 let next_time = test_clock.next_time_ns("collision_timer").unwrap();
1656 assert_eq!(next_time, start_time + 2000);
1658 }
1659
1660 #[rstest]
1661 fn test_timer_zero_interval_error(mut test_clock: TestClock) {
1662 let start_time = test_clock.timestamp_ns();
1663
1664 let result =
1666 test_clock.set_timer_ns("zero_interval", 0, Some(start_time), None, None, None, None);
1667
1668 assert!(result.is_err());
1669 assert_eq!(test_clock.timer_count(), 0);
1670 }
1671
1672 #[rstest]
1673 fn test_timer_empty_name_error(mut test_clock: TestClock) {
1674 let start_time = test_clock.timestamp_ns();
1675
1676 let result = test_clock.set_timer_ns("", 1000, Some(start_time), None, None, None, None);
1678
1679 assert!(result.is_err());
1680 assert_eq!(test_clock.timer_count(), 0);
1681 }
1682
1683 #[rstest]
1684 fn test_timer_exists(mut test_clock: TestClock) {
1685 let name = Ustr::from("exists_timer");
1686 assert!(!test_clock.timer_exists(&name));
1687
1688 test_clock
1689 .set_time_alert_ns(
1690 name.as_str(),
1691 (*test_clock.timestamp_ns() + 1_000).into(),
1692 None,
1693 None,
1694 )
1695 .unwrap();
1696
1697 assert!(test_clock.timer_exists(&name));
1698 }
1699
1700 #[rstest]
1701 fn test_timer_rejects_past_stop_time_when_not_allowed(mut test_clock: TestClock) {
1702 test_clock.set_time(UnixNanos::from(10_000));
1703 let current = test_clock.timestamp_ns();
1704
1705 let result = test_clock.set_timer_ns(
1706 "past_stop",
1707 10_000,
1708 Some(current - 500),
1709 Some(current - 100),
1710 None,
1711 Some(false),
1712 None,
1713 );
1714
1715 let err = result.expect_err("expected stop time validation error");
1716 let err_msg = err.to_string();
1717 assert!(err_msg.contains("stop time"));
1718 assert!(err_msg.contains("in the past"));
1719 }
1720
1721 #[rstest]
1722 fn test_timer_accepts_future_stop_time(mut test_clock: TestClock) {
1723 let current = test_clock.timestamp_ns();
1724
1725 let result = test_clock.set_timer_ns(
1726 "future_stop",
1727 1_000,
1728 Some(current),
1729 Some(current + 10_000),
1730 None,
1731 Some(false),
1732 None,
1733 );
1734
1735 assert!(result.is_ok());
1736 }
1737
1738 #[rstest]
1739 fn test_timer_fire_immediately_at_exact_stop_time(mut test_clock: TestClock) {
1740 let start_time = test_clock.timestamp_ns();
1741 let interval_ns = 1000;
1742 let stop_time = start_time + interval_ns; test_clock
1745 .set_timer_ns(
1746 "exact_stop",
1747 interval_ns,
1748 Some(start_time),
1749 Some(stop_time),
1750 None,
1751 None,
1752 Some(true),
1753 )
1754 .unwrap();
1755
1756 let events = test_clock.advance_time(stop_time, true);
1757
1758 assert_eq!(events.len(), 2);
1760 assert_eq!(*events[0].ts_event, *start_time); assert_eq!(*events[1].ts_event, *stop_time); }
1763
1764 #[rstest]
1765 fn test_timer_advance_to_exact_next_time(mut test_clock: TestClock) {
1766 let start_time = test_clock.timestamp_ns();
1767 let interval_ns = 1000;
1768
1769 test_clock
1770 .set_timer_ns(
1771 "exact_advance",
1772 interval_ns,
1773 Some(start_time),
1774 None,
1775 None,
1776 None,
1777 Some(false),
1778 )
1779 .unwrap();
1780
1781 let next_time = test_clock.next_time_ns("exact_advance").unwrap();
1783 let events = test_clock.advance_time(next_time, true);
1784
1785 assert_eq!(events.len(), 1);
1786 assert_eq!(*events[0].ts_event, *next_time);
1787 }
1788
1789 #[rstest]
1790 fn test_allow_past_bar_aggregation_use_case(mut test_clock: TestClock) {
1791 test_clock.set_time(UnixNanos::from(100_500)); let bar_start_time = UnixNanos::from(100_000); let interval_ns = 1000; let result = test_clock.set_timer_ns(
1801 "bar_timer",
1802 interval_ns,
1803 Some(bar_start_time),
1804 None,
1805 None,
1806 Some(false), Some(false), );
1809
1810 assert!(result.is_ok());
1812 assert_eq!(test_clock.timer_count(), 1);
1813
1814 let next_time = test_clock.next_time_ns("bar_timer").unwrap();
1816 assert_eq!(*next_time, 101_000);
1817 }
1818
1819 #[rstest]
1820 fn test_allow_past_false_rejects_when_next_event_in_past(mut test_clock: TestClock) {
1821 test_clock.set_time(UnixNanos::from(102_000)); let past_start_time = UnixNanos::from(100_000); let interval_ns = 1000; let result = test_clock.set_timer_ns(
1830 "past_event_timer",
1831 interval_ns,
1832 Some(past_start_time),
1833 None,
1834 None,
1835 Some(false), Some(false), );
1838
1839 assert!(result.is_err());
1841 assert!(
1842 result
1843 .unwrap_err()
1844 .to_string()
1845 .contains("would be in the past")
1846 );
1847 }
1848
1849 #[rstest]
1850 fn test_allow_past_false_with_fire_immediately_true(mut test_clock: TestClock) {
1851 test_clock.set_time(UnixNanos::from(100_500)); let past_start_time = UnixNanos::from(100_000); let interval_ns = 1000;
1855
1856 let result = test_clock.set_timer_ns(
1859 "immediate_past_timer",
1860 interval_ns,
1861 Some(past_start_time),
1862 None,
1863 None,
1864 Some(false), Some(true), );
1867
1868 assert!(result.is_err());
1870 assert!(
1871 result
1872 .unwrap_err()
1873 .to_string()
1874 .contains("would be in the past")
1875 );
1876 }
1877
1878 #[rstest]
1879 fn test_cancel_timer_during_execution(mut test_clock: TestClock) {
1880 let start_time = test_clock.timestamp_ns();
1881
1882 test_clock
1883 .set_timer_ns(
1884 "cancel_test",
1885 1000,
1886 Some(start_time),
1887 None,
1888 None,
1889 None,
1890 None,
1891 )
1892 .unwrap();
1893
1894 assert_eq!(test_clock.timer_count(), 1);
1895
1896 test_clock.cancel_timer("cancel_test");
1898
1899 assert_eq!(test_clock.timer_count(), 0);
1900
1901 let events = test_clock.advance_time(start_time + 2000, true);
1903 assert_eq!(events.len(), 0);
1904 }
1905
1906 #[rstest]
1907 fn test_cancelled_timer_queue_entry_is_skipped(mut test_clock: TestClock) {
1908 let start_time = test_clock.timestamp_ns();
1909 test_clock
1910 .set_time_alert_ns("cancelled", start_time + 1000, None, None)
1911 .unwrap();
1912 test_clock
1913 .set_time_alert_ns("active", start_time + 2000, None, None)
1914 .unwrap();
1915
1916 test_clock.cancel_timer("cancelled");
1917 assert_eq!(test_clock.timer_count(), 1);
1918 assert_eq!(test_clock.timer_queue.len(), 2);
1919
1920 let events = test_clock.advance_time(start_time + 1000, true);
1921 assert!(events.is_empty());
1922 assert_eq!(test_clock.timer_names(), vec!["active"]);
1923
1924 let events = test_clock.advance_time(start_time + 2000, true);
1925 assert_eq!(events.len(), 1);
1926 assert_eq!(events[0].name.as_str(), "active");
1927 }
1928
1929 #[rstest]
1930 fn test_timer_queue_compacts_stale_entries(mut test_clock: TestClock) {
1931 let start_time = test_clock.timestamp_ns();
1932 test_clock
1933 .set_time_alert_ns("active", start_time + 1000, None, None)
1934 .unwrap();
1935 test_clock
1936 .set_time_alert_ns("cancelled-1", start_time + 2000, None, None)
1937 .unwrap();
1938 test_clock
1939 .set_time_alert_ns("cancelled-2", start_time + 3000, None, None)
1940 .unwrap();
1941
1942 test_clock.cancel_timer("cancelled-1");
1943 assert_eq!(test_clock.timer_queue.len(), 3);
1944
1945 test_clock.cancel_timer("cancelled-2");
1946 assert_eq!(test_clock.timer_count(), 1);
1947 assert_eq!(test_clock.timer_queue.len(), 1);
1948 }
1949
1950 #[rstest]
1951 fn test_cancel_all_timers(mut test_clock: TestClock) {
1952 test_clock
1954 .set_timer_ns("timer1", 1000, None, None, None, None, None)
1955 .unwrap();
1956 test_clock
1957 .set_timer_ns("timer2", 1500, None, None, None, None, None)
1958 .unwrap();
1959 test_clock
1960 .set_timer_ns("timer3", 2000, None, None, None, None, None)
1961 .unwrap();
1962
1963 assert_eq!(test_clock.timer_count(), 3);
1964
1965 test_clock.cancel_timers();
1967
1968 assert_eq!(test_clock.timer_count(), 0);
1969
1970 let events = test_clock.advance_time(UnixNanos::from(5000), true);
1972 assert_eq!(events.len(), 0);
1973 }
1974
1975 #[rstest]
1976 fn test_clock_reset_clears_timers(mut test_clock: TestClock) {
1977 test_clock
1978 .set_timer_ns("reset_test", 1000, None, None, None, None, None)
1979 .unwrap();
1980
1981 assert_eq!(test_clock.timer_count(), 1);
1982
1983 test_clock.reset();
1985
1986 assert_eq!(test_clock.timer_count(), 0);
1987 assert_eq!(test_clock.timestamp_ns(), UnixNanos::default()); }
1989
1990 #[rstest]
1991 fn test_cancel_default_handler_clears_default(mut test_clock: TestClock) {
1992 test_clock.cancel_default_handler();
1994
1995 let alert_time: UnixNanos = (*test_clock.timestamp_ns() + 1000).into();
1997 let err = test_clock
1998 .set_time_alert_ns("alert", alert_time, None, None)
1999 .unwrap_err();
2000 assert!(
2001 err.to_string().contains("No callbacks provided"),
2002 "unexpected error: {err}"
2003 );
2004 }
2005
2006 #[rstest]
2007 fn test_cancel_default_handler_is_idempotent_on_empty_registry() {
2008 let mut clock = TestClock::new();
2010 clock.cancel_default_handler();
2011 clock.cancel_default_handler();
2012 }
2013
2014 #[rstest]
2015 fn test_cancel_callbacks_clears_named(mut test_clock: TestClock) {
2016 let alert_time: UnixNanos = (*test_clock.timestamp_ns() + 1000).into();
2017 let callback = TimeEventCallback::from(TestCallback::default());
2018 test_clock
2019 .set_time_alert_ns("named_alert", alert_time, Some(callback), None)
2020 .unwrap();
2021 test_clock.cancel_timer("named_alert");
2022
2023 test_clock.cancel_default_handler();
2025 test_clock.cancel_callbacks();
2026
2027 let err = test_clock
2028 .set_time_alert_ns("named_alert", alert_time, None, None)
2029 .unwrap_err();
2030 assert!(
2031 err.to_string().contains("No callbacks provided"),
2032 "unexpected error: {err}"
2033 );
2034 }
2035
2036 #[rstest]
2037 fn test_cancel_default_handler_preserves_named_callbacks(mut test_clock: TestClock) {
2038 let alert_time: UnixNanos = (*test_clock.timestamp_ns() + 1000).into();
2039 let callback = TimeEventCallback::from(TestCallback::default());
2040 test_clock
2041 .set_time_alert_ns("alert", alert_time, Some(callback), None)
2042 .unwrap();
2043 test_clock.cancel_timer("alert");
2044
2045 test_clock.cancel_default_handler();
2046
2047 test_clock
2049 .set_time_alert_ns("alert", alert_time, None, None)
2050 .unwrap();
2051 }
2052
2053 #[rstest]
2054 fn test_cancel_callbacks_preserves_default_handler(mut test_clock: TestClock) {
2055 test_clock.cancel_callbacks();
2057
2058 let alert_time: UnixNanos = (*test_clock.timestamp_ns() + 1000).into();
2059 test_clock
2060 .set_time_alert_ns("alert", alert_time, None, None)
2061 .unwrap();
2062 }
2063
2064 #[rstest]
2065 fn test_set_time_alert_default_impl(mut test_clock: TestClock) {
2066 let current_time = test_clock.utc_now();
2067 let alert_time = current_time + chrono::Duration::seconds(1);
2068
2069 test_clock
2071 .set_time_alert("alert_test", alert_time, None, None)
2072 .unwrap();
2073
2074 assert_eq!(test_clock.timer_count(), 1);
2075 assert_eq!(test_clock.timer_names(), vec!["alert_test"]);
2076
2077 let expected_ns = UnixNanos::from(alert_time);
2079 let next_time = test_clock.next_time_ns("alert_test").unwrap();
2080
2081 let diff = if next_time >= expected_ns {
2083 next_time.as_u64() - expected_ns.as_u64()
2084 } else {
2085 expected_ns.as_u64() - next_time.as_u64()
2086 };
2087 assert!(
2088 diff < 1000,
2089 "Timer should be set within 1 microsecond of expected time"
2090 );
2091 }
2092
2093 #[rstest]
2094 fn test_set_timer_default_impl(mut test_clock: TestClock) {
2095 let current_time = test_clock.utc_now();
2096 let start_time = current_time + chrono::Duration::seconds(1);
2097 let interval = Duration::from_millis(500);
2098
2099 test_clock
2101 .set_timer(
2102 "timer_test",
2103 interval,
2104 Some(start_time),
2105 None,
2106 None,
2107 None,
2108 None,
2109 )
2110 .unwrap();
2111
2112 assert_eq!(test_clock.timer_count(), 1);
2113 assert_eq!(test_clock.timer_names(), vec!["timer_test"]);
2114
2115 let start_ns = UnixNanos::from(start_time);
2117 let interval_ns = interval.as_nanos() as u64;
2118
2119 let events = test_clock.advance_time(start_ns + interval_ns * 3, true);
2120 assert_eq!(events.len(), 3); assert_eq!(*events[0].ts_event, *start_ns + interval_ns);
2124 assert_eq!(*events[1].ts_event, *start_ns + interval_ns * 2);
2125 assert_eq!(*events[2].ts_event, *start_ns + interval_ns * 3);
2126 }
2127
2128 #[rstest]
2129 fn test_set_timer_with_stop_time_default_impl(mut test_clock: TestClock) {
2130 let current_time = test_clock.utc_now();
2131 let start_time = current_time + chrono::Duration::seconds(1);
2132 let stop_time = current_time + chrono::Duration::seconds(3);
2133 let interval = Duration::from_secs(1);
2134
2135 test_clock
2137 .set_timer(
2138 "timer_with_stop",
2139 interval,
2140 Some(start_time),
2141 Some(stop_time),
2142 None,
2143 None,
2144 None,
2145 )
2146 .unwrap();
2147
2148 assert_eq!(test_clock.timer_count(), 1);
2149
2150 let stop_ns = UnixNanos::from(stop_time);
2152 let events = test_clock.advance_time(stop_ns + 1000, true);
2153
2154 assert_eq!(events.len(), 2);
2156
2157 let start_ns = UnixNanos::from(start_time);
2158 let interval_ns = interval.as_nanos() as u64;
2159 assert_eq!(*events[0].ts_event, *start_ns + interval_ns);
2160 assert_eq!(*events[1].ts_event, *start_ns + interval_ns * 2);
2161 }
2162
2163 #[rstest]
2164 fn test_set_timer_fire_immediately_default_impl(mut test_clock: TestClock) {
2165 let current_time = test_clock.utc_now();
2166 let start_time = current_time + chrono::Duration::seconds(1);
2167 let interval = Duration::from_millis(500);
2168
2169 test_clock
2171 .set_timer(
2172 "immediate_timer",
2173 interval,
2174 Some(start_time),
2175 None,
2176 None,
2177 None,
2178 Some(true),
2179 )
2180 .unwrap();
2181
2182 let start_ns = UnixNanos::from(start_time);
2183 let interval_ns = interval.as_nanos() as u64;
2184
2185 let events = test_clock.advance_time(start_ns + interval_ns, true);
2187
2188 assert_eq!(events.len(), 2);
2190 assert_eq!(*events[0].ts_event, *start_ns); assert_eq!(*events[1].ts_event, *start_ns + interval_ns); }
2193
2194 #[rstest]
2195 fn test_set_time_alert_when_alert_time_equals_current_time(mut test_clock: TestClock) {
2196 let current_time = test_clock.timestamp_ns();
2197
2198 test_clock
2200 .set_time_alert_ns("alert_at_current_time", current_time, None, None)
2201 .unwrap();
2202
2203 assert_eq!(test_clock.timer_count(), 1);
2204
2205 let events = test_clock.advance_time(current_time, true);
2207
2208 assert_eq!(events.len(), 1);
2210 assert_eq!(events[0].name.as_str(), "alert_at_current_time");
2211 assert_eq!(*events[0].ts_event, *current_time);
2212 }
2213
2214 #[rstest]
2215 fn test_cancel_and_reschedule_same_name(mut test_clock: TestClock) {
2216 let start = test_clock.timestamp_ns();
2217
2218 test_clock
2219 .set_time_alert_ns("timer", UnixNanos::from(*start + 1000), None, None)
2220 .unwrap();
2221 assert_eq!(test_clock.timer_count(), 1);
2222
2223 test_clock.cancel_timer("timer");
2224 assert_eq!(test_clock.timer_count(), 0);
2225
2226 test_clock
2227 .set_time_alert_ns("timer", UnixNanos::from(*start + 2000), None, None)
2228 .unwrap();
2229 assert_eq!(test_clock.timer_count(), 1);
2230
2231 let events = test_clock.advance_time(UnixNanos::from(*start + 1500), true);
2232 assert!(events.is_empty());
2233
2234 let events = test_clock.advance_time(UnixNanos::from(*start + 2000), true);
2235 assert_eq!(events.len(), 1);
2236 assert_eq!(*events[0].ts_event, *start + 2000);
2237 }
2238
2239 #[rstest]
2240 fn test_multiple_timers_same_timestamp_all_fire(mut test_clock: TestClock) {
2241 let fire_time = UnixNanos::from(*test_clock.timestamp_ns() + 1000);
2242
2243 for i in 0..5 {
2244 test_clock
2245 .set_time_alert_ns(&format!("timer_{i}"), fire_time, None, None)
2246 .unwrap();
2247 }
2248 assert_eq!(test_clock.timer_count(), 5);
2249
2250 let events = test_clock.advance_time(fire_time, true);
2251 assert_eq!(events.len(), 5);
2252 for event in &events {
2253 assert_eq!(*event.ts_event, *fire_time);
2254 }
2255 }
2256
2257 #[rstest]
2258 fn test_events_ordered_by_timestamp_after_advance() {
2259 let mut clock = TestClock::new();
2260 clock.register_default_handler(TestCallback::default().into());
2261 let start = clock.timestamp_ns();
2262
2263 clock
2264 .set_time_alert_ns("third", UnixNanos::from(*start + 300), None, None)
2265 .unwrap();
2266 clock
2267 .set_time_alert_ns("first", UnixNanos::from(*start + 100), None, None)
2268 .unwrap();
2269 clock
2270 .set_time_alert_ns("second", UnixNanos::from(*start + 200), None, None)
2271 .unwrap();
2272
2273 let events = clock.advance_time(UnixNanos::from(*start + 400), true);
2274 assert_eq!(events.len(), 3);
2275 assert_eq!(events[0].name.as_str(), "first");
2276 assert_eq!(events[1].name.as_str(), "second");
2277 assert_eq!(events[2].name.as_str(), "third");
2278 }
2279
2280 #[rstest]
2281 fn test_large_interval_does_not_overflow(mut test_clock: TestClock) {
2282 let start = test_clock.timestamp_ns();
2283 let large_interval: u64 = 1_000_000_000 * 60 * 60 * 24 * 365; test_clock
2286 .set_timer_ns(
2287 "large_interval",
2288 large_interval,
2289 Some(start),
2290 None,
2291 None,
2292 None,
2293 None,
2294 )
2295 .unwrap();
2296
2297 let events = test_clock.advance_time(UnixNanos::from(*start + large_interval), true);
2298 assert_eq!(events.len(), 1);
2299 assert_eq!(*events[0].ts_event, *start + large_interval);
2300 }
2301
2302 #[rstest]
2303 fn test_near_zero_interval_fires_correctly(mut test_clock: TestClock) {
2304 let start = test_clock.timestamp_ns();
2305
2306 test_clock
2307 .set_timer_ns("tiny", 1, Some(start), None, None, None, None)
2308 .unwrap();
2309
2310 let events = test_clock.advance_time(UnixNanos::from(*start + 10), true);
2311 assert_eq!(events.len(), 10);
2312
2313 for i in 1..events.len() {
2314 assert!(events[i].ts_event >= events[i - 1].ts_event);
2315 }
2316 }
2317
2318 #[rstest]
2319 fn test_repeated_advance_to_same_time_no_double_fire(mut test_clock: TestClock) {
2320 let fire_time = UnixNanos::from(*test_clock.timestamp_ns() + 1000);
2321
2322 test_clock
2323 .set_time_alert_ns("once", fire_time, None, None)
2324 .unwrap();
2325
2326 let events1 = test_clock.advance_time(fire_time, true);
2327 assert_eq!(events1.len(), 1);
2328
2329 let events2 = test_clock.advance_time(fire_time, true);
2330 assert!(events2.is_empty());
2331 }
2332
2333 #[rstest]
2334 fn test_advance_with_no_timers(mut test_clock: TestClock) {
2335 let start = test_clock.timestamp_ns();
2336
2337 let events = test_clock.advance_time(UnixNanos::from(*start + 1000), true);
2338 assert!(events.is_empty());
2339 assert_eq!(*test_clock.timestamp_ns(), *start + 1000);
2340 }
2341
2342 #[rstest]
2343 fn test_clock_api_new_uses_native_backing(test_clock: TestClock) {
2344 let clock = RefCell::new(test_clock);
2345 let api = ClockApi::new(&clock);
2346
2347 api.set_timer_ns(
2348 "native-timer",
2349 1_000,
2350 None,
2351 None,
2352 None,
2353 Some(true),
2354 Some(false),
2355 )
2356 .unwrap();
2357
2358 assert_eq!(api.timer_count(), 1);
2359 assert_eq!(api.timer_names(), vec!["native-timer".to_string()]);
2360 assert_eq!(
2361 api.next_time_ns("native-timer"),
2362 Some(UnixNanos::from(1_000))
2363 );
2364 }
2365
2366 #[rstest]
2367 fn test_clock_api_handlers_back_full_surface() {
2368 let alerts = Arc::new(Mutex::new(Vec::new()));
2369 let timers = Arc::new(Mutex::new(Vec::new()));
2370 let cancellations = Arc::new(Mutex::new(Vec::new()));
2371 let cancel_all = Arc::new(Mutex::new(false));
2372
2373 let alerts_for_handler = Arc::clone(&alerts);
2374 let timers_for_handler = Arc::clone(&timers);
2375 let cancellations_for_handler = Arc::clone(&cancellations);
2376 let cancel_all_for_handler = Arc::clone(&cancel_all);
2377
2378 let clock = ClockApi::from_handlers(
2379 || UnixNanos::from(1_700_000_000_123_456_789),
2380 move |name, alert_time_ns, _callback, allow_past| {
2381 alerts_for_handler.lock().expect(MUTEX_POISONED).push((
2382 name.to_string(),
2383 alert_time_ns,
2384 allow_past,
2385 ));
2386 Ok(())
2387 },
2388 move |name,
2389 interval_ns,
2390 start_time_ns,
2391 stop_time_ns,
2392 _callback,
2393 allow_past,
2394 fire_immediately| {
2395 timers_for_handler.lock().expect(MUTEX_POISONED).push((
2396 name.to_string(),
2397 interval_ns,
2398 start_time_ns,
2399 stop_time_ns,
2400 allow_past,
2401 fire_immediately,
2402 ));
2403 Ok(())
2404 },
2405 || vec!["alpha".to_string(), "beta".to_string()],
2406 || 2,
2407 |name| name == "alpha",
2408 |name| (name == "alpha").then(|| UnixNanos::from(1_700_000_000_999_000_000)),
2409 move |name| {
2410 cancellations_for_handler
2411 .lock()
2412 .expect(MUTEX_POISONED)
2413 .push(name.to_string());
2414 },
2415 move || {
2416 *cancel_all_for_handler.lock().expect(MUTEX_POISONED) = true;
2417 },
2418 );
2419
2420 let alert_time = DateTime::from_timestamp_nanos(1_700_000_000_333_000_000);
2421 let start_time = DateTime::from_timestamp_nanos(1_700_000_000_444_000_000);
2422 let stop_time = DateTime::from_timestamp_nanos(1_700_000_001_444_000_000);
2423 clock
2424 .set_time_alert("alert", alert_time, None, Some(false))
2425 .unwrap();
2426 clock
2427 .set_time_alert_ns(
2428 "alert-ns",
2429 UnixNanos::from(1_700_000_000_555_000_000),
2430 None,
2431 Some(true),
2432 )
2433 .unwrap();
2434 clock
2435 .set_timer(
2436 "timer",
2437 Duration::from_millis(250),
2438 Some(start_time),
2439 Some(stop_time),
2440 None,
2441 Some(true),
2442 Some(false),
2443 )
2444 .unwrap();
2445 clock
2446 .set_timer_ns(
2447 "timer-ns",
2448 500_000_000,
2449 Some(UnixNanos::from(1_700_000_000_666_000_000)),
2450 Some(UnixNanos::from(1_700_000_001_666_000_000)),
2451 None,
2452 Some(false),
2453 Some(true),
2454 )
2455 .unwrap();
2456 clock.cancel_timer("alpha");
2457 clock.cancel_timers();
2458
2459 assert_eq!(
2460 clock.timestamp_ns(),
2461 UnixNanos::from(1_700_000_000_123_456_789)
2462 );
2463 assert_eq!(clock.timestamp_us(), 1_700_000_000_123_456);
2464 assert_eq!(clock.timestamp_ms(), 1_700_000_000_123);
2465 assert_eq!(clock.timestamp(), 1_700_000_000.123_456_7);
2466 assert_eq!(
2467 clock.utc_now(),
2468 DateTime::from_timestamp_nanos(1_700_000_000_123_456_789)
2469 );
2470 assert_eq!(clock.timer_names(), vec!["alpha", "beta"]);
2471 assert_eq!(clock.timer_count(), 2);
2472 assert!(clock.timer_exists("alpha"));
2473 assert!(!clock.timer_exists("gamma"));
2474 assert_eq!(
2475 clock.next_time_ns("alpha"),
2476 Some(UnixNanos::from(1_700_000_000_999_000_000))
2477 );
2478 assert_eq!(
2479 alerts.lock().expect(MUTEX_POISONED).as_slice(),
2480 &[
2481 (
2482 "alert".to_string(),
2483 UnixNanos::from(1_700_000_000_333_000_000),
2484 Some(false)
2485 ),
2486 (
2487 "alert-ns".to_string(),
2488 UnixNanos::from(1_700_000_000_555_000_000),
2489 Some(true)
2490 )
2491 ]
2492 );
2493 assert_eq!(
2494 timers.lock().expect(MUTEX_POISONED).as_slice(),
2495 &[
2496 (
2497 "timer".to_string(),
2498 250_000_000,
2499 Some(UnixNanos::from(1_700_000_000_444_000_000)),
2500 Some(UnixNanos::from(1_700_000_001_444_000_000)),
2501 Some(true),
2502 Some(false)
2503 ),
2504 (
2505 "timer-ns".to_string(),
2506 500_000_000,
2507 Some(UnixNanos::from(1_700_000_000_666_000_000)),
2508 Some(UnixNanos::from(1_700_000_001_666_000_000)),
2509 Some(false),
2510 Some(true)
2511 )
2512 ]
2513 );
2514 assert_eq!(
2515 cancellations.lock().expect(MUTEX_POISONED).as_slice(),
2516 &["alpha".to_string()]
2517 );
2518 assert!(*cancel_all.lock().expect(MUTEX_POISONED));
2519 }
2520}