File system watcher error reporting.

src/file_system_watcher.rs

@@ -7,18 +7,21 @@ use tokio::stream::StreamExt;
 use tokio::sync::{mpsc, oneshot, Mutex};
 use tokio::task;
 
-use super::{Error, FileTree};
+use super::{Error, FileTree, SynchronizationError};
 
 pub struct FileSystemWatcher {
     state_send: mpsc::Sender<(StateChange, oneshot::Sender<()>)>,
 }
 
 impl FileSystemWatcher {
-    pub fn new(file_tree: Arc<Mutex<FileTree>>) -> Result<FileSystemWatcher, Error> {
+    pub fn new(
+        file_tree: Arc<Mutex<FileTree>>,
+        errors: mpsc::Sender<SynchronizationError>,
+    ) -> Result<FileSystemWatcher, Error> {
         // 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 {
-            Self::task_paused(file_tree, pause_receive).await;
+            Self::task_paused(file_tree, pause_receive, errors).await;
         });
 
         Ok(FileSystemWatcher { state_send })
@@ -45,6 +48,7 @@ impl FileSystemWatcher {
     async fn task_paused(
         mut file_tree: Arc<Mutex<FileTree>>,
         mut state_receive: mpsc::Receiver<(StateChange, oneshot::Sender<()>)>,
+        mut errors: mpsc::Sender<SynchronizationError>,
     ) {
         let mut paused = true;
 
@@ -61,7 +65,8 @@ impl FileSystemWatcher {
             } else {
                 // We were unpaused, initialize the file system watcher and start waiting for
                 // that as well.
-                let next_state = Self::task_unpaused(&mut file_tree, &mut state_receive).await;
+                let next_state =
+                    Self::task_unpaused(&mut file_tree, &mut state_receive, &mut errors).await;
                 match next_state {
                     StateChange::Pause => paused = true,
                     StateChange::Unpause => paused = false,
@@ -74,19 +79,23 @@ impl FileSystemWatcher {
     async fn task_unpaused(
         file_tree: &mut Arc<Mutex<FileTree>>,
         state_receive: &mut mpsc::Receiver<(StateChange, oneshot::Sender<()>)>,
+        errors: &mut mpsc::Sender<SynchronizationError>,
     ) -> StateChange {
         let root_path = file_tree.lock().await.root_path().to_owned();
-        // TODO: Proper error handling.
-        let mut file_events = FileEventDelay::new(
-            fswatcher::FileSystemWatcher::new(&root_path).unwrap(),
-            Duration::from_secs(3),
-        );
+        let fsw = match fswatcher::FileSystemWatcher::new(&root_path) {
+            Ok(fsw) => fsw,
+            Err(e) => return Self::wait_for_pause(e.into(), state_receive, errors).await,
+        };
+        let mut file_events = FileEventDelay::new(fsw, Duration::from_secs(3));
 
         loop {
+            // We only want to wait for file system events until a state change pauses the file
+            // system watcher.
             match future::select(Box::pin(file_events.next()), Box::pin(state_receive.recv())).await
             {
                 Either::Left((event, _state_future)) => {
-                    // TODO: Error handling, handle None event.
+                    // We cannot get any end-of-stream None here - we would first get a "Stopped"
+                    // file event.
                     Self::handle_file_event(file_tree, event.unwrap()).await;
                 }
                 Either::Right((next_state, _file_event_future)) => {
@@ -96,13 +105,41 @@ impl FileSystemWatcher {
                             send.send(()).unwrap();
                             return StateChange::Pause;
                         }
-                        (StateChange::Terminate, _) => return StateChange::Pause,
+                        (StateChange::Terminate, _) => return StateChange::Terminate,
                     };
                 }
             };
         }
     }
 
+    async fn wait_for_pause(
+        reason: Error,
+        state_receive: &mut mpsc::Receiver<(StateChange, oneshot::Sender<()>)>,
+        errors: &mut mpsc::Sender<SynchronizationError>,
+    ) -> StateChange {
+        match errors
+            .send(SynchronizationError {
+                needs_pause: true,
+                error: reason,
+            })
+            .await
+        {
+            Ok(()) => (),
+            Err(_) => panic!("Could not send synchronization error!"),
+        };
+        loop {
+            match state_receive.recv().await.unwrap() {
+                (StateChange::Unpause, send) => send.send(()).unwrap(),
+                (StateChange::Pause, send) => {
+                    send.send(()).unwrap();
+                    return StateChange::Pause;
+                }
+                (StateChange::Terminate, _) => return StateChange::Terminate,
+            }
+        }
+        // TODO
+    }
+
     async fn handle_file_event(_file_tree: &mut Arc<Mutex<FileTree>>, event: FileSystemEvent) {
         match event {
             FileSystemEvent::Stopped(_reason) => {

src/lib.rs

@@ -120,6 +120,11 @@ pub use file_tree::FileTree;
 //    // TODO
 //}
 
+pub struct SynchronizationError {
+    pub needs_pause: bool,
+    pub error: Error,
+}
+
 #[derive(Debug)]
 pub enum Error {
     Io(std::io::Error),
@@ -130,3 +135,11 @@ impl From<io::Error> for Error {
         Error::Io(e)
     }
 }
+
+impl From<fswatcher::Error> for Error {
+    fn from(e: fswatcher::Error) -> Error {
+        match e {
+            fswatcher::Error::Io(e) => e.into(),
+        }
+    }
+}