darkbio_clock/
timers.rs

1// clock-rs: virtual clock for testing blocking code
2// Copyright 2026 Dark Bio AG. All rights reserved.
3//
4// Use of this source code is governed by a BSD-style
5// license that can be found in the LICENSE file.
6
7//! Crossbeam timers and receive deadlines that follow a clock.
8
9use crate::Clock;
10use crossbeam_channel::{Receiver, RecvTimeoutError};
11use std::time::{Duration, Instant};
12
13impl Clock {
14    /// Returns a receiver that gets the instant `duration` from now once this
15    /// clock reaches it.
16    ///
17    /// It mirrors `crossbeam_channel::after`, which it is on the real clock. On
18    /// a test clock, it is [`Self::at`] for the instant `duration` from now. A
19    /// duration past the end of `Instant`'s range returns a receiver that never
20    /// fires.
21    ///
22    /// Wait for a job or a timeout, driven from a test through `wait_timers`,
23    /// since `wait_blocked` cannot see a thread blocked in `select!`:
24    ///
25    /// ```
26    /// # #[cfg(feature = "test-clock")] {
27    /// use darkbio_clock::TestClock;
28    /// use darkbio_clock::crossbeam_channel::{select, unbounded};
29    /// use std::thread;
30    /// use std::time::Duration;
31    ///
32    /// let mut tester = TestClock::new();
33    /// let clock = tester.clock();
34    /// let (_jobs, queue) = unbounded::<u32>();
35    ///
36    /// let worker = thread::spawn(move || {
37    ///     select! {
38    ///         recv(queue) -> job => job.ok(),
39    ///         recv(clock.after(Duration::from_secs(5))) -> _ => None,
40    ///     }
41    /// });
42    /// tester.wait_timers(1);
43    /// tester.advance(Duration::from_secs(5));
44    /// assert_eq!(worker.join().unwrap(), None);
45    /// # }
46    /// ```
47    #[cfg_attr(docsrs, doc(cfg(feature = "crossbeam")))]
48    pub fn after(&self, duration: Duration) -> Receiver<Instant> {
49        #[cfg(any(test, feature = "test-clock"))]
50        if let Some(paused) = &self.paused {
51            return match paused.now().checked_add(duration) {
52                Some(deadline) => paused.at(deadline),
53                None => crossbeam_channel::never(),
54            };
55        }
56        crossbeam_channel::after(duration)
57    }
58
59    /// Returns a receiver that gets `deadline` once this clock reaches it, at
60    /// once if it already has.
61    ///
62    /// It mirrors `crossbeam_channel::at`, which it is on the real clock. On a
63    /// test clock, an advance that reaches the deadline delivers it, even one
64    /// that overshoots, before any thread can read the new time. The channel
65    /// stays connected until the `TestClock` and every `Clock` handle are gone.
66    ///
67    /// An advance sends its due timers one at a time, in deadline order and
68    /// then in the order they were armed, and a `select!` already waiting
69    /// takes the first one sent. So a waiting `select_biased!` over timers due
70    /// at the same instant takes the one armed first, where the real clock's
71    /// takes the first of their arms.
72    #[cfg_attr(docsrs, doc(cfg(feature = "crossbeam")))]
73    pub fn at(&self, deadline: Instant) -> Receiver<Instant> {
74        #[cfg(any(test, feature = "test-clock"))]
75        if let Some(paused) = &self.paused {
76            return paused.at(deadline);
77        }
78        crossbeam_channel::at(deadline)
79    }
80
81    /// Receives a message, waiting up to `timeout` on this clock.
82    ///
83    /// It mirrors crossbeam's `Receiver::recv_timeout`, which it is on the real
84    /// clock. A buffered message or a disconnection wins over the timeout. On a
85    /// test clock, it is [`Self::recv_deadline`] for the instant `timeout` from
86    /// now. A timeout past the end of `Instant`'s range waits like
87    /// `Receiver::recv`.
88    #[cfg_attr(docsrs, doc(cfg(feature = "crossbeam")))]
89    pub fn recv_timeout<T>(
90        &self,
91        receiver: &Receiver<T>,
92        timeout: Duration,
93    ) -> Result<T, RecvTimeoutError> {
94        #[cfg(any(test, feature = "test-clock"))]
95        if let Some(paused) = &self.paused {
96            return match paused.now().checked_add(timeout) {
97                Some(deadline) => paused.recv_deadline(receiver, deadline),
98                None => receiver.recv().map_err(RecvTimeoutError::from),
99            };
100        }
101        receiver.recv_timeout(timeout)
102    }
103
104    /// Receives a message, waiting until this clock reaches `deadline`.
105    ///
106    /// It mirrors crossbeam's `Receiver::recv_deadline`, which it is on the real
107    /// clock. A buffered message or a disconnection wins over the deadline. On a
108    /// test clock, a clock timer due by the deadline holds its message before
109    /// the receive can time out, so it wins too. A waiting receive arms a timer,
110    /// which `wait_timers` counts until it fires or the receive returns.
111    #[cfg_attr(docsrs, doc(cfg(feature = "crossbeam")))]
112    pub fn recv_deadline<T>(
113        &self,
114        receiver: &Receiver<T>,
115        deadline: Instant,
116    ) -> Result<T, RecvTimeoutError> {
117        #[cfg(any(test, feature = "test-clock"))]
118        if let Some(paused) = &self.paused {
119            return paused.recv_deadline(receiver, deadline);
120        }
121        receiver.recv_deadline(deadline)
122    }
123}
124
125// The tests live in src/tests, loaded from here so that they keep this
126// module's private items in reach
127#[cfg(all(test, not(loom)))]
128#[cfg_attr(coverage_nightly, coverage(off))]
129#[path = "tests/timers.rs"]
130mod tests;
darkbio_clock/timers.rs

darkbio_clock/
timers.rs
