pipedash_core/application/
config_watcher.rs

1use std::path::Path;
2use std::sync::Arc;
3use std::time::{
4    Duration,
5    SystemTime,
6};
7
8use tokio::sync::Mutex;
9use tokio::task::JoinHandle;
10use tokio::time::{
11    interval,
12    MissedTickBehavior,
13};
14use tokio_util::sync::CancellationToken;
15
16use super::ProviderService;
17use crate::infrastructure::ConfigState;
18
19const POLL: Duration = Duration::from_secs(5);
20
21type FileStamp = (SystemTime, u64);
22
23pub struct ConfigWatcher {
24    provider_service: Arc<ProviderService>,
25    config_state: Arc<ConfigState>,
26    poll: Duration,
27    cancel: CancellationToken,
28    task: Mutex<Option<JoinHandle<()>>>,
29}
30
31impl ConfigWatcher {
32    pub fn new(provider_service: Arc<ProviderService>, config_state: Arc<ConfigState>) -> Self {
33        Self {
34            provider_service,
35            config_state,
36            poll: POLL,
37            cancel: CancellationToken::new(),
38            task: Mutex::new(None),
39        }
40    }
41
42    #[cfg(test)]
43    pub(crate) fn with_poll(mut self, poll: Duration) -> Self {
44        self.poll = poll;
45        self
46    }
47
48    pub async fn start(&self) {
49        let mut task = self.task.lock().await;
50        if task.is_some() || self.cancel.is_cancelled() {
51            return;
52        }
53
54        *task = Some(tokio::spawn(watch(
55            Arc::clone(&self.provider_service),
56            Arc::clone(&self.config_state),
57            self.poll,
58            self.cancel.clone(),
59        )));
60    }
61
62    pub async fn stop(&self) {
63        self.cancel.cancel();
64        let task = self.task.lock().await.take();
65        if let Some(task) = task {
66            if let Err(e) = task.await {
67                tracing::error!(error = %e, "ConfigWatcher task failed");
68            }
69        }
70    }
71}
72
73impl Drop for ConfigWatcher {
74    fn drop(&mut self) {
75        self.cancel.cancel();
76    }
77}
78
79fn stamp(path: &Path) -> Option<FileStamp> {
80    let metadata = std::fs::metadata(path).ok()?;
81
82    Some((metadata.modified().ok()?, metadata.len()))
83}
84
85async fn watch(
86    provider_service: Arc<ProviderService>, config_state: Arc<ConfigState>, poll: Duration,
87    cancel: CancellationToken,
88) {
89    let path = config_state.config_path().clone();
90    let mut applied = None;
91    let mut failed = None;
92    let mut tick = interval(poll);
93    tick.set_missed_tick_behavior(MissedTickBehavior::Skip);
94
95    loop {
96        tokio::select! {
97            _ = cancel.cancelled() => break,
98            _ = tick.tick() => {
99                let current = stamp(&path);
100                if current.is_none() || current == applied {
101                    continue;
102                }
103
104                match provider_service.apply_config_file(&config_state).await {
105                    Ok(_) => applied = current,
106                    Err(e) if current != failed => {
107                        failed = current;
108                        tracing::warn!(path = %path.display(), error = %e, "Failed to apply config file changes");
109                    }
110                    Err(e) => {
111                        tracing::debug!(path = %path.display(), error = %e, "Failed to apply config file changes");
112                    }
113                }
114            }
115        }
116    }
117}
pipedash_core/application/config_watcher.rs

pipedash_core/application/
config_watcher.rs
