feat(unstable): Support --watch flag for bundle and fmt subcommands (#8276)

This commit adds support for "--watch" flag for "bundle" 
and "fmt" subcommands.

In addition to this, it refactors "run --watch" command so that
module resolution will occur every time the file watcher detects 
file addition/deletion, which allows the watcher to observe a file 
that is newly added to the dependency as well.

cli/file_watcher.rs

@@ -4,7 +4,7 @@ use crate::colors;
 use core::task::{Context, Poll};
 use deno_core::error::AnyError;
 use deno_core::futures::stream::{Stream, StreamExt};
-use deno_core::futures::Future;
+use deno_core::futures::{Future, FutureExt};
 use notify::event::Event as NotifyEvent;
 use notify::event::EventKind;
 use notify::Config;
@@ -18,22 +18,21 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::select;
-use tokio::time::{interval, Interval};
+use tokio::time::{delay_for, Delay};
 
 const DEBOUNCE_INTERVAL_MS: Duration = Duration::from_millis(200);
 
-// TODO(bartlomieju): rename
+type FileWatcherFuture<T> = Pin<Box<dyn Future<Output = Result<T, AnyError>>>>;
-type WatchFuture = Pin<Box<dyn Future<Output = Result<(), AnyError>>>>;
 
 struct Debounce {
-  interval: Interval,
+  delay: Delay,
   event_detected: Arc<AtomicBool>,
 }
 
 impl Debounce {
   fn new() -> Self {
     Self {
-      interval: interval(DEBOUNCE_INTERVAL_MS),
+      delay: delay_for(DEBOUNCE_INTERVAL_MS),
       event_detected: Arc::new(AtomicBool::new(false)),
     }
   }
@@ -53,13 +52,18 @@ impl Stream for Debounce {
       inner.event_detected.store(false, Ordering::Relaxed);
       Poll::Ready(Some(()))
     } else {
-      let _ = inner.interval.poll_tick(cx);
+      match inner.delay.poll_unpin(cx) {
-      Poll::Pending
+        Poll::Ready(_) => {
+          inner.delay = delay_for(DEBOUNCE_INTERVAL_MS);
+          Poll::Pending
+        }
+        Poll::Pending => Poll::Pending,
+      }
     }
   }
 }
 
-async fn error_handler(watch_future: WatchFuture) {
+async fn error_handler(watch_future: FileWatcherFuture<()>) {
   let result = watch_future.await;
   if let Err(err) = result {
     let msg = format!("{}: {}", colors::red_bold("error"), err.to_string(),);
@@ -67,19 +71,37 @@ async fn error_handler(watch_future: WatchFuture) {
   }
 }
 
-pub async fn watch_func<F>(
+/// This function adds watcher functionality to subcommands like `fmt` or `lint`.
-  paths: &[PathBuf],
+/// The difference from [`watch_func_with_module_resolution`] is that this doesn't depend on
-  closure: F,
+/// [`ModuleGraph`].
+///
+/// - `target_resolver` is used for resolving file paths to be watched at every restarting of the watcher. The
+/// return value of this closure will then be passed to `operation` as an argument.
+///
+/// - `operation` is the actual operation we want to run every time the watcher detects file
+/// changes. For example, in the case where we would like to apply `fmt`, then `operation` would
+/// have the logic for it like calling `format_source_files`.
+///
+/// - `job_name` is just used for printing watcher status to terminal.
+///
+/// Note that the watcher will stop working if `target_resolver` fails at some point.
+///
+/// [`ModuleGraph`]: crate::module_graph::Graph
+pub async fn watch_func<F, G>(
+  target_resolver: F,
+  operation: G,
+  job_name: &str,
 ) -> Result<(), AnyError>
 where
-  F: Fn() -> WatchFuture,
+  F: Fn() -> Result<Vec<PathBuf>, AnyError>,
+  G: Fn(Vec<PathBuf>) -> FileWatcherFuture<()>,
 {
   let mut debounce = Debounce::new();
-  // This binding is required for the watcher to work properly without being dropped.
-  let _watcher = new_watcher(paths, &debounce)?;
 
   loop {
-    let func = error_handler(closure());
+    let paths = target_resolver()?;
+    let _watcher = new_watcher(&paths, &debounce)?;
+    let func = error_handler(operation(paths));
     let mut is_file_changed = false;
     select! {
       _ = debounce.next() => {
@@ -90,11 +112,95 @@ where
         );
       },
       _ = func => {},
-    }
+    };
+
     if !is_file_changed {
       info!(
-        "{} Process terminated! Restarting on file change...",
+        "{} {} finished! Restarting on file change...",
         colors::intense_blue("Watcher"),
+        job_name,
+      );
+      debounce.next().await;
+      info!(
+        "{} File change detected! Restarting!",
+        colors::intense_blue("Watcher"),
+      );
+    }
+  }
+}
+
+/// This function adds watcher functionality to subcommands like `run` or `bundle`.
+/// The difference from [`watch_func`] is that this does depend on [`ModuleGraph`].
+///
+/// - `module_resolver` is used for both resolving file paths to be watched at every restarting
+/// of the watcher and building [`ModuleGraph`] or [`ModuleSpecifier`] which will then be passed
+/// to `operation`.
+///
+/// - `operation` is the actual operation we want to run every time the watcher detects file
+/// changes. For example, in the case where we would like to bundle, then `operation` would
+/// have the logic for it like doing bundle with the help of [`ModuleGraph`].
+///
+/// - `job_name` is just used for printing watcher status to terminal.
+///
+/// Note that the watcher will try to continue watching files using the previously resolved
+/// data if `module_resolver` fails at some point, which means the watcher won't work at all
+/// if `module_resolver` fails at the first attempt.
+///
+/// [`ModuleGraph`]: crate::module_graph::Graph
+/// [`ModuleSpecifier`]: deno_core::ModuleSpecifier
+pub async fn watch_func_with_module_resolution<F, G, T>(
+  module_resolver: F,
+  operation: G,
+  job_name: &str,
+) -> Result<(), AnyError>
+where
+  F: Fn() -> FileWatcherFuture<(Vec<PathBuf>, T)>,
+  G: Fn(T) -> FileWatcherFuture<()>,
+  T: Clone,
+{
+  let mut debounce = Debounce::new();
+  // Store previous data. If module resolution fails at some point, the watcher will try to
+  // continue watching files using these data.
+  let mut paths = None;
+  let mut module = None;
+
+  loop {
+    match module_resolver().await {
+      Ok((next_paths, next_module)) => {
+        paths = Some(next_paths);
+        module = Some(next_module);
+      }
+      Err(e) => {
+        // If at least one of `paths` and `module` is `None`, the watcher cannot decide which files
+        // should be watched. So return the error immediately without watching anything.
+        if paths.is_none() || module.is_none() {
+          return Err(e);
+        }
+      }
+    }
+    // These `unwrap`s never cause panic since `None` is already checked above.
+    let cur_paths = paths.clone().unwrap();
+    let cur_module = module.clone().unwrap();
+
+    let _watcher = new_watcher(&cur_paths, &debounce)?;
+    let func = error_handler(operation(cur_module));
+    let mut is_file_changed = false;
+    select! {
+      _ = debounce.next() => {
+        is_file_changed = true;
+        info!(
+          "{} File change detected! Restarting!",
+          colors::intense_blue("Watcher"),
+        );
+      },
+      _ = func => {},
+    };
+
+    if !is_file_changed {
+      info!(
+        "{} {} finished! Restarting on file change...",
+        colors::intense_blue("Watcher"),
+        job_name,
       );
       debounce.next().await;
       info!(
@@ -125,7 +231,8 @@ fn new_watcher(
   watcher.configure(Config::PreciseEvents(true)).unwrap();
 
   for path in paths {
-    watcher.watch(path, RecursiveMode::NonRecursive)?;
+    // Ignore any error e.g. `PathNotFound`
+    let _ = watcher.watch(path, RecursiveMode::NonRecursive);
   }
 
   Ok(watcher)

cli/flags.rs

@@ -361,6 +361,7 @@ fn types_parse(flags: &mut Flags, _matches: &clap::ArgMatches) {
 }
 
 fn fmt_parse(flags: &mut Flags, matches: &clap::ArgMatches) {
+  flags.watch = matches.is_present("watch");
   let files = match matches.values_of("files") {
     Some(f) => f.map(PathBuf::from).collect(),
     None => vec![],
@@ -418,6 +419,8 @@ fn bundle_parse(flags: &mut Flags, matches: &clap::ArgMatches) {
     None
   };
 
+  flags.watch = matches.is_present("watch");
+
   flags.subcommand = DenoSubcommand::Bundle {
     source_file,
     out_file,
@@ -723,6 +726,7 @@ Ignore formatting a file by adding an ignore comment at the top of the file:
         .multiple(true)
         .required(false),
     )
+    .arg(watch_arg())
 }
 
 fn repl_subcommand<'a, 'b>() -> App<'a, 'b> {
@@ -793,6 +797,7 @@ fn bundle_subcommand<'a, 'b>() -> App<'a, 'b> {
         .required(true),
     )
     .arg(Arg::with_name("out_file").takes_value(true).required(false))
+    .arg(watch_arg())
     .about("Bundle module and dependencies into single file")
     .long_about(
       "Output a single JavaScript file with all dependencies.
@@ -1855,6 +1860,44 @@ mod tests {
         ..Flags::default()
       }
     );
+
+    let r = flags_from_vec_safe(svec!["deno", "fmt", "--watch", "--unstable"]);
+    assert_eq!(
+      r.unwrap(),
+      Flags {
+        subcommand: DenoSubcommand::Fmt {
+          ignore: vec![],
+          check: false,
+          files: vec![],
+        },
+        watch: true,
+        unstable: true,
+        ..Flags::default()
+      }
+    );
+
+    let r = flags_from_vec_safe(svec![
+      "deno",
+      "fmt",
+      "--check",
+      "--watch",
+      "--unstable",
+      "foo.ts",
+      "--ignore=bar.js"
+    ]);
+    assert_eq!(
+      r.unwrap(),
+      Flags {
+        subcommand: DenoSubcommand::Fmt {
+          ignore: vec![PathBuf::from("bar.js")],
+          check: true,
+          files: vec![PathBuf::from("foo.ts")],
+        },
+        watch: true,
+        unstable: true,
+        ..Flags::default()
+      }
+    );
   }
 
   #[test]
@@ -2405,6 +2448,29 @@ mod tests {
     );
   }
 
+  #[test]
+  fn bundle_watch() {
+    let r = flags_from_vec_safe(svec![
+      "deno",
+      "bundle",
+      "--watch",
+      "--unstable",
+      "source.ts"
+    ]);
+    assert_eq!(
+      r.unwrap(),
+      Flags {
+        subcommand: DenoSubcommand::Bundle {
+          source_file: "source.ts".to_string(),
+          out_file: None,
+        },
+        watch: true,
+        unstable: true,
+        ..Flags::default()
+      }
+    )
+  }
+
   #[test]
   fn run_import_map() {
     let r = flags_from_vec_safe(svec![

cli/fs_util.rs

@@ -88,8 +88,8 @@ pub fn is_supported_ext(path: &Path) -> bool {
 /// Collects file paths that satisfy the given predicate, by recursively walking `files`.
 /// If the walker visits a path that is listed in `ignore`, it skips descending into the directory.
 pub fn collect_files<P>(
-  files: Vec<PathBuf>,
+  files: &[PathBuf],
-  ignore: Vec<PathBuf>,
+  ignore: &[PathBuf],
   predicate: P,
 ) -> Result<Vec<PathBuf>, AnyError>
 where
@@ -99,15 +99,12 @@ where
 
   // retain only the paths which exist and ignore the rest
   let canonicalized_ignore: Vec<PathBuf> = ignore
-    .into_iter()
+    .iter()
     .filter_map(|i| i.canonicalize().ok())
     .collect();
 
-  let files = if files.is_empty() {
+  let cur_dir = [std::env::current_dir()?];
-    vec![std::env::current_dir()?]
+  let files = if files.is_empty() { &cur_dir } else { files };
-  } else {
-    files
-  };
 
   for file in files {
     for entry in WalkDir::new(file)
@@ -232,15 +229,14 @@ mod tests {
     let ignore_dir_files = ["g.d.ts", ".gitignore"];
     create_files(&ignore_dir_path, &ignore_dir_files);
 
-    let result =
+    let result = collect_files(&[root_dir_path], &[ignore_dir_path], |path| {
-      collect_files(vec![root_dir_path], vec![ignore_dir_path], |path| {
+      // exclude dotfiles
-        // exclude dotfiles
+      path
-        path
+        .file_name()
-          .file_name()
+        .and_then(|f| f.to_str())
-          .and_then(|f| f.to_str())
+        .map_or(false, |f| !f.starts_with('.'))
-          .map_or(false, |f| !f.starts_with('.'))
+    })
-      })
+    .unwrap();
-      .unwrap();
     let expected = [
       "a.ts",
       "b.js",

cli/main.rs

@@ -302,84 +302,130 @@ async fn bundle_command(
   source_file: String,
   out_file: Option<PathBuf>,
 ) -> Result<(), AnyError> {
-  let module_specifier = ModuleSpecifier::resolve_url_or_path(&source_file)?;
-
-  debug!(">>>>> bundle START");
-  let program_state = ProgramState::new(flags.clone())?;
-
-  info!(
-    "{} {}",
-    colors::green("Bundle"),
-    module_specifier.to_string()
-  );
-
-  let handler = Rc::new(RefCell::new(FetchHandler::new(
-    &program_state,
-    // when bundling, dynamic imports are only access for their type safety,
-    // therefore we will allow the graph to access any module.
-    Permissions::allow_all(),
-  )?));
-  let mut builder = module_graph::GraphBuilder::new(
-    handler,
-    program_state.maybe_import_map.clone(),
-    program_state.lockfile.clone(),
-  );
-  builder.add(&module_specifier, false).await?;
-  let graph = builder.get_graph();
-
   let debug = flags.log_level == Some(log::Level::Debug);
-  if !flags.no_check {
-    // TODO(@kitsonk) support bundling for workers
-    let lib = if flags.unstable {
-      module_graph::TypeLib::UnstableDenoWindow
-    } else {
-      module_graph::TypeLib::DenoWindow
-    };
-    let graph = graph.clone();
-    let result_info = graph.check(module_graph::CheckOptions {
-      debug,
-      emit: false,
-      lib,
-      maybe_config_path: flags.config_path.clone(),
-      reload: flags.reload,
-    })?;
 
-    debug!("{}", result_info.stats);
+  let module_resolver = || {
-    if let Some(ignored_options) = result_info.maybe_ignored_options {
+    let flags = flags.clone();
-      eprintln!("{}", ignored_options);
+    let source_file = source_file.clone();
+    async move {
+      let module_specifier =
+        ModuleSpecifier::resolve_url_or_path(&source_file)?;
+
+      debug!(">>>>> bundle START");
+      let program_state = ProgramState::new(flags.clone())?;
+
+      info!(
+        "{} {}",
+        colors::green("Bundle"),
+        module_specifier.to_string()
+      );
+
+      let handler = Rc::new(RefCell::new(FetchHandler::new(
+        &program_state,
+        // when bundling, dynamic imports are only access for their type safety,
+        // therefore we will allow the graph to access any module.
+        Permissions::allow_all(),
+      )?));
+      let mut builder = module_graph::GraphBuilder::new(
+        handler,
+        program_state.maybe_import_map.clone(),
+        program_state.lockfile.clone(),
+      );
+      builder.add(&module_specifier, false).await?;
+      let module_graph = builder.get_graph();
+
+      if !flags.no_check {
+        // TODO(@kitsonk) support bundling for workers
+        let lib = if flags.unstable {
+          module_graph::TypeLib::UnstableDenoWindow
+        } else {
+          module_graph::TypeLib::DenoWindow
+        };
+        let result_info =
+          module_graph.clone().check(module_graph::CheckOptions {
+            debug,
+            emit: false,
+            lib,
+            maybe_config_path: flags.config_path.clone(),
+            reload: flags.reload,
+          })?;
+
+        debug!("{}", result_info.stats);
+        if let Some(ignored_options) = result_info.maybe_ignored_options {
+          eprintln!("{}", ignored_options);
+        }
+        if !result_info.diagnostics.is_empty() {
+          return Err(generic_error(result_info.diagnostics.to_string()));
+        }
+      }
+
+      let mut paths_to_watch: Vec<PathBuf> = module_graph
+        .get_modules()
+        .iter()
+        .filter_map(|specifier| specifier.as_url().to_file_path().ok())
+        .collect();
+
+      if let Some(import_map) = program_state.flags.import_map_path.as_ref() {
+        paths_to_watch
+          .push(fs_util::resolve_from_cwd(std::path::Path::new(import_map))?);
+      }
+
+      Ok((paths_to_watch, module_graph))
     }
-    if !result_info.diagnostics.is_empty() {
+    .boxed_local()
-      return Err(generic_error(result_info.diagnostics.to_string()));
+  };
+
+  let operation = |module_graph: module_graph::Graph| {
+    let flags = flags.clone();
+    let out_file = out_file.clone();
+    async move {
+      let (output, stats, maybe_ignored_options) =
+        module_graph.bundle(module_graph::BundleOptions {
+          debug,
+          maybe_config_path: flags.config_path,
+        })?;
+
+      match maybe_ignored_options {
+        Some(ignored_options) if flags.no_check => {
+          eprintln!("{}", ignored_options);
+        }
+        _ => {}
+      }
+      debug!("{}", stats);
+
+      debug!(">>>>> bundle END");
+
+      if let Some(out_file) = out_file.as_ref() {
+        let output_bytes = output.as_bytes();
+        let output_len = output_bytes.len();
+        fs_util::write_file(out_file, output_bytes, 0o644)?;
+        info!(
+          "{} {:?} ({})",
+          colors::green("Emit"),
+          out_file,
+          colors::gray(&info::human_size(output_len as f64))
+        );
+      } else {
+        println!("{}", output);
+      }
+
+      Ok(())
     }
-  }
+    .boxed_local()
+  };
 
-  let (output, stats, maybe_ignored_options) =
+  if flags.watch {
-    graph.bundle(module_graph::BundleOptions {
+    file_watcher::watch_func_with_module_resolution(
-      debug,
+      module_resolver,
-      maybe_config_path: flags.config_path,
+      operation,
-    })?;
+      "Bundle",
-
+    )
-  if flags.no_check && maybe_ignored_options.is_some() {
+    .await?;
-    let ignored_options = maybe_ignored_options.unwrap();
-    eprintln!("{}", ignored_options);
-  }
-  debug!("{}", stats);
-
-  debug!(">>>>> bundle END");
-
-  if let Some(out_file_) = out_file.as_ref() {
-    let output_bytes = output.as_bytes();
-    let output_len = output_bytes.len();
-    fs_util::write_file(out_file_, output_bytes, 0o644)?;
-    info!(
-      "{} {:?} ({})",
-      colors::green("Emit"),
-      out_file_,
-      colors::gray(&info::human_size(output_len as f64))
-    );
   } else {
-    println!("{}", output);
+    let (_, module_graph) = module_resolver().await?;
+    operation(module_graph).await?;
   }
+
   Ok(())
 }
 
@@ -504,6 +550,20 @@ async fn doc_command(
   }
 }
 
+async fn format_command(
+  flags: Flags,
+  args: Vec<PathBuf>,
+  ignore: Vec<PathBuf>,
+  check: bool,
+) -> Result<(), AnyError> {
+  if args.len() == 1 && args[0].to_string_lossy() == "-" {
+    return tools::fmt::format_stdin(check);
+  }
+
+  tools::fmt::format(args, ignore, check, flags.watch).await?;
+  Ok(())
+}
+
 async fn run_repl(flags: Flags) -> Result<(), AnyError> {
   let main_module =
     ModuleSpecifier::resolve_url_or_path("./$deno$repl.ts").unwrap();
@@ -548,44 +608,49 @@ async fn run_from_stdin(flags: Flags) -> Result<(), AnyError> {
 }
 
 async fn run_with_watch(flags: Flags, script: String) -> Result<(), AnyError> {
-  let main_module = ModuleSpecifier::resolve_url_or_path(&script)?;
+  let module_resolver = || {
-  let program_state = ProgramState::new(flags.clone())?;
+    let script = script.clone();
-
+    let flags = flags.clone();
-  let handler = Rc::new(RefCell::new(FetchHandler::new(
-    &program_state,
-    Permissions::allow_all(),
-  )?));
-  let mut builder = module_graph::GraphBuilder::new(
-    handler,
-    program_state.maybe_import_map.clone(),
-    program_state.lockfile.clone(),
-  );
-  builder.add(&main_module, false).await?;
-  let module_graph = builder.get_graph();
-
-  // Find all local files in graph
-  let mut paths_to_watch: Vec<PathBuf> = module_graph
-    .get_modules()
-    .iter()
-    .filter(|specifier| specifier.as_url().scheme() == "file")
-    .map(|specifier| specifier.as_url().to_file_path().unwrap())
-    .collect();
-
-  if let Some(import_map) = program_state.flags.import_map_path.clone() {
-    paths_to_watch.push(
-      fs_util::resolve_from_cwd(std::path::Path::new(&import_map)).unwrap(),
-    );
-  }
-
-  // FIXME(bartlomieju): new file watcher is created on after each restart
-  file_watcher::watch_func(&paths_to_watch, move || {
-    // FIXME(bartlomieju): ProgramState must be created on each restart - otherwise file fetcher
-    // will use cached source files
-    let gs = ProgramState::new(flags.clone()).unwrap();
-    let permissions = Permissions::from_flags(&flags);
-    let main_module = main_module.clone();
     async move {
-      let mut worker = MainWorker::new(&gs, main_module.clone(), permissions);
+      let main_module = ModuleSpecifier::resolve_url_or_path(&script)?;
+      let program_state = ProgramState::new(flags)?;
+      let handler = Rc::new(RefCell::new(FetchHandler::new(
+        &program_state,
+        Permissions::allow_all(),
+      )?));
+      let mut builder = module_graph::GraphBuilder::new(
+        handler,
+        program_state.maybe_import_map.clone(),
+        program_state.lockfile.clone(),
+      );
+      builder.add(&main_module, false).await?;
+      let module_graph = builder.get_graph();
+
+      // Find all local files in graph
+      let mut paths_to_watch: Vec<PathBuf> = module_graph
+        .get_modules()
+        .iter()
+        .filter_map(|specifier| specifier.as_url().to_file_path().ok())
+        .collect();
+
+      if let Some(import_map) = program_state.flags.import_map_path.as_ref() {
+        paths_to_watch
+          .push(fs_util::resolve_from_cwd(std::path::Path::new(import_map))?);
+      }
+
+      Ok((paths_to_watch, main_module))
+    }
+    .boxed_local()
+  };
+
+  let operation = |main_module: ModuleSpecifier| {
+    let flags = flags.clone();
+    let permissions = Permissions::from_flags(&flags);
+    async move {
+      let main_module = main_module.clone();
+      let program_state = ProgramState::new(flags)?;
+      let mut worker =
+        MainWorker::new(&program_state, main_module.clone(), permissions);
       debug!("main_module {}", main_module);
       worker.execute_module(&main_module).await?;
       worker.execute("window.dispatchEvent(new Event('load'))")?;
@@ -594,7 +659,13 @@ async fn run_with_watch(flags: Flags, script: String) -> Result<(), AnyError> {
       Ok(())
     }
     .boxed_local()
-  })
+  };
+
+  file_watcher::watch_func_with_module_resolution(
+    module_resolver,
+    operation,
+    "Process",
+  )
   .await
 }
 
@@ -806,7 +877,7 @@ pub fn main() {
       check,
       files,
       ignore,
-    } => tools::fmt::format(files, check, ignore).boxed_local(),
+    } => format_command(flags, files, ignore, check).boxed_local(),
     DenoSubcommand::Info { file, json } => {
       info_command(flags, file, json).boxed_local()
     }

cli/tests/integration_tests.rs

@@ -474,6 +474,53 @@ fn fmt_test() {
   assert_eq!(expected, actual);
 }
 
+#[test]
+fn fmt_watch_test() {
+  let t = TempDir::new().expect("tempdir fail");
+  let fixed = util::root_path().join("cli/tests/badly_formatted_fixed.js");
+  let badly_formatted_original =
+    util::root_path().join("cli/tests/badly_formatted.mjs");
+  let badly_formatted = t.path().join("badly_formatted.js");
+  std::fs::copy(&badly_formatted_original, &badly_formatted)
+    .expect("Failed to copy file");
+
+  let mut child = util::deno_cmd()
+    .current_dir(util::root_path())
+    .arg("fmt")
+    .arg(&badly_formatted)
+    .arg("--watch")
+    .arg("--unstable")
+    .stdout(std::process::Stdio::piped())
+    .stderr(std::process::Stdio::piped())
+    .spawn()
+    .expect("Failed to spawn script");
+  let stderr = child.stderr.as_mut().unwrap();
+  let mut stderr_lines =
+    std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
+
+  // TODO(lucacasonato): remove this timeout. It seems to be needed on Linux.
+  std::thread::sleep(std::time::Duration::from_secs(1));
+
+  assert!(stderr_lines.next().unwrap().contains("badly_formatted.js"));
+
+  let expected = std::fs::read_to_string(fixed.clone()).unwrap();
+  let actual = std::fs::read_to_string(badly_formatted.clone()).unwrap();
+  assert_eq!(expected, actual);
+
+  // Change content of the file again to be badly formatted
+  std::fs::copy(&badly_formatted_original, &badly_formatted)
+    .expect("Failed to copy file");
+  std::thread::sleep(std::time::Duration::from_secs(1));
+
+  // Check if file has been automatically formatted by watcher
+  let expected = std::fs::read_to_string(fixed).unwrap();
+  let actual = std::fs::read_to_string(badly_formatted).unwrap();
+  assert_eq!(expected, actual);
+
+  child.kill().unwrap();
+  drop(t);
+}
+
 #[test]
 fn fmt_stdin_error() {
   use std::io::Write;
@@ -1142,6 +1189,103 @@ fn bundle_import_map_no_check() {
   assert_eq!(output.stderr, b"");
 }
 
+#[test]
+fn bundle_js_watch() {
+  use std::path::PathBuf;
+  // Test strategy extends this of test bundle_js by adding watcher
+  let t = TempDir::new().expect("tempdir fail");
+  let file_to_watch = t.path().join("file_to_watch.js");
+  std::fs::write(&file_to_watch, "console.log('Hello world');")
+    .expect("error writing file");
+  assert!(file_to_watch.is_file());
+  let t = TempDir::new().expect("tempdir fail");
+  let bundle = t.path().join("mod6.bundle.js");
+  let mut deno = util::deno_cmd()
+    .current_dir(util::root_path())
+    .arg("bundle")
+    .arg(&file_to_watch)
+    .arg(&bundle)
+    .arg("--watch")
+    .arg("--unstable")
+    .stdout(std::process::Stdio::piped())
+    .stderr(std::process::Stdio::piped())
+    .spawn()
+    .expect("failed to spawn script");
+
+  let stderr = deno.stderr.as_mut().unwrap();
+  let mut stderr_lines =
+    std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
+
+  std::thread::sleep(std::time::Duration::from_secs(1));
+  assert!(stderr_lines.next().unwrap().contains("file_to_watch.js"));
+  assert!(stderr_lines.next().unwrap().contains("mod6.bundle.js"));
+  let file = PathBuf::from(&bundle);
+  assert!(file.is_file());
+  assert!(stderr_lines.next().unwrap().contains("Bundle finished!"));
+
+  std::fs::write(&file_to_watch, "console.log('Hello world2');")
+    .expect("error writing file");
+  std::thread::sleep(std::time::Duration::from_secs(1));
+  assert!(stderr_lines
+    .next()
+    .unwrap()
+    .contains("File change detected!"));
+  assert!(stderr_lines.next().unwrap().contains("file_to_watch.js"));
+  assert!(stderr_lines.next().unwrap().contains("mod6.bundle.js"));
+  let file = PathBuf::from(&bundle);
+  assert!(file.is_file());
+  assert!(stderr_lines.next().unwrap().contains("Bundle finished!"));
+
+  // Confirm that the watcher keeps on working even if the file is updated and has invalid syntax
+  std::fs::write(&file_to_watch, "syntax error ^^")
+    .expect("error writing file");
+  std::thread::sleep(std::time::Duration::from_secs(1));
+  assert!(stderr_lines
+    .next()
+    .unwrap()
+    .contains("File change detected!"));
+  assert!(stderr_lines.next().unwrap().contains("file_to_watch.js"));
+  assert!(stderr_lines.next().unwrap().contains("mod6.bundle.js"));
+  let file = PathBuf::from(&bundle);
+  assert!(file.is_file());
+  assert!(stderr_lines.next().unwrap().contains("Bundle finished!"));
+
+  deno.kill().unwrap();
+  drop(t);
+}
+
+/// Confirm that the watcher exits immediately if module resolution fails at the first attempt
+#[test]
+fn bundle_watch_fail() {
+  let t = TempDir::new().expect("tempdir fail");
+  let file_to_watch = t.path().join("file_to_watch.js");
+  std::fs::write(&file_to_watch, "syntax error ^^")
+    .expect("error writing file");
+
+  let mut deno = util::deno_cmd()
+    .current_dir(util::root_path())
+    .arg("bundle")
+    .arg(&file_to_watch)
+    .arg("--watch")
+    .arg("--unstable")
+    .env("NO_COLOR", "1")
+    .stdout(std::process::Stdio::piped())
+    .stderr(std::process::Stdio::piped())
+    .spawn()
+    .expect("failed to spawn script");
+
+  let stderr = deno.stderr.as_mut().unwrap();
+  let mut stderr_lines =
+    std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
+
+  std::thread::sleep(std::time::Duration::from_secs(1));
+  assert!(stderr_lines.next().unwrap().contains("file_to_watch.js"));
+  assert!(stderr_lines.next().unwrap().contains("error:"));
+  assert!(!deno.wait().unwrap().success());
+
+  drop(t);
+}
+
 #[test]
 fn info_with_compiled_source() {
   let _g = util::http_server();
@@ -1201,7 +1345,7 @@ fn run_watch() {
     std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
 
   assert!(stdout_lines.next().unwrap().contains("Hello world"));
-  assert!(stderr_lines.next().unwrap().contains("Process terminated"));
+  assert!(stderr_lines.next().unwrap().contains("Process finished"));
 
   // TODO(lucacasonato): remove this timeout. It seems to be needed on Linux.
   std::thread::sleep(std::time::Duration::from_secs(1));
@@ -1209,18 +1353,89 @@ fn run_watch() {
   // Change content of the file
   std::fs::write(&file_to_watch, "console.log('Hello world2');")
     .expect("error writing file");
-
   // Events from the file watcher is "debounced", so we need to wait for the next execution to start
   std::thread::sleep(std::time::Duration::from_secs(1));
 
   assert!(stderr_lines.next().unwrap().contains("Restarting"));
   assert!(stdout_lines.next().unwrap().contains("Hello world2"));
-  assert!(stderr_lines.next().unwrap().contains("Process terminated"));
+  assert!(stderr_lines.next().unwrap().contains("Process finished"));
+
+  // Add dependency
+  let another_file = t.path().join("another_file.js");
+  std::fs::write(&another_file, "export const foo = 0;")
+    .expect("error writing file");
+  std::fs::write(
+    &file_to_watch,
+    "import { foo } from './another_file.js'; console.log(foo);",
+  )
+  .expect("error writing file");
+  std::thread::sleep(std::time::Duration::from_secs(1));
+  assert!(stderr_lines.next().unwrap().contains("Restarting"));
+  assert!(stdout_lines.next().unwrap().contains('0'));
+  assert!(stderr_lines.next().unwrap().contains("Process finished"));
+
+  // Confirm that restarting occurs when a new file is updated
+  std::fs::write(&another_file, "export const foo = 42;")
+    .expect("error writing file");
+  std::thread::sleep(std::time::Duration::from_secs(1));
+  assert!(stderr_lines.next().unwrap().contains("Restarting"));
+  assert!(stdout_lines.next().unwrap().contains("42"));
+  assert!(stderr_lines.next().unwrap().contains("Process finished"));
+
+  // Confirm that the watcher keeps on working even if the file is updated and has invalid syntax
+  std::fs::write(&file_to_watch, "syntax error ^^")
+    .expect("error writing file");
+  std::thread::sleep(std::time::Duration::from_secs(1));
+  assert!(stderr_lines.next().unwrap().contains("Restarting"));
+  assert!(stderr_lines.next().unwrap().contains("error:"));
+  assert!(stderr_lines.next().unwrap().contains("Process finished"));
+
+  // Then restore the file
+  std::fs::write(
+    &file_to_watch,
+    "import { foo } from './another_file.js'; console.log(foo);",
+  )
+  .expect("error writing file");
+  std::thread::sleep(std::time::Duration::from_secs(1));
+  assert!(stderr_lines.next().unwrap().contains("Restarting"));
+  assert!(stdout_lines.next().unwrap().contains("42"));
+  assert!(stderr_lines.next().unwrap().contains("Process finished"));
 
   child.kill().unwrap();
   drop(t);
 }
 
+/// Confirm that the watcher exits immediately if module resolution fails at the first attempt
+#[test]
+fn run_watch_fail() {
+  let t = TempDir::new().expect("tempdir fail");
+  let file_to_watch = t.path().join("file_to_watch.js");
+  std::fs::write(&file_to_watch, "syntax error ^^")
+    .expect("error writing file");
+
+  let mut child = util::deno_cmd()
+    .current_dir(util::root_path())
+    .arg("run")
+    .arg(&file_to_watch)
+    .arg("--watch")
+    .arg("--unstable")
+    .env("NO_COLOR", "1")
+    .stdout(std::process::Stdio::piped())
+    .stderr(std::process::Stdio::piped())
+    .spawn()
+    .expect("failed to spawn script");
+
+  let stderr = child.stderr.as_mut().unwrap();
+  let mut stderr_lines =
+    std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
+
+  std::thread::sleep(std::time::Duration::from_secs(1));
+  assert!(stderr_lines.next().unwrap().contains("error:"));
+  assert!(!child.wait().unwrap().success());
+
+  drop(t);
+}
+
 #[cfg(unix)]
 #[test]
 fn repl_test_pty_multiline() {
@@ -1355,7 +1570,7 @@ fn run_watch_with_importmap_and_relative_paths() {
   let mut stderr_lines =
     std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
 
-  assert!(stderr_lines.next().unwrap().contains("Process terminated"));
+  assert!(stderr_lines.next().unwrap().contains("Process finished"));
   assert!(stdout_lines.next().unwrap().contains("Hello world"));
 
   child.kill().unwrap();

cli/tools/fmt.rs

@@ -9,11 +9,13 @@
 
 use crate::colors;
 use crate::diff::diff;
+use crate::file_watcher;
 use crate::fs_util::{collect_files, is_supported_ext};
 use crate::text_encoding;
 use deno_core::error::generic_error;
 use deno_core::error::AnyError;
 use deno_core::futures;
+use deno_core::futures::FutureExt;
 use dprint_plugin_typescript as dprint;
 use std::fs;
 use std::io::stdin;
@@ -28,25 +30,37 @@ use std::sync::{Arc, Mutex};
 const BOM_CHAR: char = '\u{FEFF}';
 
 /// Format JavaScript/TypeScript files.
-///
-/// First argument and ignore supports globs, and if it is `None`
-/// then the current directory is recursively walked.
 pub async fn format(
   args: Vec<PathBuf>,
+  ignore: Vec<PathBuf>,
   check: bool,
-  exclude: Vec<PathBuf>,
+  watch: bool,
 ) -> Result<(), AnyError> {
-  if args.len() == 1 && args[0].to_string_lossy() == "-" {
+  let target_file_resolver = || {
-    return format_stdin(check);
+    // collect the files that are to be formatted
-  }
+    collect_files(&args, &ignore, is_supported_ext)
-  // collect the files that are to be formatted
+  };
-  let target_files = collect_files(args, exclude, is_supported_ext)?;
+
-  let config = get_config();
+  let operation = |paths: Vec<PathBuf>| {
-  if check {
+    let config = get_config();
-    check_source_files(config, target_files).await
+    async move {
+      if check {
+        check_source_files(config, paths).await?;
+      } else {
+        format_source_files(config, paths).await?;
+      }
+      Ok(())
+    }
+    .boxed_local()
+  };
+
+  if watch {
+    file_watcher::watch_func(target_file_resolver, operation, "Fmt").await?;
   } else {
-    format_source_files(config, target_files).await
+    operation(target_file_resolver()?).await?;
   }
+
+  Ok(())
 }
 
 async fn check_source_files(
@@ -166,7 +180,7 @@ async fn format_source_files(
 /// Format stdin and write result to stdout.
 /// Treats input as TypeScript.
 /// Compatible with `--check` flag.
-fn format_stdin(check: bool) -> Result<(), AnyError> {
+pub fn format_stdin(check: bool) -> Result<(), AnyError> {
   let mut source = String::new();
   if stdin().read_to_string(&mut source).is_err() {
     return Err(generic_error("Failed to read from stdin"));

cli/tools/lint.rs

@@ -47,7 +47,7 @@ pub async fn lint_files(
   if args.len() == 1 && args[0].to_string_lossy() == "-" {
     return lint_stdin(json);
   }
-  let target_files = collect_files(args, ignore, is_supported_ext)?;
+  let target_files = collect_files(&args, &ignore, is_supported_ext)?;
   debug!("Found {} files", target_files.len());
   let target_files_len = target_files.len();
 

cli/tools/test_runner.rs

@@ -44,8 +44,7 @@ pub fn prepare_test_modules_urls(
   for path in include_paths {
     let p = fs_util::normalize_path(&root_path.join(path));
     if p.is_dir() {
-      let test_files =
+      let test_files = fs_util::collect_files(&[p], &[], is_supported).unwrap();
-        crate::fs_util::collect_files(vec![p], vec![], is_supported).unwrap();
       let test_files_as_urls = test_files
         .iter()
         .map(|f| Url::from_file_path(f).unwrap())