fix: even more very much WIP, needs even more testing

Author: ErikBjare

aw-datastore/src/datastore.rs

@@ -6,6 +6,8 @@ use chrono::NaiveDateTime;
 use chrono::Utc;
 
 use rusqlite::Connection;
+use rusqlite::Transaction;
+use rusqlite::TransactionBehavior;
 
 use serde_json::value::Value;
 

aw-datastore/src/worker.rs

@@ -53,6 +53,8 @@ pub enum Response {
     Count(i64),
     KeyValue(KeyValue),
     StringVec(Vec<String>),
+    // Used to indicate that no response should occur at all (not even an empty one)
+    NoResponse(),
 }
 
 #[allow(clippy::large_enum_variant)]
@@ -78,6 +80,7 @@ pub enum Command {
     GetKeyValue(String),
     GetKeysStarting(String),
     DeleteKeyValue(String),
+    Close(),
 }
 
 fn _unwrap_response(
@@ -149,12 +152,7 @@ impl DatastoreWorker {
 
         // Start handling and respond to requests
         loop {
-            if self.quit {
-                break;
-            };
-
             let last_commit_time: DateTime<Utc> = Utc::now();
-            info!("Method '{:?}'", &method);
             let mut transaction: Transaction =
                 match conn.transaction_with_behavior(TransactionBehavior::Immediate) {
                     Ok(transaction) => transaction,
@@ -179,8 +177,13 @@ impl DatastoreWorker {
                         break;
                     }
                 };
-                let response = self.handle_request(request, &mut ds, &transaction);
-                response_sender.respond(response);
+                let response = self.handle_request(request, &mut ds);
+                match response {
+                    // The empty response is given by commands like close(), which should
+                    // not be responded to, as the requester might have disappeared.
+                    Ok(Response::NoResponse()) => (),
+                    _ => response_sender.respond(response),
+                }
                 let now: DateTime<Utc> = Utc::now();
                 let commit_interval_passed: bool = (now - last_commit_time) > Duration::seconds(15);
                 if self.commit || commit_interval_passed || self.uncommited_events > 100 {
@@ -195,6 +198,9 @@ impl DatastoreWorker {
                 Ok(_) => (),
                 Err(err) => panic!("Failed to commit datastore transaction! {}", err),
             }
+            if self.quit {
+                break;
+            };
         }
         info!("DB Worker thread finished");
     }
@@ -524,4 +530,10 @@ impl Datastore {
             Err(e) => Err(e),
         }
     }
+
+    // TODO: Should this block until worker has stopped?
+    pub fn close(&self) {
+        info!("Sending close request to database");
+        self.requester.request(Command::Close()).unwrap();
+    }
 }

aw-sync/src/accessmethod.rs

@@ -22,6 +22,7 @@ pub trait AccessMethod: std::fmt::Debug {
     fn insert_events(&self, bucket_id: &str, events: Vec<Event>) -> Result<Vec<Event>, String>;
     fn get_event_count(&self, bucket_id: &str) -> Result<i64, String>;
     fn heartbeat(&self, bucket_id: &str, event: Event, duration: f64) -> Result<(), String>;
+    fn close(&self);
 }
 
 impl AccessMethod for Datastore {
@@ -58,6 +59,9 @@ impl AccessMethod for Datastore {
     fn get_event_count(&self, bucket_id: &str) -> Result<i64, String> {
         Ok(self.get_event_count(bucket_id, None, None).unwrap())
     }
+    fn close(&self) {
+        self.close();
+    }
 }
 
 impl AccessMethod for AwClient {
@@ -105,4 +109,7 @@ impl AccessMethod for AwClient {
         self.heartbeat(bucket_id, &event, duration)
             .map_err(|e| format!("{:?}", e))
     }
+    fn close(&self) {
+        // NOP
+    }
 }

aw-sync/src/sync.rs

@@ -58,6 +58,7 @@ pub fn sync_run(client: AwClient, sync_spec: &SyncSpec) -> Result<(), String> {
         .map(|p| p.as_path())
         .map(create_datastore)
         .collect();
+    info!("Remote datastores: {:?}", ds_remotes);
 
     // Pull
     info!("Pulling...");
@@ -75,11 +76,19 @@ pub fn sync_run(client: AwClient, sync_spec: &SyncSpec) -> Result<(), String> {
         sync_spec,
     );
 
+    // Close open database connections
+    for ds_from in &ds_remotes {
+        ds_from.close();
+    }
+    ds_localremote.close();
+
+    // Will fail if db connections not closed (as it will open them again)
     list_buckets(&client, sync_spec.path.as_path());
 
     Ok(())
 }
 
+#[allow(dead_code)]
 pub fn list_buckets(client: &AwClient, sync_directory: &Path) {
     let ds_localremote = setup_local_remote(client, sync_directory).unwrap();
 
@@ -144,12 +153,12 @@ fn find_remotes_nonlocal(sync_directory: &Path, device_id: &str) -> Vec<PathBuf>
     remotes_all
         .into_iter()
         .filter(|path| {
-            !path
+            !(path
                 .clone()
                 .into_os_string()
                 .into_string()
                 .unwrap()
-                .contains(device_id)
+                .contains(device_id))
         })
         .collect()
 }

aw-sync/tests/sync.rs

@@ -211,7 +211,7 @@ mod sync_tests {
         check_synced_buckets_equal_to_src(&all_buckets_map);
     }
 
-    // TODO: Move into tests
+    // TODO: Find a way to reuse this (previously used in an integration test)
     fn setup_test(sync_directory: &Path) -> std::io::Result<Vec<Datastore>> {
         let mut datastores: Vec<Datastore> = Vec::new();
         for n in 0..2 {