client: Lots of top-level client logic.

src/bin/cli.rs

@@ -37,6 +37,8 @@ pub enum Command {
     RemoveDirectory {
         local_directory: PathBuf,
     },
+    Pause,
+    Resume,
     // TODO: Command to update login information in case of login problems.
 }
 

src/bin/client.rs

@@ -1,11 +1,12 @@
 use std::fs::create_dir_all;
 use std::sync::Arc;
-use std::ffi::OsStr;
+use std::ffi::{OsStr, OsString};
 
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{UnixListener, UnixStream};
 use tokio::stream::StreamExt;
 use tokio::sync::{mpsc, Mutex};
+use tokio::signal;
 
 use twfss::{Database, FileTree, ClientSideSync, FileSystemWatcher, Error, SynchronizationError};
 
@@ -30,18 +31,22 @@ fn db_path() -> String {
 }
 
 struct SynchronizedDirectory {
+    path: OsString,
     client_side_sync: ClientSideSync,
     file_system_watcher: FileSystemWatcher,
+    last_error: Option<Error>,
 }
 
 impl SynchronizedDirectory {
     async fn open(database: Arc<Mutex<Database>>, directory: &OsStr, errors: mpsc::Sender<SynchronizationError>) -> Result<SynchronizedDirectory, Error> {
         let file_tree = Arc::new(Mutex::new(FileTree::open(database, directory)?));
-        let client_side_sync = ClientSideSync::new(file_tree.clone(), errors.clone()).await?;
-        let file_system_watcher = FileSystemWatcher::new(file_tree, errors).await?;
+        let client_side_sync = ClientSideSync::new(file_tree.clone(), errors.clone()).await;
+        let file_system_watcher = FileSystemWatcher::new(file_tree, errors).await;
         Ok(SynchronizedDirectory{
+            path: directory.to_owned(),
             client_side_sync,
             file_system_watcher,
+            last_error: None,
         })
     }
 
@@ -65,57 +70,89 @@ async fn main() {
     // Check whether the client is already running, and exit if it is.
     // TODO
 
-    // TODO: Register signal handler for graceful shutdown.
+    let mut ctrl_c = Box::pin(signal::ctrl_c());
+    let (errors_send, mut errors) = mpsc::channel(32);
 
-    // TODO: Error handling. If a directory was removed, remove it from the
-    // list of synchronized directories (or mark it for the user).
-    let (errors_send, errors) = mpsc::channel(32);
+    // We create the socket before starting synchronization, so that the unwrap() below does not
+    // cause corruption.
+    let mut listener = UnixListener::bind(cli::socket_path()).unwrap();
+    let mut incoming = listener.incoming();
 
     // Initialize the existing synchronized directories.
     let mut directories = Vec::new();
     let db = Arc::new(Mutex::new(Database::create_or_open(&db_path()).unwrap()));
     for directory in db.lock().await.synchronized_directories() {
-        // TODO: Error handling - remove the synchronized directory instead (or mark it for the
-        // user).
+        // The function cannot fail - we just fetched the directory from the database.
         let mut sync_dir = SynchronizedDirectory::open(db.clone(), &directory, errors_send.clone()).await.unwrap();
         sync_dir.unpause().await;
         directories.push(sync_dir);
     }
 
-    // Listen for CLI commands.
-    // TODO: Graceful shutdown on errors.
-    let mut listener = UnixListener::bind(cli::socket_path()).unwrap();
-    let mut incoming = listener.incoming();
-    while let Some(stream) = incoming.next().await {
-        match stream {
-            Ok(stream) => {
-                let mut stream = stream;
-                match handle_cli_client(&mut stream).await {
-                    Ok(()) => {}
-                    Err(e) => {
-                        // Log error and try to send it to the stream.
-                        // TODO
-                        stream
-                            .write_all(format!("Error: {:?}", e).as_bytes())
-                            .await
-                            .ok();
+    loop {
+        tokio::select! {
+            result = &mut ctrl_c => {
+                result.expect("could not listen for Ctrl+C");
+                // Ctrl+C was pressed, so we terminate the program.
+                break;
+            }
+            stream = incoming.next() => {
+                match stream {
+                    Some(Ok(stream)) => {
+                        let mut stream = stream;
+                        match handle_cli_client(&mut stream, &mut directories, db.clone()).await {
+                            Ok(()) => {}
+                            Err(e) => {
+                                // Log error and try to send it to the stream.
+                                // TODO: We want to send a return code as well.
+                                stream
+                                    .write_all(format!("Error: {:?}", e).as_bytes())
+                                    .await
+                                    .ok();
+                            }
+                        };
+                    }
+                    Some(Err(err)) => {
+                        eprintln!("Error while listening on the local unix socket: {:?}", err);
+                        break;
                     }
-                };
+                    None => {
+                        eprintln!("listening for CLI connections failed");
+                        break;
+                    }
+                }
             }
-            Err(err) => {
-                eprintln!("Error while listening on the local unix socket: {:?}", err);
-                break;
+            error = errors.next() => {
+                let error = error.unwrap();
+                eprintln!("error: {:?}", error.error);
+                for directory in directories.iter_mut() {
+                    if directory.path == error.directory {
+                        // In theory, there could be a race condition here where we pause a
+                        // directory that was just recreated, when the directory which caused the
+                        // error was already deleted.
+                        if error.needs_pause {
+                            directory.pause().await;
+                        }
+                        directory.last_error = Some(error.error);
+                        break;
+                    }
+                }
             }
         }
     }
+
+    // For a clean shutdown, we need to pause all synchtonized directories. Then,
+    // we can be sure that the database is not modified anymore.
+    for mut directory in directories.into_iter() {
+        directory.pause().await;
+    }
 }
 
-async fn handle_cli_client(stream: &mut UnixStream) -> Result<(), Error> {
+async fn handle_cli_client(stream: &mut UnixStream, directories: &mut Vec<SynchronizedDirectory>, _db: Arc<Mutex<Database>>) -> Result<(), Error> {
     let mut request = String::new();
     stream.read_to_string(&mut request).await?;
 
     let options: cli::Options = serde_json::from_str(&request)?;
-    let _verbose = options.verbose;
+    let verbose = options.verbose;
     match options.command {
         _cmd @ cli::Command::ListDirectories { .. } => {
             // TODO
@@ -126,9 +163,26 @@ async fn handle_cli_client(stream: &mut UnixStream) -> Result<(), Error> {
         _cmd @ cli::Command::RemoveDirectory { .. } => {
             // TODO
         }
+        cli::Command::Pause => {
+            for  directory in directories.iter_mut() {
+                directory.pause().await;
+                if verbose {
+                    stream.write_all(format!("Paused {}.\n", directory.path.to_string_lossy()).as_bytes()).await.ok();
+                }
+            }
+            stream.write_all("Synchronization paused.\n".as_bytes()).await.ok();
+        }
+        cli::Command::Resume => {
+            for  directory in directories.iter_mut() {
+                directory.unpause().await;
+                if verbose {
+                    stream.write_all(format!("Resumed {}.\n", directory.path.to_string_lossy()).as_bytes()).await.ok();
+                }
+            }
+            stream.write_all("Synchronization resumed.\n".as_bytes()).await.ok();
+        }
     }
 
-    // TODO
     Ok(())
 }
 

src/client_side_sync.rs

@@ -17,7 +17,7 @@ impl ClientSideSync {
     pub async fn new(
         file_tree: Arc<Mutex<FileTree>>,
         errors: mpsc::Sender<SynchronizationError>,
-    ) -> Result<ClientSideSync, Error> {
+    ) -> ClientSideSync {
         let new_version_source = file_tree.lock().await.new_version_source();
         // Spawn the task that connects to the server as soon as unpause() is called.
         let (state_send, pause_receive) = mpsc::channel(1);
@@ -30,10 +30,10 @@ impl ClientSideSync {
             sync_task.task_paused(pause_receive).await;
         });
 
-        Ok(ClientSideSync{
+        ClientSideSync{
             state_send,
             new_version_source,
-        })
+        }
     }
 
     // TODO: Deduplicate this code.

src/file_system_watcher.rs

@@ -20,15 +20,19 @@ pub struct FileSystemWatcher {
 }
 
 impl FileSystemWatcher {
+    // TODO: Let the caller pass a filter to ignore the files, and then apply that filter only when
+    // files are not already in the database?
     pub async fn new(
         file_tree: Arc<Mutex<FileTree>>,
         errors: mpsc::Sender<SynchronizationError>,
-    ) -> Result<FileSystemWatcher, Error> {
+    ) -> FileSystemWatcher {
         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 directory = file_tree.lock().await.root_path().to_owned();
             let mut fsw_task = FileSystemWatcherTask {
+                directory,
                 file_tree,
                 errors,
                 new_version_source,
@@ -36,10 +40,10 @@ impl FileSystemWatcher {
             fsw_task.task_paused(pause_receive).await;
         });
 
-        Ok(FileSystemWatcher {
+        FileSystemWatcher {
             state_send,
             new_version_source,
-        })
+        }
     }
 
     pub async fn pause(&mut self) {
@@ -62,6 +66,7 @@ impl FileSystemWatcher {
 }
 
 struct FileSystemWatcherTask {
+    directory: OsString,
     file_tree: Arc<Mutex<FileTree>>,
     errors: mpsc::Sender<SynchronizationError>,
     new_version_source: NewVersionSource,
@@ -146,6 +151,7 @@ impl FileSystemWatcherTask {
         match self
             .errors
             .send(SynchronizationError {
+                directory: self.directory.clone(),
                 needs_pause: true,
                 error: reason,
             })
@@ -404,6 +410,7 @@ impl FileSystemWatcherTask {
         // TODO: Encode the path in the error!
         self.errors
             .send(SynchronizationError {
+                directory: self.directory.clone(),
                 needs_pause: false,
                 error: error,
             })

src/lib.rs

@@ -1,6 +1,7 @@
 //! Library for file-system synchronization.
 extern crate rmp_serde as rmps;
 
+use std::ffi::OsString;
 use std::io;
 
 mod client_side_sync;
@@ -87,64 +88,12 @@ pub use file_tree::FileTree;
 // - last logged in
 // - conflict (only for server, a second upload-only client logged in since the last logout)
 
-//use std::sync::{Arc, Mutex};
-//
-//pub mod protocol;
-//
-//pub struct SynchronizedDirectory {
-//    db: Arc<Mutex<Database>>,
-//    master: bool,
-//    // TODO
-//}
-//
-//impl SynchronizedDirectory {
-//    pub fn open(
-//        db: Arc<Mutex<Database>>,
-//        _path: &str,
-//        master: bool,
-//    ) -> Result<SynchronizedDirectory, Error> {
-//        // TODO
-//        Ok(SynchronizedDirectory { db, master })
-//    }
-//
-//    // TODO: Login information.
-//    pub fn new(
-//        db: Arc<Mutex<Database>>,
-//        local_path: &str,
-//        remote_path: &str,
-//        master: bool,
-//    ) -> Result<SynchronizedDirectory, Error> {
-//        // TODO
-//        Ok(SynchronizedDirectory { db, master })
-//    }
-//}
-//
-//pub struct SyncEvent {
-//    // TODO
-//}
-//
-//impl SyncEvent {
-//    // TODO
-//}
-
 pub struct SynchronizationError {
+    pub directory: OsString,
     pub needs_pause: bool,
     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),