code_moniker_workspace/live/
watcher.rs

1use std::path::PathBuf;
2use std::sync::mpsc;
3use std::thread::{self, JoinHandle};
4use std::time::Duration;
5
6use notify::{Config, Event, RecursiveMode, Watcher};
7
8use super::model::{WorkspaceLiveEvent, WorkspaceWatchRoot};
9use super::roots::{WorkspaceEventClassifier, watch_paths_for};
10
11pub struct LiveWorkspaceWatcher {
12	_watcher: WorkspaceWatcherBackend,
13	_worker: JoinHandle<()>,
14	watched_paths: usize,
15	warnings: Vec<String>,
16}
17
18const LIVE_EVENT_DEBOUNCE: Duration = Duration::from_millis(50);
19
20impl LiveWorkspaceWatcher {
21	pub fn start<F>(roots: Vec<WorkspaceWatchRoot>, publish: F) -> anyhow::Result<Self>
22	where
23		F: Fn(WorkspaceLiveEvent) + Send + 'static,
24	{
25		Self::start_with_backend(roots, WorkspaceWatcherBackendKind::Recommended, publish)
26	}
27
28	pub fn start_polling<F>(roots: Vec<WorkspaceWatchRoot>, publish: F) -> anyhow::Result<Self>
29	where
30		F: Fn(WorkspaceLiveEvent) + Send + 'static,
31	{
32		Self::start_with_backend(roots, WorkspaceWatcherBackendKind::Polling, publish)
33	}
34
35	fn start_with_backend<F>(
36		roots: Vec<WorkspaceWatchRoot>,
37		backend: WorkspaceWatcherBackendKind,
38		publish: F,
39	) -> anyhow::Result<Self>
40	where
41		F: Fn(WorkspaceLiveEvent) + Send + 'static,
42	{
43		let watch_targets = watch_paths_for(&roots);
44		let classifier = WorkspaceEventClassifier::new(roots);
45		let (tx, worker) = watcher_event_channel(publish);
46		let mut watcher = new_watcher(backend, classifier.clone(), tx)?;
47		let (watched_paths, warnings) = watch_target_paths(&mut watcher, &watch_targets);
48
49		Ok(Self {
50			_watcher: watcher,
51			_worker: worker,
52			watched_paths,
53			warnings,
54		})
55	}
56
57	pub fn status(&self) -> Option<String> {
58		if self.watched_paths == 0 {
59			if !self.warnings.is_empty() {
60				return Some(format!(
61					"live store disabled: no source path could be watched ({})",
62					self.warnings.join("; ")
63				));
64			}
65			return Some("live store disabled: no source path could be watched".to_string());
66		}
67		if self.warnings.is_empty() {
68			return Some(format!(
69				"live store watching {} path(s)",
70				self.watched_paths
71			));
72		}
73		Some(format!(
74			"live store watching {} path(s), {} warning(s)",
75			self.watched_paths,
76			self.warnings.len()
77		))
78	}
79}
80
81enum WorkspaceWatcherBackend {
82	Recommended(notify::RecommendedWatcher),
83	Polling(notify::PollWatcher),
84}
85
86impl WorkspaceWatcherBackend {
87	fn watch(&mut self, path: &std::path::Path, mode: RecursiveMode) -> notify::Result<()> {
88		match self {
89			Self::Recommended(watcher) => watcher.watch(path, mode),
90			Self::Polling(watcher) => watcher.watch(path, mode),
91		}
92	}
93}
94
95#[derive(Clone, Copy)]
96enum WorkspaceWatcherBackendKind {
97	Recommended,
98	Polling,
99}
100
101fn watcher_event_channel<F>(publish: F) -> (mpsc::Sender<WorkspaceLiveEvent>, JoinHandle<()>)
102where
103	F: Fn(WorkspaceLiveEvent) + Send + 'static,
104{
105	let (tx, rx) = mpsc::channel();
106	let worker = thread::spawn(move || publish_coalesced_events(rx, publish));
107	(tx, worker)
108}
109
110fn new_watcher(
111	backend: WorkspaceWatcherBackendKind,
112	classifier: WorkspaceEventClassifier,
113	tx: mpsc::Sender<WorkspaceLiveEvent>,
114) -> anyhow::Result<WorkspaceWatcherBackend> {
115	match backend {
116		WorkspaceWatcherBackendKind::Recommended => Ok(WorkspaceWatcherBackend::Recommended(
117			notify::RecommendedWatcher::new(
118				move |event| publish_classified_event(&classifier, &tx, event),
119				Config::default(),
120			)?,
121		)),
122		WorkspaceWatcherBackendKind::Polling => {
123			Ok(WorkspaceWatcherBackend::Polling(notify::PollWatcher::new(
124				move |event| publish_classified_event(&classifier, &tx, event),
125				polling_watcher_config(),
126			)?))
127		}
128	}
129}
130
131fn publish_classified_event(
132	classifier: &WorkspaceEventClassifier,
133	tx: &mpsc::Sender<WorkspaceLiveEvent>,
134	event: notify::Result<Event>,
135) {
136	let Ok(event) = event else {
137		return;
138	};
139	if let Some(store_event) = classifier.classify_event(&event) {
140		let _ = tx.send(store_event);
141	}
142}
143
144fn polling_watcher_config() -> Config {
145	Config::default()
146		.with_poll_interval(Duration::from_millis(50))
147		.with_compare_contents(true)
148}
149
150fn watch_target_paths(
151	watcher: &mut WorkspaceWatcherBackend,
152	targets: &[PathBuf],
153) -> (usize, Vec<String>) {
154	let mut warnings = Vec::new();
155	let mut watched_paths = 0;
156	for path in targets {
157		match watcher.watch(path.as_path(), RecursiveMode::Recursive) {
158			Ok(()) => watched_paths += 1,
159			Err(error) => warnings.push(format!("{}: {error}", path.display())),
160		}
161	}
162	(watched_paths, warnings)
163}
164
165fn publish_coalesced_events<F>(rx: mpsc::Receiver<WorkspaceLiveEvent>, publish: F)
166where
167	F: Fn(WorkspaceLiveEvent),
168{
169	while let Ok(first) = rx.recv() {
170		let mut event = first;
171		while let Ok(next) = rx.recv_timeout(LIVE_EVENT_DEBOUNCE) {
172			event = event.coalesce(next);
173		}
174		publish(event);
175	}
176}
code_moniker_workspace/live/watcher.rs

code_moniker_workspace/live/
watcher.rs
