dynamic_config/remote/
watch.rs

1//! Stopping a blocking watch.
2//!
3//! A watch nobody owns is a leak nobody asked for, so the handle stops the
4//! loop when it is dropped and `RemoteWatch::detach` is how a caller says
5//! *this one really should run forever*. Only blocking loops need this: an
6//! async watch is a future, and dropping it is stopping it.
7
8use std::sync::{Arc, Weak};
9use std::time::Duration;
10
11use crate::sync::atomic::{AtomicBool, Ordering};
12
13/// A running blocking watch, from the caller's side.
14///
15/// Dropping it stops the loop — the same contract the file watcher's
16/// `WatchHandle` has, for the same reason: a watch nobody owns is a leak nobody
17/// asked for. [`detach`](Self::detach) is the way to say *this one really should
18/// run forever*.
19///
20/// Only blocking loops need this. An async watch is a future: drop it and it is
21/// cancelled, on any executor.
22///
23/// ```no_run
24/// # use dynamic_config::RemoteWatch;
25/// # struct Consul;
26/// # impl Consul {
27/// #     fn watch(&self, _: dynamic_config::Watching, _: fn(dynamic_config::Fetched) -> Result<(), dynamic_config::Error>) -> Result<(), dynamic_config::Error> { Ok(()) }
28/// # }
29/// # fn example(consul: Consul) {
30/// # fn apply(_: dynamic_config::Fetched) -> Result<(), dynamic_config::Error> { Ok(()) }
31/// let watch = RemoteWatch::new();
32/// let watching = watch.watching();
33///
34/// std::thread::spawn(move || consul.watch(watching, apply));
35///
36/// // ... and later, or by dropping `watch`:
37/// watch.stop();
38/// # }
39/// ```
40#[must_use = "dropping the handle stops the watch; bind it, or call `.detach()` \
41              to watch for the rest of the process"]
42#[derive(Debug)]
43pub struct RemoteWatch {
44    running: Arc<AtomicBool>,
45}
46
47impl RemoteWatch {
48    /// A handle for a watch that has not been handed to a loop yet.
49    pub fn new() -> Self {
50        Self {
51            running: Arc::new(AtomicBool::new(true)),
52        }
53    }
54
55    /// The loop's half of this handle.
56    ///
57    /// Hand it to the watch; keep the `RemoteWatch` yourself.
58    #[must_use]
59    pub fn watching(&self) -> Watching {
60        Watching {
61            running: Arc::downgrade(&self.running),
62        }
63    }
64
65    /// Stops the loop at its next check.
66    ///
67    /// *At its next check* is the whole caveat, and it is not small: a loop
68    /// parked in a blocking query does not return until the store answers or
69    /// the wait expires, so the store's wait time is the worst-case delay. Each
70    /// companion crate documents its own.
71    pub fn stop(&self) {
72        self.running.store(false, Ordering::Release);
73    }
74
75    /// Whether the loop has been told to stop.
76    #[must_use]
77    pub fn is_stopped(&self) -> bool {
78        !self.running.load(Ordering::Acquire)
79    }
80
81    /// Watches for the remainder of the process.
82    ///
83    /// Leaks the handle on purpose, exactly as the file watcher's
84    /// `WatchHandle::detach` does: a watch that must never stop has no owner to
85    /// hold it, and pretending otherwise is how it ends up stopped at the end of
86    /// `main`'s first statement.
87    pub fn detach(self) {
88        std::mem::forget(self);
89    }
90}
91
92impl Default for RemoteWatch {
93    fn default() -> Self {
94        Self::new()
95    }
96}
97
98impl Drop for RemoteWatch {
99    fn drop(&mut self) {
100        self.stop();
101    }
102}
103
104/// The loop's half of a [`RemoteWatch`].
105///
106/// A `Weak`, so a handle that is dropped without anyone remembering to call
107/// `stop` still ends the loop: the upgrade fails and
108/// [`keep_going`](Self::keep_going) answers `false`.
109#[derive(Debug, Clone)]
110pub struct Watching {
111    running: Weak<AtomicBool>,
112}
113
114impl Watching {
115    /// Whether the loop should go round again.
116    ///
117    /// `false` once the caller called [`RemoteWatch::stop`] or dropped the
118    /// handle. Check it before every request, not only after one: a loop that
119    /// checks only on the way out issues one more query than it was asked to.
120    #[must_use]
121    pub fn keep_going(&self) -> bool {
122        self.running
123            .upgrade()
124            .is_some_and(|running| running.load(Ordering::Acquire))
125    }
126
127    /// Sleeps for `total`, waking early if the watch is stopped.
128    ///
129    /// The polling loop every blocking store crate writes: sleep a slice,
130    /// check [`keep_going`](Self::keep_going), repeat — so a stopped watch
131    /// ends within a quarter second instead of at the end of its interval.
132    /// Here once, rather than once per store crate.
133    pub fn sleep_for(&self, total: Duration) {
134        const SLICE: Duration = Duration::from_millis(250);
135
136        let mut slept = Duration::ZERO;
137
138        while slept < total && self.keep_going() {
139            std::thread::sleep(SLICE.min(total - slept));
140            slept += SLICE;
141        }
142    }
143
144    /// A token for a watch that should never stop.
145    ///
146    /// For a loop the caller genuinely wants to outlive everything, so there is
147    /// no handle to hold. Prefer [`RemoteWatch::detach`], which says the same
148    /// thing at the point where somebody decided it.
149    #[must_use]
150    pub fn forever() -> Self {
151        // A `Weak` that can never upgrade would stop the loop immediately, so
152        // this leaks one live flag — one allocation, once, for the life of the
153        // process.
154        let running = Box::leak(Box::new(Arc::new(AtomicBool::new(true))));
155
156        Self {
157            running: Arc::downgrade(running),
158        }
159    }
160}
dynamic_config/remote/watch.rs

dynamic_config/remote/
watch.rs
