Fs watcher (#113)

fs watcher

src/system/watcher/change.rs

@@ -0,0 +1,312 @@
+//! ## File system change
+//!
+//! this module exposes the types to describe a change to sync on the remote file system
+
+use crate::utils::path as path_utils;
+
+use std::path::{Path, PathBuf};
+
+/// Describes an operation on the remote file system to sync
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum FsChange {
+    /// Move file on remote
+    Move(FileToRename),
+    /// Remove file from remote
+    Remove(FileToRemove),
+    /// Upload file to remote
+    Update(FileUpdate),
+}
+
+impl FsChange {
+    /// Instantiate a new `FsChange::Move`
+    pub fn mov(
+        source: PathBuf,
+        destination: PathBuf,
+        local_watched_path: &Path,
+        remote_synched_path: &Path,
+    ) -> Self {
+        Self::Move(FileToRename::new(
+            source,
+            destination,
+            local_watched_path,
+            remote_synched_path,
+        ))
+    }
+
+    /// Instantiate a new `FsChange::Remove`
+    pub fn remove(
+        removed_path: PathBuf,
+        local_watched_path: &Path,
+        remote_synched_path: &Path,
+    ) -> Self {
+        Self::Remove(FileToRemove::new(
+            removed_path,
+            local_watched_path,
+            remote_synched_path,
+        ))
+    }
+
+    /// Instantiate a new `FsChange::Update`
+    pub fn update(
+        changed_path: PathBuf,
+        local_watched_path: &Path,
+        remote_synched_path: &Path,
+    ) -> Self {
+        Self::Update(FileUpdate::new(
+            changed_path,
+            local_watched_path,
+            remote_synched_path,
+        ))
+    }
+}
+
+/// Describes a file to rename on the remote fs
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct FileToRename {
+    /// Path to file which has to be renamed
+    source: PathBuf,
+    /// new filename
+    destination: PathBuf,
+}
+
+impl FileToRename {
+    /// Instantiate a new `FileToRename` given
+    ///
+    /// - the path of the source on local fs
+    /// - the path of the destination on local fs
+    /// - the path of the file/directory watched on the local fs
+    /// - the path of the remote file/directory synched with the local fs
+    ///
+    /// the `remote` is resolved pushing to `remote_synched_path` the diff between `changed_path` and `local_watched_path`
+    fn new(
+        source: PathBuf,
+        destination: PathBuf,
+        local_watched_path: &Path,
+        remote_synched_path: &Path,
+    ) -> Self {
+        Self {
+            source: remote_relative_path(&source, local_watched_path, remote_synched_path),
+            destination: remote_relative_path(
+                &destination,
+                local_watched_path,
+                remote_synched_path,
+            ),
+        }
+    }
+
+    /// Get path to the source to rename
+    pub fn source(&self) -> &Path {
+        self.source.as_path()
+    }
+
+    /// Get path to the destination name
+    pub fn destination(&self) -> &Path {
+        self.destination.as_path()
+    }
+}
+
+/// Describes a file to remove on remote fs
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct FileToRemove {
+    /// Path to the file which has to be removed
+    path: PathBuf,
+}
+
+impl FileToRemove {
+    /// Instantiate a new `FileToRemove` given
+    ///
+    /// - the path of the file which has been removed on localhost
+    /// - the path of the file/directory watched on the local fs
+    /// - the path of the remote file/directory synched with the local fs
+    ///
+    /// the `remote` is resolved pushing to `remote_synched_path` the diff between `removed_path` and `local_watched_path`
+    fn new(removed_path: PathBuf, local_watched_path: &Path, remote_synched_path: &Path) -> Self {
+        Self {
+            path: remote_relative_path(&removed_path, local_watched_path, remote_synched_path),
+        }
+    }
+
+    /// Get path to the file to unlink
+    pub fn path(&self) -> &Path {
+        self.path.as_path()
+    }
+}
+
+/// Describes a file changed to sync
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct FileUpdate {
+    /// Path to file which has changed
+    local: PathBuf,
+    /// Path to remote file to update
+    remote: PathBuf,
+}
+
+impl FileUpdate {
+    /// Instantiate a new `FileUpdate` given
+    ///
+    /// - the path of the file which has changed
+    /// - the path of the file/directory watched on the local fs
+    /// - the path of the remote file/directory synched with the local fs
+    ///
+    /// the `remote` is resolved pushing to `remote_synched_path` the diff between `changed_path` and `local_watched_path`
+    fn new(changed_path: PathBuf, local_watched_path: &Path, remote_synched_path: &Path) -> Self {
+        Self {
+            remote: remote_relative_path(&changed_path, local_watched_path, remote_synched_path),
+            local: changed_path,
+        }
+    }
+
+    /// Get path to local file to sync
+    pub fn local(&self) -> &Path {
+        self.local.as_path()
+    }
+
+    /// Get path to remote file to sync
+    pub fn remote(&self) -> &Path {
+        self.remote.as_path()
+    }
+}
+
+// -- utils
+
+/// Get remote relative path, given the local target, the path of the local watched path and the path of the remote synched directory/file
+fn remote_relative_path(
+    target: &Path,
+    local_watched_path: &Path,
+    remote_synched_path: &Path,
+) -> PathBuf {
+    let local_diff = path_utils::diff_paths(target, local_watched_path);
+    // get absolute path to remote file associated to local file
+    match local_diff {
+        None => remote_synched_path.to_path_buf(),
+        Some(p) => {
+            let mut remote = remote_synched_path.to_path_buf();
+            remote.push(p);
+            remote
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+
+    use super::*;
+
+    use pretty_assertions::assert_eq;
+
+    #[test]
+    fn should_get_remote_relative_path_from_subdir() {
+        assert_eq!(
+            remote_relative_path(
+                Path::new("/tmp/abc/test.txt"),
+                Path::new("/tmp"),
+                Path::new("/home/foo")
+            )
+            .as_path(),
+            Path::new("/home/foo/abc/test.txt")
+        );
+    }
+
+    #[test]
+    fn should_get_remote_relative_path_same_path() {
+        assert_eq!(
+            remote_relative_path(
+                Path::new("/tmp/abc/test.txt"),
+                Path::new("/tmp/abc/test.txt"),
+                Path::new("/home/foo/test.txt")
+            )
+            .as_path(),
+            Path::new("/home/foo/test.txt")
+        );
+    }
+
+    #[test]
+    fn should_make_fs_change_move_from_same_directory() {
+        let change = FsChange::mov(
+            PathBuf::from("/tmp/foo.txt"),
+            PathBuf::from("/tmp/bar.txt"),
+            Path::new("/tmp"),
+            Path::new("/home/foo"),
+        );
+        if let FsChange::Move(change) = change {
+            assert_eq!(change.source(), Path::new("/home/foo/foo.txt"));
+            assert_eq!(change.destination(), Path::new("/home/foo/bar.txt"));
+        } else {
+            panic!("not a Move");
+        }
+    }
+
+    #[test]
+    fn should_make_fs_change_move_from_subdirectory() {
+        let change = FsChange::mov(
+            PathBuf::from("/tmp/abc/foo.txt"),
+            PathBuf::from("/tmp/abc/bar.txt"),
+            Path::new("/tmp/abc"),
+            Path::new("/home/foo"),
+        );
+        if let FsChange::Move(change) = change {
+            assert_eq!(change.source(), Path::new("/home/foo/foo.txt"));
+            assert_eq!(change.destination(), Path::new("/home/foo/bar.txt"));
+        } else {
+            panic!("not a Move");
+        }
+    }
+
+    #[test]
+    fn should_make_fs_change_remove_from_same_directory() {
+        let change = FsChange::remove(
+            PathBuf::from("/tmp/bar.txt"),
+            Path::new("/tmp/bar.txt"),
+            Path::new("/home/foo/bar.txt"),
+        );
+        if let FsChange::Remove(change) = change {
+            assert_eq!(change.path(), Path::new("/home/foo/bar.txt"));
+        } else {
+            panic!("not a remove");
+        }
+    }
+
+    #[test]
+    fn should_make_fs_change_remove_from_subdirectory() {
+        let change = FsChange::remove(
+            PathBuf::from("/tmp/abc/bar.txt"),
+            Path::new("/tmp/abc"),
+            Path::new("/home/foo"),
+        );
+        if let FsChange::Remove(change) = change {
+            assert_eq!(change.path(), Path::new("/home/foo/bar.txt"));
+        } else {
+            panic!("not a remove");
+        }
+    }
+
+    #[test]
+    fn should_make_fs_change_update_from_same_directory() {
+        let change = FsChange::update(
+            PathBuf::from("/tmp/bar.txt"),
+            Path::new("/tmp/bar.txt"),
+            Path::new("/home/foo/bar.txt"),
+        );
+        if let FsChange::Update(change) = change {
+            assert_eq!(change.local(), Path::new("/tmp/bar.txt"),);
+            assert_eq!(change.remote(), Path::new("/home/foo/bar.txt"));
+        } else {
+            panic!("not an update");
+        }
+    }
+
+    #[test]
+    fn should_make_fs_change_update_from_subdirectory() {
+        let change = FsChange::update(
+            PathBuf::from("/tmp/abc/foo.txt"),
+            Path::new("/tmp"),
+            Path::new("/home/foo/temp"),
+        );
+        if let FsChange::Update(change) = change {
+            assert_eq!(change.local(), Path::new("/tmp/abc/foo.txt"),);
+            assert_eq!(change.remote(), Path::new("/home/foo/temp/abc/foo.txt"));
+        } else {
+            panic!("not an update");
+        }
+    }
+}

src/system/watcher/mod.rs

@@ -0,0 +1,390 @@
+//! ## File system watcher
+//!
+//! A watcher for file system paths, which reports changes on local fs
+
+mod change;
+
+// -- export
+pub use change::FsChange;
+
+use crate::utils::path as path_utils;
+
+use notify::{
+    watcher, DebouncedEvent, Error as WatcherError, RecommendedWatcher, RecursiveMode, Watcher,
+};
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::mpsc::{channel, Receiver, RecvTimeoutError};
+use std::time::Duration;
+use thiserror::Error;
+
+type FsWatcherResult<T> = Result<T, FsWatcherError>;
+
+/// Describes an error returned by the `FsWatcher`
+#[derive(Debug, Error)]
+pub enum FsWatcherError {
+    #[error("unable to unwatch this path, since is not currently watched")]
+    PathNotWatched,
+    #[error("unable to watch path, since it's already watched")]
+    PathAlreadyWatched,
+    #[error("worker error: {0}")]
+    WorkerError(WatcherError),
+}
+
+impl From<WatcherError> for FsWatcherError {
+    fn from(err: WatcherError) -> Self {
+        Self::WorkerError(err)
+    }
+}
+
+/// File system watcher
+pub struct FsWatcher {
+    paths: HashMap<PathBuf, PathBuf>,
+    receiver: Receiver<DebouncedEvent>,
+    watcher: RecommendedWatcher,
+}
+
+impl FsWatcher {
+    /// Initialize a new `FsWatcher`
+    pub fn init(delay: Duration) -> FsWatcherResult<Self> {
+        let (tx, receiver) = channel();
+
+        Ok(Self {
+            paths: HashMap::default(),
+            receiver,
+            watcher: watcher(tx, delay)?,
+        })
+    }
+
+    /// Poll searching for the first available disk change
+    pub fn poll(&self) -> FsWatcherResult<Option<FsChange>> {
+        match self.receiver.recv_timeout(Duration::from_millis(1)) {
+            Ok(DebouncedEvent::Rename(source, dest)) => Ok(self.build_fs_move(source, dest)),
+            Ok(DebouncedEvent::Remove(p)) => Ok(self.build_fs_remove(p)),
+            Ok(DebouncedEvent::Chmod(p) | DebouncedEvent::Create(p) | DebouncedEvent::Write(p)) => {
+                Ok(self.build_fs_update(p))
+            }
+            Ok(
+                DebouncedEvent::Rescan
+                | DebouncedEvent::NoticeRemove(_)
+                | DebouncedEvent::NoticeWrite(_),
+            ) => Ok(None),
+            Ok(DebouncedEvent::Error(e, _)) => {
+                error!("FsWatcher reported error: {}", e);
+                Err(e.into())
+            }
+            Err(RecvTimeoutError::Timeout) => Ok(None),
+            Err(RecvTimeoutError::Disconnected) => panic!("File watcher died"),
+        }
+    }
+
+    /// Watch `local` path on localhost
+    pub fn watch(&mut self, local: &Path, remote: &Path) -> FsWatcherResult<()> {
+        // Start watcher if unwatched
+        if !self.watched(local) {
+            self.watcher.watch(local, RecursiveMode::Recursive)?;
+            // Insert new path to paths
+            self.paths.insert(local.to_path_buf(), remote.to_path_buf());
+            Ok(())
+        } else {
+            Err(FsWatcherError::PathAlreadyWatched)
+        }
+    }
+
+    /// Returns whether `path` is currently watched.
+    /// This method looks also in path ancestors.
+    ///
+    /// Example:
+    /// if `/home` is watched, then if we call `watched("/home/foo/file.txt")` will return `true`
+    pub fn watched(&self, path: &Path) -> bool {
+        self.find_watched_path(path).is_some()
+    }
+
+    /// Returns the list of watched paths
+    pub fn watched_paths(&self) -> Vec<&Path> {
+        Vec::from_iter(self.paths.keys().map(|x| x.as_path()))
+    }
+
+    /// Unwatch provided path.
+    /// When unwatching the path, it searches for the ancestor watched path if any.
+    /// Returns the unwatched resolved path
+    pub fn unwatch(&mut self, path: &Path) -> FsWatcherResult<PathBuf> {
+        let watched_path = self.find_watched_path(path).map(|x| x.0.to_path_buf());
+        if let Some(watched_path) = watched_path {
+            self.watcher.unwatch(watched_path.as_path())?;
+            self.paths.remove(watched_path.as_path());
+            Ok(watched_path)
+        } else {
+            Err(FsWatcherError::PathNotWatched)
+        }
+    }
+
+    /// Given a certain path, returns the path data associated to the path which
+    /// is ancestor of that path in the current watched path
+    fn find_watched_path(&self, p: &Path) -> Option<(&Path, &Path)> {
+        self.paths
+            .iter()
+            .find(|(k, _)| path_utils::is_child_of(p, k))
+            .map(|(k, v)| (k.as_path(), v.as_path()))
+    }
+
+    /// Build `FsChange` from path to local `changed_file`
+    fn build_fs_move(&self, source: PathBuf, destination: PathBuf) -> Option<FsChange> {
+        if let Some((watched_local, watched_remote)) = self.find_watched_path(&source) {
+            Some(FsChange::mov(
+                source,
+                destination,
+                watched_local,
+                watched_remote,
+            ))
+        } else {
+            None
+        }
+    }
+
+    /// Build `FsChange` from path to local `changed_file`
+    fn build_fs_remove(&self, removed_path: PathBuf) -> Option<FsChange> {
+        if let Some((watched_local, watched_remote)) = self.find_watched_path(&removed_path) {
+            Some(FsChange::remove(
+                removed_path,
+                watched_local,
+                watched_remote,
+            ))
+        } else {
+            None
+        }
+    }
+
+    /// Build `FsChange` from path to local `changed_file`
+    fn build_fs_update(&self, changed_file: PathBuf) -> Option<FsChange> {
+        if let Some((watched_local, watched_remote)) = self.find_watched_path(&changed_file) {
+            Some(FsChange::update(
+                changed_file,
+                watched_local,
+                watched_remote,
+            ))
+        } else {
+            None
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+
+    use super::*;
+    use crate::utils::test_helpers;
+
+    use pretty_assertions::assert_eq;
+    use tempfile::TempDir;
+
+    #[test]
+    fn should_init_fswatcher() {
+        let watcher = FsWatcher::init(Duration::from_secs(5)).unwrap();
+        assert!(watcher.paths.is_empty());
+    }
+
+    #[test]
+    fn should_watch_path() {
+        let mut watcher = FsWatcher::init(Duration::from_secs(5)).unwrap();
+        let tempdir = TempDir::new().unwrap();
+        assert!(watcher
+            .watch(tempdir.path(), Path::new("/tmp/test"))
+            .is_ok());
+        // check if in paths
+        assert_eq!(
+            watcher.paths.get(tempdir.path()).unwrap(),
+            Path::new("/tmp/test")
+        );
+        // close tempdir
+        assert!(tempdir.close().is_ok());
+    }
+
+    #[test]
+    fn should_not_watch_path_if_subdir_of_watched_path() {
+        let mut watcher = FsWatcher::init(Duration::from_secs(5)).unwrap();
+        let tempdir = TempDir::new().unwrap();
+        assert!(watcher
+            .watch(tempdir.path(), Path::new("/tmp/test"))
+            .is_ok());
+        // watch subdir
+        let mut subdir = tempdir.path().to_path_buf();
+        subdir.push("abc/def");
+        // should return already watched
+        assert!(watcher
+            .watch(subdir.as_path(), Path::new("/tmp/test/abc/def"))
+            .is_err());
+        // close tempdir
+        assert!(tempdir.close().is_ok());
+    }
+
+    #[test]
+    fn should_unwatch_path() {
+        let mut watcher = FsWatcher::init(Duration::from_secs(5)).unwrap();
+        let tempdir = TempDir::new().unwrap();
+        assert!(watcher
+            .watch(tempdir.path(), Path::new("/tmp/test"))
+            .is_ok());
+        // unwatch
+        assert!(watcher.unwatch(tempdir.path()).is_ok());
+        assert!(watcher.paths.get(tempdir.path()).is_none());
+        // close tempdir
+        assert!(tempdir.close().is_ok());
+    }
+
+    #[test]
+    fn should_unwatch_path_when_subdir() {
+        let mut watcher = FsWatcher::init(Duration::from_secs(5)).unwrap();
+        let tempdir = TempDir::new().unwrap();
+        assert!(watcher
+            .watch(tempdir.path(), Path::new("/tmp/test"))
+            .is_ok());
+        // unwatch
+        let mut subdir = tempdir.path().to_path_buf();
+        subdir.push("abc/def");
+        assert_eq!(
+            watcher.unwatch(subdir.as_path()).unwrap().as_path(),
+            Path::new(tempdir.path())
+        );
+        assert!(watcher.paths.get(tempdir.path()).is_none());
+        // close tempdir
+        assert!(tempdir.close().is_ok());
+    }
+
+    #[test]
+    fn should_return_err_when_unwatching_unwatched_path() {
+        let mut watcher = FsWatcher::init(Duration::from_secs(5)).unwrap();
+        assert!(watcher.unwatch(Path::new("/tmp")).is_err());
+    }
+
+    #[test]
+    fn should_tell_whether_path_is_watched() {
+        let mut watcher = FsWatcher::init(Duration::from_secs(5)).unwrap();
+        let tempdir = TempDir::new().unwrap();
+        assert!(watcher
+            .watch(tempdir.path(), Path::new("/tmp/test"))
+            .is_ok());
+        assert_eq!(watcher.watched(tempdir.path()), true);
+        let mut subdir = tempdir.path().to_path_buf();
+        subdir.push("abc/def");
+        assert_eq!(watcher.watched(subdir.as_path()), true);
+        assert_eq!(watcher.watched(Path::new("/tmp")), false);
+        // close tempdir
+        assert!(tempdir.close().is_ok());
+    }
+
+    #[test]
+    #[cfg(target_os = "macos")]
+    fn should_poll_file_update() {
+        let mut watcher = FsWatcher::init(Duration::from_millis(100)).unwrap();
+        let tempdir = TempDir::new().unwrap();
+        let tempdir_path = PathBuf::from(format!("/private{}", tempdir.path().display()));
+        assert!(watcher
+            .watch(tempdir_path.as_path(), Path::new("/tmp/test"))
+            .is_ok());
+        // create file
+        let file_path = test_helpers::make_file_at(tempdir_path.as_path(), "test.txt").unwrap();
+        // wait
+        std::thread::sleep(Duration::from_millis(500));
+        // wait till update
+        loop {
+            let fs_change = watcher.poll().unwrap();
+            if let Some(FsChange::Update(_)) = fs_change {
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(500));
+        }
+        assert!(std::fs::remove_file(file_path.as_path()).is_ok());
+        // close tempdir
+        assert!(tempdir.close().is_ok());
+    }
+
+    #[test]
+    #[cfg(target_os = "macos")]
+    fn should_poll_file_removed() {
+        let mut watcher = FsWatcher::init(Duration::from_millis(100)).unwrap();
+        let tempdir = TempDir::new().unwrap();
+        let tempdir_path = PathBuf::from(format!("/private{}", tempdir.path().display()));
+        assert!(watcher
+            .watch(tempdir_path.as_path(), Path::new("/tmp/test"))
+            .is_ok());
+        // create file
+        let file_path = test_helpers::make_file_at(tempdir_path.as_path(), "test.txt").unwrap();
+        std::thread::sleep(Duration::from_millis(500));
+        // wait
+        assert!(std::fs::remove_file(file_path.as_path()).is_ok());
+        // poll till remove
+        loop {
+            let fs_change = watcher.poll().unwrap();
+            if let Some(FsChange::Remove(remove)) = fs_change {
+                assert_eq!(remove.path(), Path::new("/tmp/test/test.txt"));
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(500));
+        }
+        // close tempdir
+        assert!(tempdir.close().is_ok());
+    }
+
+    /*
+    #[test]
+    #[cfg(target_family = "unix")]
+    fn should_poll_file_moved() {
+        let mut watcher = FsWatcher::init(Duration::from_millis(100)).unwrap();
+        let tempdir = TempDir::new().unwrap();
+        let tempdir_path = PathBuf::from(format!("/private{}", tempdir.path().display()));
+        assert!(watcher
+            .watch(tempdir_path.as_path(), Path::new("/tmp/test"))
+            .is_ok());
+        // create file
+        let file_path = test_helpers::make_file_at(tempdir_path.as_path(), "test.txt").unwrap();
+        // wait
+        std::thread::sleep(Duration::from_millis(500));
+        // move file
+        let mut new_file_path = tempdir.path().to_path_buf();
+        new_file_path.push("new.txt");
+        assert!(std::fs::rename(file_path.as_path(), new_file_path.as_path()).is_ok());
+        std::thread::sleep(Duration::from_millis(500));
+        // wait till rename
+        loop {
+            let fs_change = watcher.poll().unwrap();
+            if let Some(FsChange::Move(mov)) = fs_change {
+                assert_eq!(mov.source(), Path::new("/tmp/test/test.txt"));
+                assert_eq!(mov.destination(), Path::new("/tmp/test/new.txt"));
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(500));
+        }
+        // remove file
+        assert!(std::fs::remove_file(new_file_path.as_path()).is_ok());
+        // close tempdir
+        assert!(tempdir.close().is_ok());
+    }
+     */
+
+    #[test]
+    #[cfg(target_os = "macos")]
+    fn should_poll_nothing() {
+        let mut watcher = FsWatcher::init(Duration::from_secs(5)).unwrap();
+        let tempdir = TempDir::new().unwrap();
+        assert!(watcher
+            .watch(tempdir.path(), Path::new("/tmp/test"))
+            .is_ok());
+        assert!(watcher.poll().ok().unwrap().is_none());
+        // close tempdir
+        assert!(tempdir.close().is_ok());
+    }
+
+    #[test]
+    #[cfg(target_os = "macos")]
+    fn should_get_watched_paths() {
+        let mut watcher = FsWatcher::init(Duration::from_secs(5)).unwrap();
+        assert!(watcher.watch(Path::new("/tmp"), Path::new("/tmp")).is_ok());
+        assert!(watcher
+            .watch(Path::new("/home"), Path::new("/home"))
+            .is_ok());
+        let mut watched_paths = watcher.watched_paths();
+        watched_paths.sort();
+        assert_eq!(watched_paths, vec![Path::new("/home"), Path::new("/tmp")]);
+    }
+}