nautilus_common/live/
clock.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Live clock implementation using Tokio for real-time operations.
17
18use std::{collections::BTreeMap, ops::Deref, sync::Arc};
19
20use nautilus_core::{
21    AtomicTime, UnixNanos, correctness::check_predicate_true, time::get_atomic_clock_realtime,
22};
23use ustr::Ustr;
24
25use super::timer::LiveTimer;
26use crate::{
27    clock::{
28        CallbackRegistry, Clock, replace_existing_timer, validate_and_prepare_time_alert,
29        validate_and_prepare_timer,
30    },
31    runner::{TimeEventSender, try_get_time_event_sender},
32    timer::{TimeEvent, TimeEventCallback, TimeEventHandler, create_valid_interval},
33};
34
35/// A real-time clock which uses system time.
36///
37/// Timestamps are guaranteed to be unique and monotonically increasing.
38///
39/// # Threading
40///
41/// The clock holds thread-local runtime state and must remain on its originating thread.
42#[derive(Debug)]
43pub struct LiveClock {
44    time: &'static AtomicTime,
45    timers: BTreeMap<Ustr, LiveTimer>,
46    callbacks: CallbackRegistry,
47    sender: Option<Arc<dyn TimeEventSender>>,
48}
49
50impl LiveClock {
51    /// Creates a new [`LiveClock`] instance.
52    #[must_use]
53    pub fn new(sender: Option<Arc<dyn TimeEventSender>>) -> Self {
54        Self {
55            time: get_atomic_clock_realtime(),
56            timers: BTreeMap::new(),
57            callbacks: CallbackRegistry::new(),
58            sender,
59        }
60    }
61
62    #[must_use]
63    pub const fn get_timers(&self) -> &BTreeMap<Ustr, LiveTimer> {
64        &self.timers
65    }
66
67    fn clear_expired_timers(&mut self) {
68        self.timers.retain(|_, timer| !timer.is_expired());
69    }
70
71    fn replace_existing_timer_if_needed(&mut self, name: &Ustr) {
72        replace_existing_timer(&mut self.timers, name);
73    }
74}
75
76impl Default for LiveClock {
77    /// Creates a new default [`LiveClock`] instance.
78    ///
79    /// Uses `try_get_time_event_sender()` to allow creation before channels are initialized.
80    fn default() -> Self {
81        Self::new(try_get_time_event_sender())
82    }
83}
84
85impl Deref for LiveClock {
86    type Target = AtomicTime;
87
88    fn deref(&self) -> &Self::Target {
89        self.time
90    }
91}
92
93impl Clock for LiveClock {
94    fn timestamp_ns(&self) -> UnixNanos {
95        self.time.get_time_ns()
96    }
97
98    fn timestamp_us(&self) -> u64 {
99        self.time.get_time_us()
100    }
101
102    fn timestamp_ms(&self) -> u64 {
103        self.time.get_time_ms()
104    }
105
106    fn timestamp(&self) -> f64 {
107        self.time.get_time()
108    }
109
110    fn timer_names(&self) -> Vec<&str> {
111        self.timers
112            .iter()
113            .filter(|(_, timer)| !timer.is_expired())
114            .map(|(k, _)| k.as_str())
115            .collect()
116    }
117
118    fn timer_count(&self) -> usize {
119        self.timers
120            .iter()
121            .filter(|(_, timer)| !timer.is_expired())
122            .count()
123    }
124
125    fn timer_exists(&self, name: &Ustr) -> bool {
126        self.timers.contains_key(name)
127    }
128
129    fn register_default_handler(&mut self, handler: TimeEventCallback) {
130        self.callbacks.register_default_handler(handler);
131    }
132
133    fn cancel_default_handler(&mut self) {
134        self.callbacks.cancel_default_handler();
135    }
136
137    fn cancel_callbacks(&mut self) {
138        self.callbacks.clear();
139    }
140
141    /// # Panics
142    ///
143    /// This function panics if:
144    /// - The event does not have an associated handler (see trait documentation).
145    #[allow(unused_variables)]
146    fn get_handler(&self, event: TimeEvent) -> TimeEventHandler {
147        self.callbacks.get_handler(event)
148    }
149
150    fn set_time_alert_ns(
151        &mut self,
152        name: &str,
153        alert_time_ns: UnixNanos,
154        callback: Option<TimeEventCallback>,
155        allow_past: Option<bool>,
156    ) -> anyhow::Result<()> {
157        let ts_now = self.get_time_ns();
158        let (name, alert_time_ns) =
159            validate_and_prepare_time_alert(name, alert_time_ns, allow_past, ts_now)?;
160
161        self.replace_existing_timer_if_needed(&name);
162
163        check_predicate_true(
164            callback.is_some() | self.callbacks.has_any_callback(&name),
165            "No callbacks provided",
166        )?;
167
168        let callback = if let Some(callback) = callback {
169            self.callbacks.register_callback(name, callback.clone());
170            callback
171        } else {
172            self.callbacks
173                .get_callback(&name)
174                .expect("Callback should exist")
175        };
176
177        // Safe to calculate interval now that we've ensured alert_time_ns >= ts_now
178        let interval_ns = create_valid_interval((alert_time_ns - ts_now).into());
179
180        let mut timer = LiveTimer::new(
181            name,
182            interval_ns,
183            ts_now,
184            Some(alert_time_ns),
185            callback,
186            false,
187            self.sender.clone(),
188        );
189
190        timer.start();
191
192        self.clear_expired_timers();
193        self.timers.insert(name, timer);
194
195        Ok(())
196    }
197
198    fn set_timer_ns(
199        &mut self,
200        name: &str,
201        interval_ns: u64,
202        start_time_ns: Option<UnixNanos>,
203        stop_time_ns: Option<UnixNanos>,
204        callback: Option<TimeEventCallback>,
205        allow_past: Option<bool>,
206        fire_immediately: Option<bool>,
207    ) -> anyhow::Result<()> {
208        let ts_now = self.get_time_ns();
209        let (name, start_time_ns, stop_time_ns, _allow_past, fire_immediately) =
210            validate_and_prepare_timer(
211                name,
212                interval_ns,
213                start_time_ns,
214                stop_time_ns,
215                allow_past,
216                fire_immediately,
217                ts_now,
218            )?;
219
220        check_predicate_true(
221            callback.is_some() | self.callbacks.has_any_callback(&name),
222            "No callbacks provided",
223        )?;
224
225        self.replace_existing_timer_if_needed(&name);
226
227        let callback = if let Some(callback) = callback {
228            self.callbacks.register_callback(name, callback.clone());
229            callback
230        } else {
231            self.callbacks
232                .get_callback(&name)
233                .expect("Callback should exist")
234        };
235
236        let interval_ns = create_valid_interval(interval_ns);
237
238        let mut timer = LiveTimer::new(
239            name,
240            interval_ns,
241            start_time_ns,
242            stop_time_ns,
243            callback,
244            fire_immediately,
245            self.sender.clone(),
246        );
247        timer.start();
248
249        self.clear_expired_timers();
250        self.timers.insert(name, timer);
251
252        Ok(())
253    }
254
255    fn next_time_ns(&self, name: &str) -> Option<UnixNanos> {
256        self.timers
257            .get(&Ustr::from(name))
258            .map(LiveTimer::next_time_ns)
259    }
260
261    fn cancel_timer(&mut self, name: &str) {
262        let timer = self.timers.remove(&Ustr::from(name));
263        if let Some(mut timer) = timer {
264            timer.cancel();
265        }
266    }
267
268    fn cancel_timers(&mut self) {
269        for timer in &mut self.timers.values_mut() {
270            timer.cancel();
271        }
272
273        self.timers.clear();
274    }
275
276    fn reset(&mut self) {
277        self.cancel_timers();
278        self.callbacks.clear();
279    }
280}
281
282#[cfg(test)]
283mod tests {
284    use std::{
285        sync::{Arc, Mutex},
286        time::Duration,
287    };
288
289    use nautilus_core::{MUTEX_POISONED, UnixNanos, time::get_atomic_clock_realtime};
290    use rstest::rstest;
291    use ustr::Ustr;
292
293    use super::*;
294    use crate::{
295        clock::Clock,
296        runner::TimeEventSender,
297        testing::wait_until,
298        timer::{TimeEvent, TimeEventCallback, TimeEventHandler},
299    };
300
301    #[derive(Debug)]
302    struct CollectingSender {
303        events: Arc<Mutex<Vec<(TimeEvent, UnixNanos)>>>,
304    }
305
306    impl CollectingSender {
307        fn new(events: Arc<Mutex<Vec<(TimeEvent, UnixNanos)>>>) -> Self {
308            Self { events }
309        }
310    }
311
312    impl TimeEventSender for CollectingSender {
313        fn send(&self, handler: TimeEventHandler) {
314            let TimeEventHandler { event, callback } = handler;
315            let now_ns = get_atomic_clock_realtime().get_time_ns();
316            let event_clone = event.clone();
317            callback.call(event);
318            self.events
319                .lock()
320                .expect(MUTEX_POISONED)
321                .push((event_clone, now_ns));
322        }
323    }
324
325    fn wait_for_events(
326        events: &Arc<Mutex<Vec<(TimeEvent, UnixNanos)>>>,
327        target: usize,
328        timeout: Duration,
329    ) {
330        wait_until(
331            || events.lock().expect(MUTEX_POISONED).len() >= target,
332            timeout,
333        );
334    }
335
336    #[rstest]
337    fn test_live_clock_timer_replacement_cancels_previous_task() {
338        let events = Arc::new(Mutex::new(Vec::new()));
339        let sender = Arc::new(CollectingSender::new(Arc::clone(&events)));
340
341        let mut clock = LiveClock::new(Some(sender));
342        clock.register_default_handler(TimeEventCallback::from(|_| {}));
343
344        let fast_interval = Duration::from_millis(10).as_nanos() as u64;
345        clock
346            .set_timer_ns("replace", fast_interval, None, None, None, None, None)
347            .unwrap();
348
349        wait_for_events(&events, 2, Duration::from_millis(200));
350        events.lock().expect(MUTEX_POISONED).clear();
351
352        let slow_interval = Duration::from_millis(30).as_nanos() as u64;
353        clock
354            .set_timer_ns("replace", slow_interval, None, None, None, None, None)
355            .unwrap();
356
357        wait_for_events(&events, 3, Duration::from_millis(300));
358
359        let snapshot = events.lock().expect(MUTEX_POISONED).clone();
360        let diffs: Vec<u64> = snapshot
361            .array_windows()
362            .map(|[a, b]| b.0.ts_event.as_u64() - a.0.ts_event.as_u64())
363            .collect();
364
365        assert!(!diffs.is_empty());
366        for diff in diffs {
367            assert_ne!(diff, fast_interval);
368        }
369
370        clock.cancel_timers();
371    }
372
373    #[rstest]
374    fn test_live_clock_time_alert_persists_callback() {
375        let events = Arc::new(Mutex::new(Vec::new()));
376        let sender = Arc::new(CollectingSender::new(Arc::clone(&events)));
377
378        let mut clock = LiveClock::new(Some(sender));
379        clock.register_default_handler(TimeEventCallback::from(|_| {}));
380
381        let now = clock.timestamp_ns();
382        let alert_time = now + 1_000_u64;
383
384        clock
385            .set_time_alert_ns("alert-callback", alert_time, None, None)
386            .unwrap();
387
388        assert!(
389            clock
390                .callbacks
391                .has_any_callback(&Ustr::from("alert-callback"))
392        );
393
394        clock.cancel_timers();
395    }
396
397    #[rstest]
398    fn test_live_clock_reset_stops_active_timers() {
399        let events = Arc::new(Mutex::new(Vec::new()));
400        let sender = Arc::new(CollectingSender::new(Arc::clone(&events)));
401
402        let mut clock = LiveClock::new(Some(sender));
403        clock.register_default_handler(TimeEventCallback::from(|_| {}));
404
405        clock
406            .set_timer_ns(
407                "reset-test",
408                Duration::from_millis(15).as_nanos() as u64,
409                None,
410                None,
411                None,
412                None,
413                None,
414            )
415            .unwrap();
416
417        wait_for_events(&events, 2, Duration::from_millis(250));
418
419        clock.reset();
420
421        // Wait for any in-flight events to arrive
422        let start = std::time::Instant::now();
423        wait_until(
424            || start.elapsed() >= Duration::from_millis(50),
425            Duration::from_secs(2),
426        );
427
428        // Clear any events that arrived before reset took effect
429        events.lock().expect(MUTEX_POISONED).clear();
430
431        // Verify no new events arrive (timer should be stopped)
432        let start = std::time::Instant::now();
433        wait_until(
434            || start.elapsed() >= Duration::from_millis(50),
435            Duration::from_secs(2),
436        );
437        assert!(events.lock().expect(MUTEX_POISONED).is_empty());
438    }
439
440    #[rstest]
441    fn test_live_timer_short_delay_not_early() {
442        let events = Arc::new(Mutex::new(Vec::new()));
443        let sender = Arc::new(CollectingSender::new(Arc::clone(&events)));
444
445        let mut clock = LiveClock::new(Some(sender));
446        clock.register_default_handler(TimeEventCallback::from(|_| {}));
447
448        let now = clock.timestamp_ns();
449        let start_time = UnixNanos::from(*now + 500_000); // 0.5 ms in the future
450        let interval_ns = 1_000_000;
451
452        clock
453            .set_timer_ns(
454                "short-delay",
455                interval_ns,
456                Some(start_time),
457                None,
458                None,
459                None,
460                Some(true),
461            )
462            .unwrap();
463
464        wait_for_events(&events, 1, Duration::from_millis(100));
465
466        let snapshot = events.lock().expect(MUTEX_POISONED).clone();
467        assert!(!snapshot.is_empty());
468
469        for (event, actual_ts) in &snapshot {
470            assert!(actual_ts.as_u64() >= event.ts_event.as_u64());
471        }
472
473        clock.cancel_timers();
474    }
475}
nautilus_common/live/clock.rs

nautilus_common/live/
clock.rs
