Add FileTree API for central synchronization mechanisms.

Cargo.lock

@@ -877,6 +877,7 @@ version = "0.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "bytes 0.5.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "fnv 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "iovec 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "lazy_static 1.4.0 (registry+https://github.com/rust-lang/crates.io-index)",

Cargo.toml

@@ -11,7 +11,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 dirs = "2.0"
 byteorder = "1.3"
-tokio = { version = "0.2", features = ["fs", "io-util", "macros", "net", "stream", "tcp"] }
+tokio = { version = "0.2", features = ["fs", "io-util", "macros", "net", "stream", "sync", "tcp"] }
 futures = "0.3.1"
 futures-tokio-compat = { git = "https://github.com/mgottschlag/futures-tokio-compat.git" }
 async-tungstenite = { version = "0.3", features = ["tokio-runtime"] }

src/bin/client.rs

@@ -6,7 +6,7 @@ use tokio::net::{UnixListener, UnixStream};
 use tokio::stream::StreamExt;
 use tokio::io::{AsyncWriteExt, AsyncReadExt};
 
-use twfss::{Database, SynchronizedDirectory};
+use twfss::Database;
 
 mod cli;
 
@@ -40,13 +40,13 @@ async fn main() {
     // TODO: Register signal handler for graceful shutdown.
 
     // 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().unwrap().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().unwrap().synchronized_directories() {
         // TODO: Error handling. If the directory was removed, remove it from the
         // list of synchronized directories.
         directories.push(SynchronizedDirectory::open(db.clone(), &directory, false).unwrap());
-    }
+    }*/
 
     // Listen for CLI commands.
     // TODO: Graceful shutdown on errors.

src/bin/server.rs

@@ -1,4 +1,6 @@
 use std::net::SocketAddr;
+use std::sync::{Arc, Mutex};
+use std::fs::create_dir_all;
 
 use async_tungstenite::accept_async;
 use log::*;
@@ -7,6 +9,8 @@ use tokio::stream::StreamExt;
 use futures_tokio_compat::Compat;
 use futures::sink::SinkExt;
 
+use twfss::Database;
+
 fn config_dir() -> String {
     format!(
         "{}/.config/twfss",
@@ -29,7 +33,11 @@ fn db_path() -> String {
 async fn main() {
     env_logger::init();
 
+    // Create the configuration directory if necessary.
+    create_dir_all(config_dir()).unwrap();
+
     // Read configuration.
+    let _db = Arc::new(Mutex::new(Database::create_or_open(&db_path()).unwrap()));
     // TODO
 
     // Start a thread that periodically checks whether the maximum database size

src/database.rs

@@ -0,0 +1,15 @@
+pub struct Database {
+    // TODO
+}
+
+impl Database {
+    pub fn create_or_open(_path: &str) -> Result<Database, crate::Error> {
+        // TODO
+        Ok(Database {})
+    }
+
+    pub fn synchronized_directories(&self) -> Vec<String> {
+        // TODO
+        Vec::new()
+    }
+}

src/filetree.rs

@@ -0,0 +1,124 @@
+use std::sync::Arc;
+
+use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
+use tokio::sync::Mutex;
+
+use super::Database;
+use super::Error;
+
+pub struct FileTree {
+    db: Arc<Mutex<Database>>,
+    new_version_notifiers: Vec<UnboundedSender<NewVersionEvent>>,
+    next_version_source: u64,
+}
+
+impl FileTree {
+    pub fn open(db: Arc<Mutex<Database>>, _directory: &str) -> Result<FileTree, Error> {
+        // Check whether the database entry for this file tree exists and fetch the root
+        // file ID.
+        // TODO
+
+        // Create a lock file to make sure that the directory is only synchronized once.
+        // TODO
+
+        // Clear up any temporary files which were left over.
+        // TODO
+
+        // Enqueue the root directory to be rechecked against the database.
+        // TODO
+
+        Ok(FileTree {
+            db,
+            new_version_notifiers: Vec::new(),
+            next_version_source: 0,
+        })
+    }
+
+    pub fn new_notifier(&mut self) -> NewVersionNotifier {
+        let (send, receive) = unbounded_channel();
+        self.new_version_notifiers.push(send);
+        NewVersionNotifier { receive }
+    }
+
+    pub fn new_version_source(&mut self) -> NewVersionSource {
+        let source = NewVersionSource {
+            id: self.next_version_source,
+        };
+        self.next_version_source += 1;
+        source
+    }
+
+    pub fn get_file_info(_path: &str) -> Option<FileInfo> {
+        // TODO
+        panic!("Not yet implemented.");
+    }
+
+    pub fn get_directory_listing(_path: &str) -> Option<Vec<FileInfo>> {
+        // TODO
+        panic!("Not yet implemented.");
+    }
+
+    pub fn local_file_changed(_source: NewVersionSource, _path: &str, _info: FileInfo) {
+        // TODO
+        panic!("Not yet implemented.");
+    }
+
+    pub fn local_file_removed(_source: NewVersionSource, _path: &str) {
+        // 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(_target_path: &str) -> Result<TemporaryFile, Error> {
+        // TODO
+        panic!("Not yet implemented.");
+    }
+
+    pub fn update_file(
+        _source: NewVersionSource,
+        _path: &str,
+        _info: FileInfo,
+        _contents: TemporaryFile,
+        _origin: OriginInfo,
+    ) -> Result<NewVersionEvent, Error> {
+        // TODO
+        panic!("Not yet implemented.");
+    }
+
+    pub fn create_directory(
+        _source: NewVersionSource,
+        _path: &str,
+        _origin: OriginInfo,
+    ) -> Result<NewVersionEvent, Error> {
+        // TODO
+        panic!("Not yet implemented.");
+    }
+}
+
+pub struct NewVersionNotifier {
+    // TODO: Probably should not be public.
+    pub receive: UnboundedReceiver<NewVersionEvent>,
+}
+
+pub struct NewVersionEvent {
+    pub source: NewVersionSource,
+    // TODO
+}
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub struct NewVersionSource {
+    id: u64,
+}
+
+pub struct FileInfo {
+    // TODO
+}
+
+pub struct TemporaryFile {
+    // TODO
+}
+
+pub struct OriginInfo {
+    // TODO
+}

src/lib.rs

@@ -1,57 +1,122 @@
 use std::io;
-use std::sync::{Arc, Mutex};
 
-pub struct SynchronizedDirectory {
-    db: Arc<Mutex<Database>>,
-    master: bool,
-    // TODO
-}
+mod database;
+mod filetree;
 
-impl SynchronizedDirectory {
-    pub fn open(
-        db: Arc<Mutex<Database>>,
-        _path: &str,
-        master: bool,
-    ) -> Result<SynchronizedDirectory, Error> {
-        // TODO
-        Ok(SynchronizedDirectory { db, master })
-    }
+pub use database::Database;
+pub use filetree::FileTree;
 
-    // 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 })
-    }
-}
+// Sketch of the synchronization algorithm:
+//
+// if (filesystem_change) {
+//     if the change is already represented in the mode in the database, ignore it;
+//     scan the directory/file for changes (recalculate hashes if necessary);
+//     if necessary, update the local version;
+//     notify client connections about the change, push the change to server connections;
+//     if a server rejects the change, copy contents to a conflict file and revert;
+// }
+// if (new_version_notification_from_server) {
+//     the server sent us a path and a version for each element in the path;
+//     if the path is a file, download the hashes for the file and start downloading
+//     content;
+//     if the path is a directory, list the directory (recursively) and download changes;
+//     update the version whenever an element was locally updated;
+//     update the version of directories when all files/directories below it were updated;
+//     notify clients;
+// }
+// if (update_from_client) {
+//     compare the client's server version with our version, if different, early reject;
+//     download file information and store in temporary file;
+//     check whether our version changed, if so, late reject;
+//     move file and update;
+//     update database version and hashes;
+//     push update to server, similar to local file system change;
+// }
+//
+// Synchronization issues between the three code paths:
+// - File system operations (except for temporary files) and database changes (final
+// conflict checks as well as hash updates?) need to be synchronized.
+// - If any operation locks a directory, operations to the files within the directory can
+// be delayed due to locking, and have to re-check whether the file in question is still
+// the same file (version check + ID check?).
 
-pub struct SyncEvent {
-    // TODO
-}
-
-impl SyncEvent {
-    // TODO
-}
+// Sketch of the database structure:
+// (TODO: Rework and adapt to reality?)
+//
+// The server-side database has a list of synchronization roots and stores a file tree for each.
+// It also stores all recent/valid synchronization "connection" declarations, to be able to prune
+// synchronization roots periodically.
+//
+// The client-side databse has a list of synchronization roots and stores a file tree for each.
+// It also stores one synchronization "connection" for each root, which contains the server address,
+// login information and the mode of synchronization.
+//
+// FileTree:
+// - path
+// Object (file or directory):
+// - path
+// - type
+// - version
+// - remote version
+// - remote author (connection which created the last remote version)
+// - modification time
+// - size
+// - checksum
+// - content-id (points to file contents, can stay the same for file move)
+// Content (for files, one per ):
+// - content-id
+// - offset
+// - length
+// - checksum
+//
+// Synchronization connection:
+// - id
+// - key
+// - path
+// - address (only for client)
+// - direction
+// - last logged in
+// - conflict (only for server, a second upload-only client logged in since the last logout)
 
-pub struct Database {
-    // TODO
-}
-
-impl Database {
-    pub fn create_or_open(_path: &str) -> Result<Database, Error> {
-        // TODO
-        Ok(Database {})
-    }
-
-    pub fn synchronized_directories(&self) -> Vec<String> {
-        // TODO
-        Vec::new()
-    }
-}
+//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
+//}
 
 #[derive(Debug)]
 pub enum Error {