More file system watcher code.

src/file_system_watcher.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 use std::ffi::OsString;
 use std::fs;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -10,25 +11,35 @@ use tokio::stream::StreamExt;
 use tokio::sync::{mpsc, oneshot, Mutex};
 use tokio::task;
 
-use super::{Error, FileTree, SynchronizationError};
+use super::file_tree::{FileInfo, FileTree, FileType, LockedPath, NewVersionSource, OriginInfo};
+use super::{Error, SynchronizationError};
 
 pub struct FileSystemWatcher {
     state_send: mpsc::Sender<(StateChange, oneshot::Sender<()>)>,
+    new_version_source: NewVersionSource,
 }
 
 impl FileSystemWatcher {
-    pub fn new(
+    pub async fn new(
         file_tree: Arc<Mutex<FileTree>>,
         errors: mpsc::Sender<SynchronizationError>,
     ) -> Result<FileSystemWatcher, Error> {
+        let new_version_source = file_tree.lock().await.new_version_source();
         // Spawn the task that watches the file system as soon as unpause() is called.
         let (state_send, pause_receive) = mpsc::channel(1);
         tokio::spawn(async move {
-            let mut fsw_task = FileSystemWatcherTask { file_tree, errors };
+            let mut fsw_task = FileSystemWatcherTask {
+                file_tree,
+                errors,
+                new_version_source,
+            };
             fsw_task.task_paused(pause_receive).await;
         });
 
-        Ok(FileSystemWatcher { state_send })
+        Ok(FileSystemWatcher {
+            state_send,
+            new_version_source,
+        })
     }
 
     pub async fn pause(&mut self) {
@@ -53,6 +64,7 @@ impl FileSystemWatcher {
 struct FileSystemWatcherTask {
     file_tree: Arc<Mutex<FileTree>>,
     errors: mpsc::Sender<SynchronizationError>,
+    new_version_source: NewVersionSource,
 }
 
 impl FileSystemWatcherTask {
@@ -160,77 +172,102 @@ impl FileSystemWatcherTask {
         // directory is not readable, we do not want to synchronize it.
         match event {
             FileSystemEvent::Stopped(reason) => match reason {
-                StopReason::DirectoryRemoved => {
-                    return Err(Error::DirectoryRemoved);
-                }
+                StopReason::DirectoryRemoved => Err(Error::DirectoryRemoved),
             },
-            FileSystemEvent::DirectoryWatched(path) => {
-                // Fetch the directory contents as well as the database contents.
-                // TODO: We load all entries into a vector, which might be problematic for *very*
-                // large directories.
-                let fs_entries = match fs::read_dir(&path) {
-                    Ok(entries) => entries,
-                    Err(_) => return Ok(()),
-                };
-                let fs_entries = fs_entries
-                    .filter(|x| x.is_ok())
-                    .map(|x| (x.as_ref().unwrap().file_name(), x.unwrap()))
-                    .collect::<HashMap<_, _>>();
-                let db_entries = match self.file_tree.lock().await.get_directory_listing(&path) {
-                    Some(entries) => entries,
-                    None => return Ok(()),
-                };
-                let db_entries = db_entries
-                    .into_iter()
-                    .map(|x| (x.file_name.clone(), x))
-                    .collect::<HashMap<_, _>>();
-
-                // Compare file metadata and report changed, deleted and created files and
-                // directories.
-                for (name, _entry) in fs_entries.iter() {
-                    if db_entries.contains_key(name) {
-                        // The entry is known, compare the file attributes.
-                        // TODO
-                        // If the file attributes do not match, compare the file checksums.
-                        // TODO
-                    } else {
-                        // The file seems to be new, insert it into the database.
-                        // TODO
-                    }
-                }
-                for (name, _entry) in db_entries.iter() {
-                    if !fs_entries.contains_key(name) {
-                        // The file has been deleted, remove it from the database.
-                        // TODO
-                    }
-                }
-            }
-            FileSystemEvent::DirectoryCreated(_path) => {
-                // TODO
-            }
+            FileSystemEvent::DirectoryWatched(path) => self.handle_directory_watched(path).await,
+            FileSystemEvent::DirectoryCreated(path) => self.handle_directory_created(path).await,
             FileSystemEvent::DirectoryModified(_path) => {
                 // TODO
+                Ok(())
             }
-            FileSystemEvent::DirectoryRemoved(_path) => {
-                // TODO
-            }
+            FileSystemEvent::DirectoryRemoved(path) => self.handle_directory_removed(path).await,
             FileSystemEvent::DirectoryMoved(_from, _to) => {
                 // TODO
+                Ok(())
             }
-            FileSystemEvent::FileCreated(_path) => {
+            FileSystemEvent::FileCreated(path) => self.handle_file_modified(path).await,
+            FileSystemEvent::FileModified(path) => self.handle_file_modified(path).await,
+            FileSystemEvent::FileRemoved(path) => self.handle_file_removed(path).await,
+            FileSystemEvent::FileMoved(_from, _to) => {
                 // TODO
+                Ok(())
             }
-            FileSystemEvent::FileModified(_path) => {
-                // TODO
+            FileSystemEvent::Error(e) => {
+                self.send_error_no_pause(e.into(), None).await;
+                Ok(())
             }
-            FileSystemEvent::FileRemoved(_path) => {
-                // TODO
+        }
+    }
+
+    async fn handle_directory_watched(&mut self, path: OsString) -> Result<(), Error> {
+        // We simply lock the whole directory to remove the need for
+        // double-checking before/after locks.
+        // TODO: More fine-grained locking.
+        let path_lock = match self.file_tree.lock().await.lock_path(&path).await {
+            Some(lock) => lock,
+            None => {
+                // The entry does not exist anymore.
+                return Ok(());
             }
-            FileSystemEvent::FileMoved(_from, _to) => {
+        };
+
+        // Fetch the directory contents as well as the database contents.
+        // TODO: We load all entries into a vector, which might be problematic for *very*
+        // large directories.
+        let fs_entries = match fs::read_dir(&path) {
+            Ok(entries) => entries,
+            Err(_) => return Ok(()),
+        };
+        let fs_entries = fs_entries
+            .filter(|x| x.is_ok())
+            .map(|x| (x.as_ref().unwrap().file_name(), x.unwrap()))
+            .collect::<HashMap<_, _>>();
+        let db_entries = match path_lock.get_directory_listing(&path) {
+            Some(entries) => entries,
+            None => return Ok(()),
+        };
+        let db_entries = db_entries
+            .into_iter()
+            .map(|x| (x.file_name.clone(), x))
+            .collect::<HashMap<_, _>>();
+
+        // Compare file metadata and report changed, deleted and created files and
+        // directories.
+        for (name, entry) in fs_entries.iter() {
+            if db_entries.contains_key(name) {
+                let db_entry = &db_entries[name];
+                let mut entry_path = PathBuf::from(&path);
+                entry_path.push(name);
+
+                // The entry is known, compare the file attributes.
+                let fs_metadata = match entry.metadata() {
+                    Ok(m) => m,
+                    Err(_) => continue,
+                };
+                let local_type = match FileType::from(fs_metadata.file_type()) {
+                    Some(t) => t,
+                    None => continue,
+                };
+                // If the type is different, interpret as deletion+creation.
+                if local_type != db_entry.file_type {
+                    path_lock.local_file_removed(
+                        self.new_version_source,
+                        entry_path.as_path().as_os_str(),
+                    );
+                // TODO: Create the new file/directory.
+                } else if local_type != FileType::Directory {
+                    // If the file attributes do not match, compare the file checksums.
+                    // TODO
+                }
+            } else {
+                // The file seems to be new, insert it into the database.
                 // TODO
             }
-            FileSystemEvent::Error(e) => {
-                self.send_error_no_pause(e.into(), None).await;
+        }
+        for (name, _entry) in db_entries.iter() {
+            if !fs_entries.contains_key(name) {
+                // The file has been deleted, remove it from the database.
+                // TODO
             }
         }
 
@@ -242,6 +279,127 @@ impl FileSystemWatcherTask {
         Ok(())
     }
 
+    async fn handle_directory_created<P>(&mut self, path: P) -> Result<(), Error>
+    where
+        P: AsRef<Path>,
+    {
+        let path = path.as_ref();
+        let parent_path = path.parent().unwrap();
+        let (path_lock, fs_metadata, db_metadata) =
+            self.prepare_single_path_change(parent_path, path).await;
+        if path_lock.is_none() {
+            return Ok(());
+        }
+        if fs_metadata.is_none() || !fs_metadata.unwrap().is_dir() {
+            return Ok(());
+        }
+        if db_metadata.is_some() {
+            if db_metadata.unwrap().file_type != FileType::Directory {
+                // TODO: Log an error.
+            }
+            return Ok(());
+        }
+        path_lock
+            .unwrap()
+            .create_directory(self.new_version_source, path, OriginInfo::Local)?;
+        Ok(())
+    }
+
+    async fn handle_directory_removed<P>(&mut self, path: P) -> Result<(), Error>
+    where
+        P: AsRef<Path>,
+    {
+        let path = path.as_ref();
+        let parent_path = path.parent().unwrap();
+        let (path_lock, fs_metadata, db_metadata) =
+            self.prepare_single_path_change(parent_path, path).await;
+        if path_lock.is_none() || db_metadata.is_none() {
+            return Ok(());
+        }
+        if fs_metadata.is_some() && fs_metadata.unwrap().is_dir() {
+            // The directory exists again.
+            return Ok(());
+        }
+        path_lock
+            .unwrap()
+            .local_file_removed(self.new_version_source, path);
+        Ok(())
+    }
+
+    async fn handle_file_modified<P>(&mut self, path: P) -> Result<(), Error>
+    where
+        P: AsRef<Path>,
+    {
+        // This function is also called for new files. If we create a file ourselves, and the user
+        // immediately modifies it, we need to interpref "FileCreated" as a modification.
+        // However, that means that we also need to expect that the file does not yet exist in the
+        // database yet.
+        let path = path.as_ref();
+        let path_lock = match self.file_tree.lock().await.lock_path(path).await {
+            Some(lock) => lock,
+            None => {
+                // The entry does not exist in the database, lock the parent.
+                // TODO
+                return Ok(());
+            }
+        };
+        let metadata = match fs::symlink_metadata(path) {
+            Ok(metadata) => metadata,
+            Err(_) => return Ok(()),
+        };
+        if metadata.is_dir() {
+            // The directory exists again, ignore.
+            return Ok(());
+        }
+        let db_metadata = match path_lock.get_file_info(path) {
+            Some(m) => m,
+            None => {
+                // The directory was already removed from the database - i.e., it was deleted via
+                // the network.
+                return Ok(());
+            }
+        };
+
+        // TODO
+        Ok(())
+    }
+
+    async fn handle_file_removed<P>(&mut self, path: P) -> Result<(), Error>
+    where
+        P: AsRef<Path>,
+    {
+        let path = path.as_ref();
+        let parent_path = path.parent().unwrap();
+        let (path_lock, fs_metadata, db_metadata) =
+            self.prepare_single_path_change(parent_path, path).await;
+        if path_lock.is_none() || db_metadata.is_none() {
+            return Ok(());
+        }
+        // TODO: What about symlinks?
+        if fs_metadata.is_some() && fs_metadata.unwrap().is_file() {
+            // The file exists again.
+            return Ok(());
+        }
+        path_lock
+            .unwrap()
+            .local_file_removed(self.new_version_source, path);
+        Ok(())
+    }
+
+    async fn prepare_single_path_change(
+        &mut self,
+        lock_path: &Path,
+        path: &Path,
+    ) -> (Option<LockedPath>, Option<fs::Metadata>, Option<FileInfo>) {
+        let path_lock = match self.file_tree.lock().await.lock_path(lock_path).await {
+            Some(lock) => lock,
+            None => return (None, None, None),
+        };
+        let fs_metadata = fs::symlink_metadata(path).ok();
+        let db_metadata = path_lock.get_file_info(path);
+        (Some(path_lock), fs_metadata, db_metadata)
+    }
+
     async fn send_error_no_pause(&mut self, error: Error, _path: Option<OsString>) {
         // TODO: Encode the path in the error!
         self.errors

src/file_tree.rs

@@ -1,4 +1,6 @@
 use std::ffi::{OsStr, OsString};
+use std::fs;
+use std::path::Path;
 use std::sync::Arc;
 
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
@@ -54,29 +56,62 @@ impl FileTree {
         source
     }
 
-    pub fn get_file_info(&self, _path: &OsStr) -> Option<FileInfo> {
+    // TODO: Limit on size of all temporary files, make this function return a
+    // future which yields a temporary file once enough space is available?
+    pub fn create_temporary_file(&self, _target_path: &OsStr) -> Result<TemporaryFile, Error> {
         // TODO
         panic!("Not yet implemented.");
     }
 
-    pub fn get_directory_listing(&self, _path: &OsStr) -> Option<Vec<FileInfo>> {
+    pub async fn lock_path<P>(&mut self, path: P) -> Option<LockedPath>
+    where
+        P: AsRef<Path>,
+    {
+        let _path = path.as_ref();
         // TODO
         panic!("Not yet implemented.");
     }
 
-    pub fn local_file_changed(&self, _source: NewVersionSource, _path: &OsStr, _info: FileInfo) {
+    pub async fn lock_two_paths<P1, P2>(&mut self, path1: P1, path2: P2) -> Option<LockedPath>
+    where
+        P1: AsRef<Path>,
+        P2: AsRef<Path>,
+    {
+        let _path1 = path1.as_ref();
+        let _path2 = path2.as_ref();
         // TODO
         panic!("Not yet implemented.");
     }
+}
+
+pub struct LockedPath {
+    // TODO
+}
 
-    pub fn local_file_removed(&self, _source: NewVersionSource, _path: &OsStr) {
+impl LockedPath {
+    pub fn get_file_info<P>(&self, path: P) -> Option<FileInfo>
+    where
+        P: AsRef<Path>,
+    {
+        let _path = path.as_ref();
         // TODO
         panic!("Not yet implemented.");
     }
 
-    // TODO: Limit on size of all temporary files, make this function return a
-    // future which yields a temporary file once enough space is available?
-    pub fn create_temporary_file(&self, _target_path: &OsStr) -> Result<TemporaryFile, Error> {
+    pub fn get_directory_listing(&self, _path: &OsStr) -> Option<Vec<FileInfo>> {
+        // TODO
+        panic!("Not yet implemented.");
+    }
+
+    pub fn local_file_changed(&self, _source: NewVersionSource, _path: &OsStr, _info: FileInfo) {
+        // TODO
+        panic!("Not yet implemented.");
+    }
+
+    pub fn local_file_removed<P>(&self, _source: NewVersionSource, _path: P)
+    where
+        P: AsRef<Path>,
+    {
         // TODO
         panic!("Not yet implemented.");
     }
@@ -93,17 +128,27 @@ impl FileTree {
         panic!("Not yet implemented.");
     }
 
-    pub fn create_directory(
+    pub fn create_directory<P>(
         &self,
         _source: NewVersionSource,
-        _path: &OsStr,
+        _path: P,
         _origin: OriginInfo,
-    ) -> Result<NewVersionEvent, Error> {
+    ) -> Result<NewVersionEvent, Error>
+    where
+        P: AsRef<Path>,
+    {
         // TODO
         panic!("Not yet implemented.");
     }
 }
 
+impl Drop for LockedPath {
+    fn drop(&mut self) {
+        // Release the lock again.
+        // TODO
+    }
+}
+
 pub struct NewVersionNotifier {
     // TODO: Probably should not be public.
     pub receive: UnboundedReceiver<NewVersionEvent>,
@@ -121,13 +166,42 @@ pub struct NewVersionSource {
 
 pub struct FileInfo {
     pub file_name: OsString,
+    pub file_type: FileType,
+    /// Local modification time in seconds since the Unix epoch. Only valid for files and symlinks,
+    /// not for directories.
+    pub local_modification_time: Option<u64>,
+    /// Local size of the file in bytes. Only valid for files and symlinks, not for directories.
+    pub local_size: Option<u64>,
     // TODO
 }
 
+#[derive(PartialEq)]
+pub enum FileType {
+    File,
+    Directory,
+    Symlink,
+}
+
+impl FileType {
+    pub fn from(ft: fs::FileType) -> Option<FileType> {
+        if ft.is_file() {
+            Some(FileType::File)
+        } else if ft.is_dir() {
+            Some(FileType::Directory)
+        } else if ft.is_symlink() {
+            Some(FileType::Symlink)
+        } else {
+            None
+        }
+    }
+}
+
 pub struct TemporaryFile {
     // TODO
 }
 
-pub struct OriginInfo {
-    // TODO
+pub enum OriginInfo {
+    Local,
+    // TODO: For remote sources, include the previous version to detect conflicts?
+    Remote,
 }

src/lib.rs

@@ -125,6 +125,19 @@ pub struct SynchronizationError {
     pub error: Error,
 }
 
+impl<T> From<T> for SynchronizationError
+where
+    T: Into<Error>,
+{
+    fn from(e: T) -> SynchronizationError {
+        // By default, errors do not set the pause flag.
+        SynchronizationError {
+            needs_pause: false,
+            error: e.into(),
+        }
+    }
+}
+
 #[derive(Debug)]
 pub enum Error {
     Io(std::io::Error),