fix: very much WIP stuff, needs review

Author: ErikBjare

aw-datastore/src/lib.rs

@@ -22,6 +22,7 @@ mod worker;
 pub use self::datastore::DatastoreInstance;
 pub use self::worker::Datastore;
 
+#[derive(Debug, Clone)]
 pub enum DatastoreMethod {
     Memory(),
     File(String),

aw-datastore/src/worker.rs

@@ -118,7 +118,7 @@ impl DatastoreWorker {
 
     fn work_loop(&mut self, method: DatastoreMethod) {
         // Open SQLite connection
-        let mut conn = match method {
+        let mut conn = match &method {
             DatastoreMethod::Memory() => {
                 Connection::open_in_memory().expect("Failed to create in-memory datastore")
             }
@@ -149,13 +149,26 @@ impl DatastoreWorker {
 
         // Start handling and respond to requests
         loop {
+            if self.quit {
+                break;
+            };
+
             let last_commit_time: DateTime<Utc> = Utc::now();
-            let mut transaction = conn
-                .transaction_with_behavior(TransactionBehavior::Immediate)
-                .unwrap();
+            info!("Method '{:?}'", &method);
+            let mut transaction: Transaction =
+                match conn.transaction_with_behavior(TransactionBehavior::Immediate) {
+                    Ok(transaction) => transaction,
+                    Err(err) => {
+                        error!("Unable to start transaction! {:?}", err);
+                        // Wait 1s before retrying
+                        std::thread::sleep(std::time::Duration::from_millis(1000));
+                        continue;
+                    }
+                };
+            transaction.set_drop_behavior(DropBehavior::Commit);
+
             self.uncommited_events = 0;
             self.commit = false;
-            transaction.set_drop_behavior(DropBehavior::Commit);
             loop {
                 let (request, response_sender) = match self.responder.poll() {
                     Ok((req, res_sender)) => (req, res_sender),
@@ -182,9 +195,6 @@ impl DatastoreWorker {
                 Ok(_) => (),
                 Err(err) => panic!("Failed to commit datastore transaction! {}", err),
             }
-            if self.quit {
-                break;
-            };
         }
         info!("DB Worker thread finished");
     }
@@ -196,29 +206,27 @@ impl DatastoreWorker {
         transaction: &Transaction,
     ) -> Result<Response, DatastoreError> {
         match request {
-            Command::CreateBucket(bucket) => match ds.create_bucket(&transaction, bucket) {
+            Command::CreateBucket(bucket) => match ds.create_bucket(transaction, bucket) {
                 Ok(_) => {
                     self.commit = true;
                     Ok(Response::Empty())
                 }
                 Err(e) => Err(e),
             },
-            Command::DeleteBucket(bucketname) => {
-                match ds.delete_bucket(&transaction, &bucketname) {
-                    Ok(_) => {
-                        self.commit = true;
-                        Ok(Response::Empty())
-                    }
-                    Err(e) => Err(e),
+            Command::DeleteBucket(bucketname) => match ds.delete_bucket(transaction, &bucketname) {
+                Ok(_) => {
+                    self.commit = true;
+                    Ok(Response::Empty())
                 }
-            }
+                Err(e) => Err(e),
+            },
             Command::GetBucket(bucketname) => match ds.get_bucket(&bucketname) {
                 Ok(b) => Ok(Response::Bucket(b)),
                 Err(e) => Err(e),
             },
             Command::GetBuckets() => Ok(Response::BucketMap(ds.get_buckets())),
             Command::InsertEvents(bucketname, events) => {
-                match ds.insert_events(&transaction, &bucketname, events) {
+                match ds.insert_events(transaction, &bucketname, events) {
                     Ok(events) => {
                         self.uncommited_events += events.len();
                         self.last_heartbeat.insert(bucketname.to_string(), None); // invalidate last_heartbeat cache
@@ -229,7 +237,7 @@ impl DatastoreWorker {
             }
             Command::Heartbeat(bucketname, event, pulsetime) => {
                 match ds.heartbeat(
-                    &transaction,
+                    transaction,
                     &bucketname,
                     event,
                     pulsetime,
@@ -243,14 +251,14 @@ impl DatastoreWorker {
                 }
             }
             Command::GetEvent(bucketname, event_id) => {
-                match ds.get_event(&transaction, &bucketname, event_id) {
+                match ds.get_event(transaction, &bucketname, event_id) {
                     Ok(el) => Ok(Response::Event(el)),
                     Err(e) => Err(e),
                 }
             }
             Command::GetEvents(bucketname, starttime_opt, endtime_opt, limit_opt) => {
                 match ds.get_events(
-                    &transaction,
+                    transaction,
                     &bucketname,
                     starttime_opt,
                     endtime_opt,
@@ -261,13 +269,13 @@ impl DatastoreWorker {
                 }
             }
             Command::GetEventCount(bucketname, starttime_opt, endtime_opt) => {
-                match ds.get_event_count(&transaction, &bucketname, starttime_opt, endtime_opt) {
+                match ds.get_event_count(transaction, &bucketname, starttime_opt, endtime_opt) {
                     Ok(n) => Ok(Response::Count(n)),
                     Err(e) => Err(e),
                 }
             }
             Command::DeleteEventsById(bucketname, event_ids) => {
-                match ds.delete_events_by_id(&transaction, &bucketname, event_ids) {
+                match ds.delete_events_by_id(transaction, &bucketname, event_ids) {
                     Ok(()) => Ok(Response::Empty()),
                     Err(e) => Err(e),
                 }
@@ -277,22 +285,22 @@ impl DatastoreWorker {
                 Ok(Response::Empty())
             }
             Command::InsertKeyValue(key, data) => {
-                match ds.insert_key_value(&transaction, &key, &data) {
+                match ds.insert_key_value(transaction, &key, &data) {
                     Ok(()) => Ok(Response::Empty()),
                     Err(e) => Err(e),
                 }
             }
-            Command::GetKeyValue(key) => match ds.get_key_value(&transaction, &key) {
+            Command::GetKeyValue(key) => match ds.get_key_value(transaction, &key) {
                 Ok(result) => Ok(Response::KeyValue(result)),
                 Err(e) => Err(e),
             },
             Command::GetKeysStarting(pattern) => {
-                match ds.get_keys_starting(&transaction, &pattern) {
+                match ds.get_keys_starting(transaction, &pattern) {
                     Ok(result) => Ok(Response::StringVec(result)),
                     Err(e) => Err(e),
                 }
             }
-            Command::DeleteKeyValue(key) => match ds.delete_key_value(&transaction, &key) {
+            Command::DeleteKeyValue(key) => match ds.delete_key_value(transaction, &key) {
                 Ok(()) => Ok(Response::Empty()),
                 Err(e) => Err(e),
             },

aw-sync/src/main.rs

@@ -15,7 +15,7 @@ extern crate serde_json;
 
 use std::path::Path;
 
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, Datelike, TimeZone, Utc};
 use clap::{Parser, Subcommand};
 
 use aw_client_rust::AwClient;
@@ -31,38 +31,42 @@ struct Opts {
     #[clap(subcommand)]
     command: Commands,
 
-    /// Host of instance to connect to
+    /// Host of instance to connect to.
     #[clap(long, default_value = "127.0.0.1")]
     host: String,
-    /// Port of instance to connect to
+    /// Port of instance to connect to.
     #[clap(long, default_value = DEFAULT_PORT)]
     port: String,
     /// Convenience option for using the default testing host and port.
     #[clap(long)]
     testing: bool,
-    /// Path to sync directory
-    /// If not specified, exit
+    /// Path to sync directory.
+    /// If not specified, exit.
     #[clap(long)]
     sync_dir: String,
 }
 
 #[derive(Subcommand)]
 enum Commands {
-    /// Clones repos
+    /// Sync subcommand.
+    ///
+    /// Pulls remote buckets then pushes local buckets.
+    /// First pulls remote buckets in the sync directory to the local aw-server.
+    /// Then pushes local buckets from the aw-server to the local sync directory.
     #[clap(arg_required_else_help = true)]
     Sync {
-        /// Date to start syncing from
-        /// If not specified, start from beginning
+        /// Date to start syncing from.
+        /// If not specified, start from beginning.
         /// NOTE: might be unstable, as count cannot be used to verify integrity of sync.
         /// Format: YYYY-MM-DD
         #[clap(long)]
         start_date: Option<String>,
-        /// Specify buckets to sync
-        /// If not specified, all buckets will be synced
+        /// Specify buckets to sync.
+        /// If not specified, all buckets will be synced.
         #[clap(long)]
         buckets: Option<Vec<String>>,
     },
-    /// List buckets and their sync status
+    /// List buckets and their sync status.
     List {},
 }
 
@@ -95,10 +99,13 @@ fn main() -> std::io::Result<()> {
             buckets,
         } => {
             let start: Option<DateTime<Utc>> = start_date.as_ref().map(|date| {
-                let date_copy = date.clone();
-                chrono::DateTime::parse_from_rfc3339(&date_copy)
-                    .unwrap()
-                    .with_timezone(&chrono::Utc)
+                println!("{}", date.clone());
+                chrono::NaiveDate::parse_from_str(&date.clone(), "%Y-%m-%d")
+                    .map(|nd| {
+                        let dt = Utc.ymd(nd.year(), nd.month(), nd.day());
+                        dt.and_hms(0, 0, 0)
+                    })
+                    .expect("Date was not on the format YYYY-MM-DD")
             });
             let sync_spec = sync::SyncSpec {
                 path: sync_directory.to_path_buf(),

aw-sync/src/sync.rs

@@ -13,7 +13,7 @@ use std::fs;
 use std::path::{Path, PathBuf};
 
 use aw_client_rust::AwClient;
-use chrono::{DateTime, Duration, Utc};
+use chrono::{DateTime, Utc};
 
 use aw_datastore::{Datastore, DatastoreError};
 use aw_models::{Bucket, Event};
@@ -159,61 +159,6 @@ fn create_datastore(path: &Path) -> Datastore {
     Datastore::new(pathstr.to_string(), false)
 }
 
-// TODO: Move into tests
-fn setup_test(sync_directory: &Path) -> std::io::Result<Vec<Datastore>> {
-    let mut datastores: Vec<Datastore> = Vec::new();
-    for n in 0..2 {
-        let dspath = sync_directory.join(format!("test-remote-{}.db", n));
-        let ds_ = create_datastore(&dspath);
-        let ds = &ds_ as &dyn AccessMethod;
-
-        // Create a bucket
-        // NOTE: Created with duplicate name to make sure it still works under such conditions
-        let bucket_jsonstr = format!(
-            r#"{{
-                "id": "bucket",
-                "type": "test",
-                "hostname": "device-{}",
-                "client": "test"
-            }}"#,
-            n
-        );
-        let bucket: Bucket = serde_json::from_str(&bucket_jsonstr)?;
-        match ds.create_bucket(&bucket) {
-            Ok(()) => (),
-            Err(e) => match e {
-                DatastoreError::BucketAlreadyExists(_) => {
-                    debug!("bucket already exists, skipping");
-                }
-                e => panic!("woops! {:?}", e),
-            },
-        };
-
-        // Insert some testing events into the bucket
-        let events: Vec<Event> = (0..3)
-            .map(|i| {
-                let timestamp: DateTime<Utc> = Utc::now() + Duration::milliseconds(i * 10);
-                let event_jsonstr = format!(
-                    r#"{{
-                "timestamp": "{}",
-                "duration": 0,
-                "data": {{"test": {} }}
-            }}"#,
-                    timestamp.to_rfc3339(),
-                    i
-                );
-                serde_json::from_str(&event_jsonstr).unwrap()
-            })
-            .collect::<Vec<Event>>();
-
-        ds.insert_events(bucket.id.as_str(), events).unwrap();
-        //let new_eventcount = ds.get_event_count(bucket.id.as_str(), None, None).unwrap();
-        //info!("Eventcount: {:?} ({} new)", new_eventcount, events.len());
-        datastores.push(ds_);
-    }
-    Ok(datastores)
-}
-
 /// Returns the sync-destination bucket for a given bucket, creates it if it doesn't exist.
 fn get_or_create_sync_bucket(
     bucket_from: &Bucket,

aw-sync/tests/sync.rs

@@ -210,4 +210,59 @@ mod sync_tests {
         // Check again that new events were indeed synced
         check_synced_buckets_equal_to_src(&all_buckets_map);
     }
+
+    // TODO: Move into tests
+    fn setup_test(sync_directory: &Path) -> std::io::Result<Vec<Datastore>> {
+        let mut datastores: Vec<Datastore> = Vec::new();
+        for n in 0..2 {
+            let dspath = sync_directory.join(format!("test-remote-{}.db", n));
+            let ds_ = create_datastore(&dspath);
+            let ds = &ds_ as &dyn AccessMethod;
+
+            // Create a bucket
+            // NOTE: Created with duplicate name to make sure it still works under such conditions
+            let bucket_jsonstr = format!(
+                r#"{{
+                    "id": "bucket",
+                    "type": "test",
+                    "hostname": "device-{}",
+                    "client": "test"
+                }}"#,
+                n
+            );
+            let bucket: Bucket = serde_json::from_str(&bucket_jsonstr)?;
+            match ds.create_bucket(&bucket) {
+                Ok(()) => (),
+                Err(e) => match e {
+                    DatastoreError::BucketAlreadyExists(_) => {
+                        debug!("bucket already exists, skipping");
+                    }
+                    e => panic!("woops! {:?}", e),
+                },
+            };
+
+            // Insert some testing events into the bucket
+            let events: Vec<Event> = (0..3)
+                .map(|i| {
+                    let timestamp: DateTime<Utc> = Utc::now() + Duration::milliseconds(i * 10);
+                    let event_jsonstr = format!(
+                        r#"{{
+                            "timestamp": "{}",
+                            "duration": 0,
+                            "data": {{"test": {} }}
+                        }}"#,
+                        timestamp.to_rfc3339(),
+                        i
+                    );
+                    serde_json::from_str(&event_jsonstr).unwrap()
+                })
+                .collect::<Vec<Event>>();
+
+            ds.insert_events(bucket.id.as_str(), events).unwrap();
+            //let new_eventcount = ds.get_event_count(bucket.id.as_str(), None, None).unwrap();
+            //info!("Eventcount: {:?} ({} new)", new_eventcount, events.len());
+            datastores.push(ds_);
+        }
+        Ok(datastores)
+    }
 }