1use itertools::Either;
2use notify_debouncer_full::{
3 DebouncedEvent, Debouncer, new_debouncer,
4 notify::{
5 Event, EventKind, RecommendedWatcher, RecursiveMode,
6 event::{DataChange, ModifyKind, RenameMode},
7 },
8};
9use nu_engine::{ClosureEval, command_prelude::*};
10use nu_protocol::{
11 Signals,
12 engine::Closure,
13 report_shell_error, report_shell_warning,
14 shell_error::{generic::GenericError, io::IoError},
15};
16use std::{
17 path::{Path, PathBuf},
18 sync::mpsc::{Receiver, RecvTimeoutError, channel},
19 time::Duration,
20};
21
22const CHECK_CTRL_C_FREQUENCY: Duration = Duration::from_millis(100);
24const DEFAULT_WATCH_DEBOUNCE_DURATION: Duration = Duration::from_millis(100);
25
26enum WatchGlob {
28 Legacy(nu_glob::Pattern),
29 DcGlob(nu_glob::dc_glob::DcPattern),
30}
31
32impl WatchGlob {
33 fn matches_path(&self, path: &Path) -> bool {
34 match self {
35 WatchGlob::Legacy(p) => p.matches_path(path),
36 WatchGlob::DcGlob(p) => p.matches_path(path),
37 }
38 }
39}
40
41#[derive(Clone)]
42pub struct Watch;
43
44impl Command for Watch {
45 fn name(&self) -> &str {
46 "watch"
47 }
48
49 fn description(&self) -> &str {
50 "Watch for file changes and execute Nu code when they happen."
51 }
52
53 fn extra_description(&self) -> &str {
54 "When run without a closure, `watch` returns a stream of events instead."
55 }
56
57 fn search_terms(&self) -> Vec<&str> {
58 vec!["watcher", "reload", "filesystem"]
59 }
60
61 fn signature(&self) -> nu_protocol::Signature {
62 Signature::build("watch")
63 .input_output_types(vec![
64 (Type::Nothing, Type::Nothing),
65 (
66 Type::Nothing,
67 Type::Table(
68 vec![
69 ("operation".into(), Type::String),
70 ("path".into(), Type::one_of([Type::String, Type::Nothing])),
71 (
72 "new_path".into(),
73 Type::one_of([Type::String, Type::Nothing]),
74 ),
75 ]
76 .into(),
77 ),
78 ),
79 ])
80 .required(
81 "path",
82 SyntaxShape::Filepath,
83 "The path to watch. Can be a file or directory.",
84 )
85 .optional(
86 "closure",
87 SyntaxShape::Closure(Some(vec![
88 SyntaxShape::String,
89 SyntaxShape::String,
90 SyntaxShape::String,
91 ])),
92 "Some Nu code to run whenever a file changes. \
93 The closure will be passed `operation`, `path`, \
94 and `new_path` (for renames only) arguments in that order (deprecated).",
95 )
96 .named(
97 "debounce",
98 SyntaxShape::Duration,
99 "Debounce changes for this duration (default: 100ms). \
100 Adjust if you find that single writes are reported as multiple events.",
101 Some('d'),
102 )
103 .named(
104 "glob",
105 SyntaxShape::String,
107 "Only report changes for files that match this glob pattern (default: all files)",
108 Some('g'),
109 )
110 .named(
111 "recursive",
112 SyntaxShape::Boolean,
113 "Watch all directories under `<path>` recursively. \
114 Will be ignored if `<path>` is a file (default: true).",
115 Some('r'),
116 )
117 .switch(
118 "quiet",
119 "Hide the initial status message (default: false).",
120 Some('q'),
121 )
122 .switch(
123 "verbose",
124 "Operate in verbose mode (default: false).",
125 Some('v'),
126 )
127 .category(Category::FileSystem)
128 }
129
130 fn run(
131 &self,
132 engine_state: &EngineState,
133 stack: &mut Stack,
134 call: &Call,
135 _input: PipelineData,
136 ) -> Result<PipelineData, ShellError> {
137 let head = call.head;
138 let path = {
139 let cwd = engine_state.cwd_as_string(Some(stack))?;
140 let path_arg: Spanned<String> = call.req(engine_state, stack, 0)?;
141 let path_no_whitespace = path_arg
142 .item
143 .trim_end_matches(|x| matches!(x, '\x09'..='\x0d'));
144
145 nu_path::absolute_with(path_no_whitespace, cwd).map_err(|err| {
146 ShellError::Io(IoError::new(
147 err,
148 path_arg.span,
149 PathBuf::from(path_no_whitespace),
150 ))
151 })?
152 };
153 let closure: Option<Spanned<Closure>> = call.opt(engine_state, stack, 1)?;
154 let verbose = call.has_flag(engine_state, stack, "verbose")?;
155 let quiet = call.has_flag(engine_state, stack, "quiet")?;
156 let debounce_duration: Duration = call
157 .get_flag(engine_state, stack, "debounce")?
158 .unwrap_or(DEFAULT_WATCH_DEBOUNCE_DURATION);
159
160 let glob_pattern = call
161 .get_flag::<Spanned<String>>(engine_state, stack, "glob")?
162 .map(|glob| {
163 let absolute_path = path.join(glob.item);
164 if verbose {
165 eprintln!("Absolute glob path: {absolute_path:?}");
166 }
167 let path_str = absolute_path.to_string_lossy();
168 if nu_experimental::DC_GLOB.get() {
169 nu_glob::dc_glob::DcPattern::new(path_str.as_ref())
170 .map(WatchGlob::DcGlob)
171 .map_err(|_| ShellError::TypeMismatch {
172 err_message: "Glob pattern is invalid".to_string(),
173 span: glob.span,
174 })
175 } else {
176 nu_glob::Pattern::new(path_str.as_ref())
177 .map(WatchGlob::Legacy)
178 .map_err(|_| ShellError::TypeMismatch {
179 err_message: "Glob pattern is invalid".to_string(),
180 span: glob.span,
181 })
182 }
183 })
184 .transpose()?;
185
186 let recursive_mode = {
187 let recursive_flag = call
188 .get_flag::<bool>(engine_state, stack, "recursive")?
189 .unwrap_or(true);
190 match recursive_flag {
191 true => RecursiveMode::Recursive,
192 false => RecursiveMode::NonRecursive,
193 }
194 };
195
196 let iter = {
197 let (tx, rx) = channel();
198
199 let mut debouncer = new_debouncer(debounce_duration, None, move |result| {
200 let _ = tx.send(result);
201 })
202 .map_err(|err| {
203 ShellError::Generic(GenericError::new(
204 "Failed to create watcher",
205 err.to_string(),
206 call.head,
207 ))
208 })?;
209
210 debouncer.watch(&path, recursive_mode).map_err(|err| {
211 ShellError::Generic(GenericError::new(
212 "Failed to create watcher",
213 err.to_string(),
214 call.head,
215 ))
216 })?;
217
218 WatchIterator::new(debouncer, rx, engine_state.signals().clone())
219 };
220
221 if let Some(closure) = closure {
222 report_shell_warning(
223 Some(stack),
224 engine_state,
225 &ShellWarning::Deprecated {
226 dep_type: "Argument".into(),
227 label: "remove this".into(),
228 span: closure.span,
229 help: "Since 0.113.0, running a closure with `watch` is deprecated. \
230 Instead, use `watch` by piping its output to `each` \
231 or as the source of a `for` loop."
232 .to_string()
233 .into(),
234 report_mode: nu_protocol::ReportMode::FirstUse,
235 },
236 );
237
238 #[allow(deprecated)]
239 run_closure(
240 engine_state,
241 stack,
242 head,
243 quiet,
244 verbose,
245 &path,
246 glob_pattern,
247 iter,
248 closure.item,
249 )
250 } else {
251 let out = iter
252 .flat_map(|e| match e {
253 Ok(events) => Either::Right(events.into_iter().map(Ok)),
254 Err(err) => Either::Left(std::iter::once(Err(err))),
255 })
256 .filter_map(move |e| match e {
257 Ok(ev) if glob_filter(glob_pattern.as_ref(), &ev) => Some(ev.into_value(head)),
258 Ok(_) => None,
259 Err(err) => Some(Value::error(err, head)),
260 })
261 .into_pipeline_data(head, engine_state.signals().clone());
262 Ok(out)
263 }
264 }
265
266 fn examples(&self) -> Vec<Example<'_>> {
267 vec![
268 Example {
269 description: "Run `cargo test` whenever a Rust file changes.",
270 example: "for _ in (watch . --glob=**/*.rs) { cargo test }",
271 result: None,
272 },
273 Example {
274 description: "Watch all changes in the current directory.",
275 example: "watch . | each { print }",
276 result: None,
277 },
278 Example {
279 description: "Filter, limit and modify `watch`'s output by using it as part of a pipeline.",
280 example: r#"watch /foo/bar
281 | where operation == Create
282 | first 5
283 | each {|e| $"New file!: ($e.path)" }
284 | to text
285 | save --append changes_in_bar.log"#,
286 result: None,
287 },
288 Example {
289 description: "Print file changes with a debounce time of 5 minutes.",
290 example: r#"watch /foo/bar --debounce 5min | each {|e| $"Registered ($e.operation) on ($e.path)" | print }"#,
291 result: None,
292 },
293 Example {
294 description: "Note: if you are looking to run a command every N units of time, this can be accomplished with a loop and sleep.",
295 example: "loop { command; sleep duration }",
296 result: None,
297 },
298 Example {
299 description: "Run `cargo test` whenever a Rust file changes (with the deprecated closure argument).",
300 example: "watch . --glob=**/*.rs {|| cargo test }",
301 result: None,
302 },
303 ]
304 }
305}
306
307fn glob_filter(glob: Option<&WatchGlob>, ev: &WatchEvent) -> bool {
308 let Some(glob) = glob else { return true };
309 let Some(path) = ev.path.as_deref().or(ev.new_path.as_deref()) else {
310 return false;
311 };
312
313 glob.matches_path(path)
314}
315
316#[allow(clippy::too_many_arguments)]
317#[inline]
318#[deprecated(since = "0.113.0")]
319fn run_closure(
320 engine_state: &EngineState,
321 stack: &mut Stack,
322 head: Span,
323 quiet: bool,
324 verbose: bool,
325 path: &Path,
326 glob_pattern: Option<WatchGlob>,
327 iter: WatchIterator,
328 closure: Closure,
329) -> Result<PipelineData, ShellError> {
330 if !quiet {
331 eprintln!("Now watching files at {path:?}. Press ctrl+c to abort.");
332 }
333
334 let mut closure = ClosureEval::new(engine_state, stack, closure);
335 for events in iter {
336 for event in events? {
337 let matches_glob = glob_filter(glob_pattern.as_ref(), &event);
338
339 if verbose && glob_pattern.is_some() {
340 eprintln!("Matches glob: {matches_glob}");
341 }
342
343 if matches_glob {
344 let result = closure
345 .add_arg(event.operation.into_value(head))?
346 .add_arg(event.path.into_value(head))?
347 .add_arg(event.new_path.into_value(head))?
348 .run_with_input(PipelineData::empty());
349
350 match result {
351 Ok(val) => val.print_table(engine_state, stack, false, false)?,
352 Err(err) => report_shell_error(Some(stack), engine_state, &err),
353 };
354 }
355 }
356 }
357
358 Ok(PipelineData::empty())
359}
360
361#[derive(IntoValue)]
362struct WatchEvent {
363 operation: WatchEventKind,
364 path: Option<PathBuf>,
365 new_path: Option<PathBuf>,
366}
367
368#[derive(IntoValue)]
369#[nu_value(rename_all = "UpperCamelCase")]
370enum WatchEventKind {
371 Create,
372 Write,
373 Rename,
374 Remove,
375}
376
377impl TryFrom<EventKind> for WatchEventKind {
378 type Error = ();
379
380 fn try_from(value: EventKind) -> Result<Self, Self::Error> {
381 Ok(match value {
382 EventKind::Create(_) => Self::Create,
383 EventKind::Remove(_) => Self::Remove,
384 EventKind::Modify(
385 ModifyKind::Data(DataChange::Content | DataChange::Any) | ModifyKind::Any,
386 ) => Self::Write,
387 EventKind::Modify(ModifyKind::Name(
388 RenameMode::Both | RenameMode::From | RenameMode::To,
389 )) => Self::Rename,
390 _ => return Err(()),
391 })
392 }
393}
394
395impl TryFrom<DebouncedEvent> for WatchEvent {
396 type Error = ();
397
398 fn try_from(ev: DebouncedEvent) -> Result<Self, Self::Error> {
399 let DebouncedEvent {
401 event: Event {
402 kind, mut paths, ..
403 },
404 ..
405 } = ev;
406
407 let (path, new_path) = match paths.as_mut_slice() {
408 [path] => (std::mem::take(path), None),
409 [path, new_path] => (std::mem::take(path), Some(std::mem::take(new_path))),
410 _ => return Err(()),
411 };
412
413 if let EventKind::Modify(ModifyKind::Name(RenameMode::To)) = kind {
414 Ok(WatchEvent {
415 operation: WatchEventKind::Rename,
416 path: None,
417 new_path: Some(path),
418 })
419 } else {
420 Ok(WatchEvent {
421 operation: kind.try_into()?,
422 path: Some(path),
423 new_path,
424 })
425 }
426 }
427}
428
429struct WatchIterator {
430 _debouncer: Debouncer<RecommendedWatcher, notify_debouncer_full::RecommendedCache>,
432 rx: Option<Receiver<notify_debouncer_full::DebounceEventResult>>,
433 signals: Signals,
434}
435
436impl WatchIterator {
437 fn new(
438 debouncer: Debouncer<RecommendedWatcher, notify_debouncer_full::RecommendedCache>,
439 rx: Receiver<notify_debouncer_full::DebounceEventResult>,
440 signals: Signals,
441 ) -> Self {
442 Self {
443 _debouncer: debouncer,
444 rx: Some(rx),
445 signals,
446 }
447 }
448}
449
450impl Iterator for WatchIterator {
451 type Item = Result<Vec<WatchEvent>, ShellError>;
452
453 fn next(&mut self) -> Option<Self::Item> {
454 let rx = self.rx.as_ref()?;
455 while !self.signals.interrupted() {
456 let x = match rx.recv_timeout(CHECK_CTRL_C_FREQUENCY) {
457 Ok(x) => x,
458 Err(RecvTimeoutError::Timeout) => continue,
459 Err(RecvTimeoutError::Disconnected) => {
460 self.rx = None;
461 return Some(Err(ShellError::Generic(GenericError::new_internal(
462 "Disconnected",
463 "Unexpected disconnect from file watcher",
464 ))));
465 }
466 };
467
468 let Ok(events) = x else {
469 self.rx = None;
470 return Some(Err(ShellError::Generic(GenericError::new_internal(
471 "Receiving events failed",
472 "Unexpected errors when receiving events",
473 ))));
474 };
475
476 let watch_events = events
477 .into_iter()
478 .filter_map(|ev| WatchEvent::try_from(ev).ok())
479 .collect::<Vec<_>>();
480
481 return Some(Ok(watch_events));
482 }
483 self.rx = None;
484 None
485 }
486}