reifydb_runtime/sync/
waiter.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::fmt;
5
6use reifydb_value::value::duration::Duration;
7
8use crate::sync::{condvar::Condvar, mutex::Mutex};
9
10pub struct WaiterHandle {
11	notified: Mutex<bool>,
12	condvar: Condvar,
13	on_notify: Mutex<Option<Box<dyn FnOnce() + Send>>>,
14}
15
16impl fmt::Debug for WaiterHandle {
17	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
18		f.debug_struct("WaiterHandle").finish_non_exhaustive()
19	}
20}
21
22impl Default for WaiterHandle {
23	fn default() -> Self {
24		Self::new()
25	}
26}
27
28impl WaiterHandle {
29	pub fn new() -> Self {
30		Self {
31			notified: Mutex::new(false),
32			condvar: Condvar::new(),
33			on_notify: Mutex::new(None),
34		}
35	}
36
37	pub fn with_callback(callback: Box<dyn FnOnce() + Send>) -> Self {
38		Self {
39			notified: Mutex::new(false),
40			condvar: Condvar::new(),
41			on_notify: Mutex::new(Some(callback)),
42		}
43	}
44
45	pub fn notify(&self) {
46		let mut guard = self.notified.lock();
47		*guard = true;
48		self.condvar.notify_one();
49		drop(guard);
50		if let Some(callback) = self.on_notify.lock().take() {
51			callback();
52		}
53	}
54
55	pub fn wait_timeout(&self, timeout: Duration) -> bool {
56		let mut guard = self.notified.lock();
57		if *guard {
58			return true;
59		}
60		!self.condvar.wait_for(&mut guard, timeout).timed_out()
61	}
62}
63
64#[cfg(test)]
65mod tests {
66	use std::sync::{
67		Arc,
68		atomic::{AtomicUsize, Ordering},
69	};
70
71	use super::*;
72
73	#[test]
74	fn callback_fires_exactly_once() {
75		let count = Arc::new(AtomicUsize::new(0));
76		let c = count.clone();
77		let waiter = WaiterHandle::with_callback(Box::new(move || {
78			c.fetch_add(1, Ordering::SeqCst);
79		}));
80
81		waiter.notify();
82		waiter.notify();
83
84		assert_eq!(count.load(Ordering::SeqCst), 1, "one-shot callback must fire exactly once");
85		assert!(
86			waiter.wait_timeout(Duration::from_milliseconds(0).unwrap()),
87			"an already-notified waiter returns immediately"
88		);
89	}
90}
reifydb_runtime/sync/waiter.rs

reifydb_runtime/sync/
waiter.rs
