feat: fs events

cli/Cargo.toml

@@ -43,6 +43,7 @@ indexmap = "1.3.0"
 lazy_static = "1.4.0"
 libc = "0.2.66"
 log = "0.4.8"
+notify = { version = "5.0.0-pre.2" }
 rand = "0.7.2"
 regex = "1.3.1"
 remove_dir_all = "0.5.2"
@@ -53,7 +54,7 @@ serde = { version = "1.0.104", features = ["derive"] }
 serde_derive = "1.0.104"
 serde_json = { version = "1.0.44", features = [ "preserve_order" ] }
 source-map-mappings = "0.5.0"
-sys-info = "0.5.8"
+sys-info = "=0.5.8" # 0.5.9 seems to be broken on windows.
 tempfile = "3.1.0"
 termcolor = "1.0.5"
 tokio = { version = "0.2", features = ["rt-core", "tcp", "udp", "process", "fs", "blocking", "sync", "io-std", "macros", "time"] }

cli/js/deno.ts

@@ -43,6 +43,7 @@ export {
   OpenOptions,
   OpenMode
 } from "./files.ts";
+export { FsEvent, fsEvents } from "./fs_events.ts";
 export {
   EOF,
   copy,

cli/js/dispatch.ts

@@ -73,6 +73,8 @@ export let OP_CWD: number;
 export let OP_CONNECT_TLS: number;
 export let OP_HOSTNAME: number;
 export let OP_OPEN_PLUGIN: number;
+export let OP_FS_EVENTS_OPEN: number;
+export let OP_FS_EVENTS_POLL: number;
 export let OP_COMPILE: number;
 export let OP_TRANSPILE: number;
 export let OP_SIGNAL_BIND: number;

cli/js/fs_events.ts

@@ -0,0 +1,40 @@
+// Copyright 2019 the Deno authors. All rights reserved. MIT license.
+import { sendSync, sendAsync } from "./dispatch_json.ts";
+import * as dispatch from "./dispatch.ts";
+import { close } from "./files.ts";
+
+export interface FsEvent {
+  kind: "any" | "access" | "create" | "modify" | "remove";
+  paths: string[];
+}
+
+class FsEvents implements AsyncIterableIterator<FsEvent> {
+  readonly rid: number;
+
+  constructor(paths: string[], options: { recursive: boolean }) {
+    const { recursive } = options;
+    this.rid = sendSync(dispatch.OP_FS_EVENTS_OPEN, { recursive, paths });
+  }
+
+  async next(): Promise<IteratorResult<FsEvent>> {
+    return await sendAsync(dispatch.OP_FS_EVENTS_POLL, {
+      rid: this.rid
+    });
+  }
+
+  async return(value?: FsEvent): Promise<IteratorResult<FsEvent>> {
+    close(this.rid);
+    return { value, done: true };
+  }
+
+  [Symbol.asyncIterator](): AsyncIterableIterator<FsEvent> {
+    return this;
+  }
+}
+
+export function fsEvents(
+  paths: string | string[],
+  options = { recursive: true }
+): AsyncIterableIterator<FsEvent> {
+  return new FsEvents(Array.isArray(paths) ? paths : [paths], options);
+}

cli/js/fs_events_test.ts

@@ -0,0 +1,52 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+import { testPerm, assert } from "./test_util.ts";
+
+// TODO(ry) Add more tests to specify format.
+
+testPerm({ read: false }, function fsEventsPermissions() {
+  let thrown = false;
+  try {
+    Deno.fsEvents(".");
+  } catch (err) {
+    assert(err instanceof Deno.Err.PermissionDenied);
+    thrown = true;
+  }
+  assert(thrown);
+});
+
+async function getTwoEvents(
+  iter: AsyncIterableIterator<Deno.FsEvent>
+): Promise<Deno.FsEvent[]> {
+  const events = [];
+  for await (const event of iter) {
+    console.log(">>>> event", event);
+    events.push(event);
+    if (events.length > 2) break;
+  }
+  return events;
+}
+
+testPerm({ read: true, write: true }, async function fsEventsBasic(): Promise<
+  void
+> {
+  const testDir = await Deno.makeTempDir();
+  const iter = Deno.fsEvents(testDir);
+
+  // Asynchornously capture two fs events.
+  const eventsPromise = getTwoEvents(iter);
+
+  // Make some random file system activity.
+  const file1 = testDir + "/file1.txt";
+  const file2 = testDir + "/file2.txt";
+  Deno.writeFileSync(file1, new Uint8Array([0, 1, 2]));
+  Deno.writeFileSync(file2, new Uint8Array([0, 1, 2]));
+
+  // We should have gotten two fs events.
+  const events = await eventsPromise;
+  console.log("events", events);
+  assert(events.length >= 2);
+  assert(events[0].kind == "create");
+  assert(events[0].paths[0].includes(testDir));
+  assert(events[1].kind == "create" || events[1].kind == "modify");
+  assert(events[1].paths[0].includes(testDir));
+});

cli/js/lib.deno.ns.d.ts

@@ -1620,6 +1620,21 @@ declare namespace Deno {
    */
   export function resources(): ResourceMap;
 
+  /** UNSTABLE: new API. Needs docs. */
+  export interface FsEvent {
+    kind: "any" | "access" | "create" | "modify" | "remove";
+    paths: string[];
+  }
+
+  /** UNSTABLE: new API. Needs docs.
+   *
+   * recursive option is true by default.
+   */
+  export function fsEvents(
+    paths: string | string[],
+    options?: { recursive: boolean }
+  ): AsyncIterableIterator<FsEvent>;
+
   /** How to handle subprocess stdio.
    *
    * "inherit" The default if unspecified. The child inherits from the

cli/js/unit_tests.ts

@@ -23,6 +23,7 @@ import "./fetch_test.ts";
 import "./file_test.ts";
 import "./files_test.ts";
 import "./form_data_test.ts";
+import "./fs_events_test.ts";
 import "./get_random_values_test.ts";
 import "./globals_test.ts";
 import "./headers_test.ts";

cli/ops/fs_events.rs

@@ -0,0 +1,129 @@
+// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
+use super::dispatch_json::{Deserialize, JsonOp, Value};
+use crate::deno_error::bad_resource;
+use crate::ops::json_op;
+use crate::state::State;
+use deno_core::*;
+use futures::future::poll_fn;
+use futures::future::FutureExt;
+use notify::event::Event as NotifyEvent;
+use notify::Error as NotifyError;
+use notify::EventKind;
+use notify::RecommendedWatcher;
+use notify::RecursiveMode;
+use notify::Watcher;
+use serde::Serialize;
+use std::convert::From;
+use std::path::PathBuf;
+use tokio::sync::mpsc;
+
+pub fn init(i: &mut Isolate, s: &State) {
+  i.register_op(
+    "fs_events_open",
+    s.core_op(json_op(s.stateful_op(op_fs_events_open))),
+  );
+  i.register_op(
+    "fs_events_poll",
+    s.core_op(json_op(s.stateful_op(op_fs_events_poll))),
+  );
+}
+
+struct FsEventsResource {
+  #[allow(unused)]
+  watcher: RecommendedWatcher,
+  receiver: mpsc::Receiver<Result<FsEvent, ErrBox>>,
+}
+
+/// Represents a file system event.
+///
+/// We do not use the event directly from the notify crate. We flatten
+/// the structure into this simpler structure. We want to only make it more
+/// complex as needed.
+///
+/// Feel free to expand this struct as long as you can add tests to demonstrate
+/// the complexity.
+#[derive(Serialize, Debug)]
+struct FsEvent {
+  kind: String,
+  paths: Vec<PathBuf>,
+}
+
+impl From<NotifyEvent> for FsEvent {
+  fn from(e: NotifyEvent) -> Self {
+    let kind = match e.kind {
+      EventKind::Any => "any",
+      EventKind::Access(_) => "access",
+      EventKind::Create(_) => "create",
+      EventKind::Modify(_) => "modify",
+      EventKind::Remove(_) => "remove",
+      EventKind::Other => todo!(), // What's this for? Leaving it out for now.
+    }
+    .to_string();
+    FsEvent {
+      kind,
+      paths: e.paths,
+    }
+  }
+}
+
+pub fn op_fs_events_open(
+  state: &State,
+  args: Value,
+  _zero_copy: Option<ZeroCopyBuf>,
+) -> Result<JsonOp, ErrBox> {
+  #[derive(Deserialize)]
+  struct OpenArgs {
+    recursive: bool,
+    paths: Vec<String>,
+  }
+  let args: OpenArgs = serde_json::from_value(args)?;
+  let (sender, receiver) = mpsc::channel::<Result<FsEvent, ErrBox>>(16);
+  let sender = std::sync::Mutex::new(sender);
+  let mut watcher: RecommendedWatcher =
+    Watcher::new_immediate(move |res: Result<NotifyEvent, NotifyError>| {
+      let res2 = res.map(FsEvent::from).map_err(ErrBox::from);
+      let mut sender = sender.lock().unwrap();
+      futures::executor::block_on(sender.send(res2)).expect("fs events error");
+    })?;
+  let recursive_mode = if args.recursive {
+    RecursiveMode::Recursive
+  } else {
+    RecursiveMode::NonRecursive
+  };
+  for path in &args.paths {
+    state.check_read(&PathBuf::from(path))?;
+    watcher.watch(path, recursive_mode)?;
+  }
+  let resource = FsEventsResource { watcher, receiver };
+  let table = &mut state.borrow_mut().resource_table;
+  let rid = table.add("fsEvents", Box::new(resource));
+  Ok(JsonOp::Sync(json!(rid)))
+}
+
+pub fn op_fs_events_poll(
+  state: &State,
+  args: Value,
+  _zero_copy: Option<ZeroCopyBuf>,
+) -> Result<JsonOp, ErrBox> {
+  #[derive(Deserialize)]
+  struct PollArgs {
+    rid: u32,
+  }
+  let PollArgs { rid } = serde_json::from_value(args)?;
+  let state = state.clone();
+  let f = poll_fn(move |cx| {
+    let resource_table = &mut state.borrow_mut().resource_table;
+    let watcher = resource_table
+      .get_mut::<FsEventsResource>(rid)
+      .ok_or_else(bad_resource)?;
+    watcher
+      .receiver
+      .poll_recv(cx)
+      .map(|maybe_result| match maybe_result {
+        Some(Ok(value)) => Ok(json!({ "value": value, "done": false })),
+        Some(Err(err)) => Err(err),
+        None => Ok(json!({ "done": true })),
+      })
+  });
+  Ok(JsonOp::Async(f.boxed_local()))
+}

cli/ops/mod.rs

@@ -13,6 +13,7 @@ pub mod errors;
 pub mod fetch;
 pub mod files;
 pub mod fs;
+pub mod fs_events;
 pub mod io;
 pub mod net;
 pub mod os;

cli/worker.rs

@@ -206,6 +206,7 @@ impl MainWorker {
       ops::fetch::init(isolate, &state);
       ops::files::init(isolate, &state);
       ops::fs::init(isolate, &state);
+      ops::fs_events::init(isolate, &state);
       ops::io::init(isolate, &state);
       ops::plugins::init(isolate, &state, op_registry);
       ops::net::init(isolate, &state);

std/manual.md

@@ -467,6 +467,23 @@ for await (const _ of sig) {
 
 The above for-await loop exits after 5 seconds when sig.dispose() is called.
 
+### File system events
+
+To poll for file system events:
+
+```ts
+const iter = Deno.fsEvents("/");
+for await (const event of iter) {
+  console.log(">>>> event", event);
+  // { kind: "create", paths: [ "/foo.txt" ] }
+}
+```
+
+Note that the exact ordering of the events can vary between operating systems.
+This feature uses different syscalls depending on the platform:
+
+Linux: inotify macOS: FSEvents Windows: ReadDirectoryChangesW
+
 ### Linking to third party code
 
 In the above examples, we saw that Deno could execute scripts from URLs. Like