1 files changed, 215 insertions, 56 deletions

diff --git a/crates/ra_vfs/src/io.rs b/crates/ra_vfs/src/io.rs
index 7ca1e9835..d764c534a 100644
--- a/crates/ra_vfs/src/io.rs
+++ b/crates/ra_vfs/src/io.rs
@@ -1,19 +1,22 @@
-use std::{fs, sync::Arc, thread};
-
-use crossbeam_channel::{Receiver, Sender};
+use std::{
+    fs,
+    thread,
+    path::{Path, PathBuf},
+    sync::{mpsc, Arc},
+    time::Duration,
+};
+use crossbeam_channel::{Receiver, Sender, unbounded, RecvError, select};
 use relative_path::RelativePathBuf;
 use thread_worker::WorkerHandle;
 use walkdir::WalkDir;
+use notify::{DebouncedEvent, RecommendedWatcher, RecursiveMode, Watcher as _Watcher};
 
-mod watcher;
-use watcher::Watcher;
-
-use crate::{RootFilter, Roots, VfsRoot};
+use crate::{RootConfig, Roots, VfsRoot};
 
 pub(crate) enum Task {
     AddRoot {
         root: VfsRoot,
-        filter: Arc<RootFilter>,
+        config: Arc<RootConfig>,
     },
 }
 
@@ -39,6 +42,15 @@ pub enum TaskResult {
     },
 }
 
+#[derive(Debug)]
+enum ChangeKind {
+    Create,
+    Write,
+    Remove,
+}
+
+const WATCHER_DELAY: Duration = Duration::from_millis(250);
+
 pub(crate) struct Worker {
     worker: thread_worker::Worker<Task, TaskResult>,
     worker_handle: WorkerHandle,
@@ -46,24 +58,75 @@ pub(crate) struct Worker {
 
 impl Worker {
     pub(crate) fn start(roots: Arc<Roots>) -> Worker {
-        let (worker, worker_handle) =
-            thread_worker::spawn("vfs", 128, move |input_receiver, output_sender| {
-                let mut watcher = match Watcher::start(roots, output_sender.clone()) {
-                    Ok(w) => Some(w),
-                    Err(e) => {
-                        log::error!("could not start watcher: {}", e);
-                        None
+        // This is a pretty elaborate setup of threads & channels! It is
+        // explained by the following concerns:
+        //    * we need to burn a thread translating from notify's mpsc to
+        //      crossbeam_channel.
+        //    * we want to read all files from a single thread, to gurantee that
+        //      we always get fresher versions and never go back in time.
+        //    * we want to tear down everything neatly during shutdown.
+        let (worker, worker_handle) = thread_worker::spawn(
+            "vfs",
+            128,
+            // This are the channels we use to communicate with outside world.
+            // If `input_receiver` is closed we need to tear ourselves down.
+            // `output_sender` should not be closed unless the parent died.
+            move |input_receiver, output_sender| {
+                // These are `std` channels notify will send events to
+                let (notify_sender, notify_receiver) = mpsc::channel();
+                // These are the corresponding crossbeam channels
+                let (watcher_sender, watcher_receiver) = unbounded();
+
+                let mut watcher = notify::watcher(notify_sender, WATCHER_DELAY)
+                    .map_err(|e| log::error!("failed to spawn notify {}", e))
+                    .ok();
+                // Start a silly thread to tranform between two channels
+                let thread = thread::spawn(move || {
+                    notify_receiver
+                        .into_iter()
+                        .for_each(|event| convert_notify_event(event, &watcher_sender))
+                });
+
+                // Process requests from the called or notifications from
+                // watcher until the caller says stop.
+                loop {
+                    select! {
+                        // Received request from the caller. If this channel is
+                        // closed, we should shutdown everything.
+                        recv(input_receiver) -> t => match t {
+                            Err(RecvError) => {
+                                drop(input_receiver);
+                                break
+                            },
+                            Ok(Task::AddRoot { root, config }) => {
+                                watch_root(watcher.as_mut(), &output_sender, root, Arc::clone(&config));
+                            }
+                        },
+                        // Watcher send us changes. If **this** channel is
+                        // closed, the watcher has died, which indicates a bug
+                        // -- escalate!
+                        recv(watcher_receiver) -> event => match event {
+                            Err(RecvError) => panic!("watcher is dead"),
+                            Ok((path, change)) => {
+                                handle_change(watcher.as_mut(), &output_sender, &*roots, path, change);
+                            }
+                        },
                     }
-                };
-                let res = input_receiver
-                    .into_iter()
-                    .filter_map(|t| handle_task(t, &mut watcher))
-                    .try_for_each(|it| output_sender.send(it));
-                if let Some(watcher) = watcher {
-                    let _ = watcher.shutdown();
                 }
-                res.unwrap()
-            });
+                // Stopped the watcher
+                drop(watcher.take());
+                // Drain pending events: we are not inrerested in them anyways!
+                watcher_receiver.into_iter().for_each(|_| ());
+
+                let res = thread.join();
+                match &res {
+                    Ok(()) => log::info!("... Watcher terminated with ok"),
+                    Err(_) => log::error!("... Watcher terminated with err"),
+                }
+                res.unwrap();
+            },
+        );
+
         Worker {
             worker,
             worker_handle,
@@ -84,46 +147,142 @@ impl Worker {
     }
 }
 
-fn handle_task(task: Task, watcher: &mut Option<Watcher>) -> Option<TaskResult> {
-    match task {
-        Task::AddRoot { root, filter } => {
-            if let Some(watcher) = watcher {
-                watcher.watch_root(&filter)
+fn watch_root(
+    watcher: Option<&mut RecommendedWatcher>,
+    sender: &Sender<TaskResult>,
+    root: VfsRoot,
+    config: Arc<RootConfig>,
+) {
+    log::debug!("loading {} ...", config.root.as_path().display());
+    let files = watch_recursive(watcher, config.root.as_path(), &*config)
+        .into_iter()
+        .filter_map(|path| {
+            let abs_path = path.to_path(&config.root);
+            let text = read_to_string(&abs_path)?;
+            Some((path, text))
+        })
+        .collect();
+    sender
+        .send(TaskResult::BulkLoadRoot { root, files })
+        .unwrap();
+    log::debug!("... loaded {}", config.root.as_path().display());
+}
+
+fn convert_notify_event(event: DebouncedEvent, sender: &Sender<(PathBuf, ChangeKind)>) {
+    // forward relevant events only
+    match event {
+        DebouncedEvent::NoticeWrite(_)
+        | DebouncedEvent::NoticeRemove(_)
+        | DebouncedEvent::Chmod(_) => {
+            // ignore
+        }
+        DebouncedEvent::Rescan => {
+            // TODO rescan all roots
+        }
+        DebouncedEvent::Create(path) => {
+            sender.send((path, ChangeKind::Create)).unwrap();
+        }
+        DebouncedEvent::Write(path) => {
+            sender.send((path, ChangeKind::Write)).unwrap();
+        }
+        DebouncedEvent::Remove(path) => {
+            sender.send((path, ChangeKind::Remove)).unwrap();
+        }
+        DebouncedEvent::Rename(src, dst) => {
+            sender.send((src, ChangeKind::Remove)).unwrap();
+            sender.send((dst, ChangeKind::Create)).unwrap();
+        }
+        DebouncedEvent::Error(err, path) => {
+            // TODO should we reload the file contents?
+            log::warn!("watcher error \"{}\", {:?}", err, path);
+        }
+    }
+}
+
+fn handle_change(
+    watcher: Option<&mut RecommendedWatcher>,
+    sender: &Sender<TaskResult>,
+    roots: &Roots,
+    path: PathBuf,
+    kind: ChangeKind,
+) {
+    let (root, rel_path) = match roots.find(&path) {
+        None => return,
+        Some(it) => it,
+    };
+    let config = &roots[root];
+    match kind {
+        ChangeKind::Create => {
+            let mut paths = Vec::new();
+            if path.is_dir() {
+                paths.extend(watch_recursive(watcher, &path, &config));
+            } else {
+                paths.push(rel_path);
             }
-            log::debug!("loading {} ...", filter.root.as_path().display());
-            let files = load_root(filter.as_ref());
-            log::debug!("... loaded {}", filter.root.as_path().display());
-            Some(TaskResult::BulkLoadRoot { root, files })
+            paths
+                .into_iter()
+                .filter_map(|rel_path| {
+                    let abs_path = rel_path.to_path(&config.root);
+                    let text = read_to_string(&abs_path)?;
+                    Some((rel_path, text))
+                })
+                .try_for_each(|(path, text)| {
+                    sender.send(TaskResult::AddSingleFile { root, path, text })
+                })
+                .unwrap()
         }
+        ChangeKind::Write => {
+            if let Some(text) = read_to_string(&path) {
+                sender
+                    .send(TaskResult::ChangeSingleFile {
+                        root,
+                        path: rel_path,
+                        text,
+                    })
+                    .unwrap();
+            }
+        }
+        ChangeKind::Remove => sender
+            .send(TaskResult::RemoveSingleFile {
+                root,
+                path: rel_path,
+            })
+            .unwrap(),
     }
 }
 
-fn load_root(filter: &RootFilter) -> Vec<(RelativePathBuf, String)> {
-    let mut res = Vec::new();
-    for entry in WalkDir::new(&filter.root)
+fn watch_recursive(
+    mut watcher: Option<&mut RecommendedWatcher>,
+    dir: &Path,
+    config: &RootConfig,
+) -> Vec<RelativePathBuf> {
+    let mut files = Vec::new();
+    for entry in WalkDir::new(dir)
         .into_iter()
-        .filter_entry(filter.entry_filter())
+        .filter_entry(|it| config.contains(it.path()).is_some())
+        .filter_map(|it| it.map_err(|e| log::warn!("watcher error: {}", e)).ok())
     {
-        let entry = match entry {
-            Ok(entry) => entry,
-            Err(e) => {
-                log::warn!("watcher error: {}", e);
-                continue;
+        if entry.file_type().is_dir() {
+            if let Some(watcher) = &mut watcher {
+                watch_one(watcher, entry.path());
             }
-        };
-        if !entry.file_type().is_file() {
-            continue;
+        } else {
+            let path = config.contains(entry.path()).unwrap();
+            files.push(path.to_owned());
         }
-        let path = entry.path();
-        let text = match fs::read_to_string(path) {
-            Ok(text) => text,
-            Err(e) => {
-                log::warn!("watcher error: {}", e);
-                continue;
-            }
-        };
-        let path = RelativePathBuf::from_path(path.strip_prefix(&filter.root).unwrap()).unwrap();
-        res.push((path.to_owned(), text))
     }
-    res
+    files
+}
+
+fn watch_one(watcher: &mut RecommendedWatcher, dir: &Path) {
+    match watcher.watch(dir, RecursiveMode::NonRecursive) {
+        Ok(()) => log::debug!("watching \"{}\"", dir.display()),
+        Err(e) => log::warn!("could not watch \"{}\": {}", dir.display(), e),
+    }
+}
+
+fn read_to_string(path: &Path) -> Option<String> {
+    fs::read_to_string(&path)
+        .map_err(|e| log::warn!("failed to read file {}", e))
+        .ok()
 }