diff --git a/README.md b/README.md index bc11b089..43431d2e 100644 --- a/README.md +++ b/README.md @@ -25,14 +25,14 @@ use seshat::{Database, Event, Profile}; use tempfile::tempdir; let tmpdir = tempdir().unwrap(); -let mut db = Database::new(tmpdir.path()).unwrap(); +let db = Database::new(tmpdir.path()).unwrap(); /// Method to call for every live event that gets received during a sync. fn add_live_event(event: Event, profile: Profile, database: &Database) { database.add_event(event, profile); } /// Method to call on every successful sync after live events were added. -fn on_sync(database: &mut Database) { +fn on_sync(database: &Database) { database.commit().unwrap(); } ``` diff --git a/seshat-node/src/lib.rs b/seshat-node/src/lib.rs index 04347d20..e4850822 100644 --- a/seshat-node/src/lib.rs +++ b/seshat-node/src/lib.rs @@ -85,10 +85,8 @@ impl SeshatRecovery { Ok(c) => match c { Ok(c) => c, Err(e) => { - return cx.throw_type_error(format!( - "Unable to get a database connection {}", - e.to_string() - )) + return cx + .throw_type_error(format!("Unable to get a database connection {}", e)) } }, Err(e) => return cx.throw_type_error(e), @@ -191,10 +189,8 @@ impl Seshat { Ok(c) => match c { Ok(c) => c, Err(e) => { - return cx.throw_type_error(format!( - "Unable to get a database connection {}", - e.to_string() - )) + return cx + .throw_type_error(format!("Unable to get a database connection {}", e)) } }, Err(e) => return cx.throw_type_error(e), @@ -267,8 +263,8 @@ impl Seshat { }; let receiver = { - let db = &mut this.borrow_mut().database; - db.as_mut().map_or_else( + let db = &this.borrow().database; + db.as_ref().map_or_else( || Err(CLOSED_ERROR), |db| { if force { @@ -324,10 +320,8 @@ impl Seshat { Ok(c) => match c { Ok(c) => c, Err(e) => { - return cx.throw_type_error(format!( - "Unable to get a database connection {}", - e.to_string() - )) + return cx + .throw_type_error(format!("Unable to get a database connection {}", e)) } }, Err(e) => return cx.throw_type_error(e), @@ -369,10 +363,8 @@ impl Seshat { Ok(c) => match c { Ok(c) => c, Err(e) => { - return cx.throw_type_error(format!( - "Unable to get a database connection {}", - e.to_string() - )) + return cx + .throw_type_error(format!("Unable to get a database connection {}", e)) } }, Err(e) => return cx.throw_type_error(e), @@ -397,10 +389,8 @@ impl Seshat { Ok(c) => match c { Ok(c) => c, Err(e) => { - return cx.throw_type_error(format!( - "Unable to get a database connection {}", - e.to_string() - )) + return cx + .throw_type_error(format!("Unable to get a database connection {}", e)) } }, Err(e) => return cx.throw_type_error(e), @@ -427,10 +417,8 @@ impl Seshat { Ok(c) => match c { Ok(c) => c, Err(e) => { - return cx.throw_type_error(format!( - "Unable to get a database connection {}", - e.to_string() - )) + return cx + .throw_type_error(format!("Unable to get a database connection {}", e)) } }, Err(e) => return cx.throw_type_error(e), @@ -455,10 +443,8 @@ impl Seshat { Ok(c) => match c { Ok(c) => c, Err(e) => { - return cx.throw_type_error(format!( - "Unable to get a database connection {}", - e.to_string() - )) + return cx + .throw_type_error(format!("Unable to get a database connection {}", e)) } }, Err(e) => return cx.throw_type_error(e), @@ -491,10 +477,10 @@ impl Seshat { }; let ret = { - let db = &mut this.borrow_mut().database; + let db = &this.borrow().database; if wait { - db.as_mut().map_or_else( + db.as_ref().map_or_else( || Err(CLOSED_ERROR), |db| { if force { @@ -505,7 +491,7 @@ impl Seshat { }, ) } else { - db.as_mut().map_or_else( + db.as_ref().map_or_else( || Err(CLOSED_ERROR), |db| { db.commit_no_wait(); @@ -585,7 +571,7 @@ impl Seshat { let searcher = match searcher { Ok(s) => s, - Err(e) => return cx.throw_type_error(e.to_string()), + Err(e) => return cx.throw_type_error(e), }; let task = SearchTask { @@ -697,7 +683,7 @@ impl Seshat { Ok(s) => s, Err(e) => return cx.throw_type_error(e.to_string()), }, - Err(e) => return cx.throw_type_error(e.to_string()), + Err(e) => return cx.throw_type_error(e), }; let task = LoadFileEventsTask { diff --git a/seshat-node/src/tasks.rs b/seshat-node/src/tasks.rs index f542976d..241a623d 100644 --- a/seshat-node/src/tasks.rs +++ b/seshat-node/src/tasks.rs @@ -106,7 +106,7 @@ impl Task for CommitTask { ) -> JsResult<'a, Self::JsEvent> { match result { Ok(_) => Ok(cx.undefined()), - Err(e) => cx.throw_error(format!("Error writing to database: {}", e.to_string())), + Err(e) => cx.throw_error(format!("Error writing to database: {}", e)), } } } @@ -503,7 +503,7 @@ impl Task for DeleteEventTask { ) -> JsResult<'a, Self::JsEvent> { match result { Ok(b) => Ok(cx.boolean(b)), - Err(e) => cx.throw_error(format!("Error deleting an event: {}", e.to_string())), + Err(e) => cx.throw_error(format!("Error deleting an event: {}", e)), } } } @@ -535,10 +535,7 @@ impl Task for ChangePassphraseTask { ) -> JsResult<'a, Self::JsEvent> { match result { Ok(_) => Ok(cx.undefined()), - Err(e) => cx.throw_error(format!( - "Error while changing the passphrase: {}", - e.to_string() - )), + Err(e) => cx.throw_error(format!("Error while changing the passphrase: {}", e)), } } } @@ -566,10 +563,7 @@ impl Task for GetUserVersionTask { let version = cx.number(version as f64); Ok(version) } - Err(e) => cx.throw_error(format!( - "Error while getting the user version: {}", - e.to_string() - )), + Err(e) => cx.throw_error(format!("Error while getting the user version: {}", e)), } } } @@ -595,10 +589,7 @@ impl Task for SetUserVersionTask { ) -> JsResult<'a, Self::JsEvent> { match result { Ok(_) => Ok(cx.undefined()), - Err(e) => cx.throw_error(format!( - "Error while setting the user version: {}", - e.to_string() - )), + Err(e) => cx.throw_error(format!("Error while setting the user version: {}", e)), } } } diff --git a/seshat-node/src/utils.rs b/seshat-node/src/utils.rs index 9ced60a3..2bdfa0a5 100644 --- a/seshat-node/src/utils.rs +++ b/seshat-node/src/utils.rs @@ -212,10 +212,7 @@ pub(crate) fn deserialize_event<'a, C: Context<'a>>( let source: serde_json::Value = match source { Ok(s) => s, Err(e) => { - return cx.throw_type_error(format!( - "Couldn't load the event from the store: {}", - e.to_string() - )) + return cx.throw_type_error(format!("Couldn't load the event from the store: {}", e)) } }; @@ -372,7 +369,7 @@ pub(crate) fn parse_event( "m.room.message" => EventType::Message, "m.room.name" => EventType::Name, "m.room.topic" => EventType::Topic, - e => return cx.throw_type_error(format!("Unsupported event type {e}")), + e => return cx.throw_type_error(format!("Unsupported event type {}", e)), }; let key = match event_type { diff --git a/seshat-node/test/test.js b/seshat-node/test/test.js index 428d4765..e6d99e85 100644 --- a/seshat-node/test/test.js +++ b/seshat-node/test/test.js @@ -363,7 +363,7 @@ describe('Database', function() { it('should not barf on nul bytes in the event', async function() { const db = createDb(); const events = [{event: nulByteEvent, profile: matrixProfileOnlyDisplayName}]; - db.addHistoricEvents(events); + await db.addHistoricEvents(events); await db.commit(true); db.reload(); }); diff --git a/src/database/mod.rs b/src/database/mod.rs index 55f955a2..23f6b55e 100644 --- a/src/database/mod.rs +++ b/src/database/mod.rs @@ -70,6 +70,7 @@ pub(crate) enum ThreadMessage { Write(Sender>, bool), Delete(Sender>, EventId), ShutDown(Sender>), + Reload(Sender>), } impl ThreadMessage { @@ -81,6 +82,7 @@ impl ThreadMessage { ThreadMessage::Write(_, _) => "Write", ThreadMessage::Delete(_, _) => "Delete", ThreadMessage::ShutDown(_) => "ShutDown", + ThreadMessage::Reload(_) => "Reload", } } } @@ -430,6 +432,13 @@ impl Database { let _e = sender.send(ret); return; } + ThreadMessage::Reload(sender) => { + // Since Tokio's mpsc is FIFO, receiving this message ensures all + // previous write operations have been processed. + // + // Same as the previous one, fine to ignore the error on the send. + let _e = sender.send(Ok(())); + } }; } } @@ -497,7 +506,7 @@ impl Database { receiver } - fn commit_helper(&mut self, force: bool) -> Receiver> { + fn commit_helper(&self, force: bool) -> Receiver> { let (sender, receiver): (_, Receiver>) = channel(); self.send_message_to_writer(ThreadMessage::Write(sender, force)); receiver @@ -506,7 +515,7 @@ impl Database { /// Commit the currently queued up events. This method will block. A /// non-blocking version of this method exists in the `commit_no_wait()` /// method. - pub fn commit(&mut self) -> Result<()> { + pub fn commit(&self) -> Result<()> { self.commit_helper(false).recv().unwrap() } @@ -519,14 +528,17 @@ impl Database { /// the `force_commit_no_wait()` method. /// /// This should only be used for testing purposes. - pub fn force_commit(&mut self) -> Result<()> { + pub fn force_commit(&self) -> Result<()> { self.commit_helper(true).recv().unwrap() } /// Reload the database so that a search reflects the state of the last /// commit. Note that this happens automatically and this method should be /// used only in unit tests. - pub fn reload(&mut self) -> Result<()> { + pub fn reload(&self) -> Result<()> { + let (sender, receiver) = channel(); + self.send_message_to_writer(ThreadMessage::Reload(sender)); + receiver.recv().unwrap()?; self.index.reload()?; Ok(()) } @@ -536,7 +548,7 @@ impl Database { /// /// Returns a receiver that will receive an empty message once the commit is /// done. - pub fn commit_no_wait(&mut self) -> Receiver> { + pub fn commit_no_wait(&self) -> Receiver> { self.commit_helper(false) } @@ -549,7 +561,7 @@ impl Database { /// /// Returns a receiver that will receive an empty message once the commit is /// done. - pub fn force_commit_no_wait(&mut self) -> Receiver> { + pub fn force_commit_no_wait(&self) -> Receiver> { self.commit_helper(true) } @@ -735,14 +747,14 @@ fn load_event() { #[test] fn commit_a_write() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); db.commit().unwrap(); } #[test] fn save_the_event_multithreaded() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile); @@ -783,7 +795,7 @@ fn load_a_profile() { #[test] fn load_event_context() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile.clone()); @@ -919,7 +931,7 @@ fn duplicate_empty_profiles() { #[test] fn is_empty() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let connection = db.get_connection().unwrap(); assert!(connection.is_empty().unwrap()); @@ -934,7 +946,7 @@ fn is_empty() { fn encrypted_db() { let tmpdir = tempdir().unwrap(); let db_config = Config::new().set_passphrase("test"); - let mut db = match Database::new_with_config(tmpdir.path(), &db_config) { + let db = match Database::new_with_config(tmpdir.path(), &db_config) { Ok(db) => db, Err(e) => panic!("Coulnd't open encrypted database {}", e), }; @@ -975,7 +987,7 @@ fn encrypted_db() { fn change_passphrase() { let tmpdir = tempdir().unwrap(); let db_config = Config::new().set_passphrase("test"); - let mut db = match Database::new_with_config(tmpdir.path(), &db_config) { + let db = match Database::new_with_config(tmpdir.path(), &db_config) { Ok(db) => db, Err(e) => panic!("Coulnd't open encrypted database {}", e), }; @@ -1018,7 +1030,7 @@ fn change_passphrase() { #[test] fn resume_committing() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); // Check that we don't have any uncommitted events. @@ -1064,7 +1076,7 @@ fn resume_committing() { db = Database::new(tmpdir.path()) } - let mut db = db.unwrap(); + let db = db.unwrap(); // We still have uncommitted events. assert_eq!( @@ -1093,7 +1105,7 @@ fn resume_committing() { #[test] fn delete_uncommitted() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); for i in 1..1000 { @@ -1117,7 +1129,7 @@ fn delete_uncommitted() { #[test] fn stats_getting() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); for i in 0..1000 { @@ -1209,7 +1221,7 @@ fn database_upgrade_v1_2() { #[test] fn delete_an_event() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile.clone()); @@ -1235,7 +1247,7 @@ fn delete_an_event() { drop(db); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); assert_eq!( Database::load_pending_deletion_events(&db.connection.lock().unwrap()) .unwrap() @@ -1285,7 +1297,7 @@ fn add_events_with_null_byte() { #[test] fn is_room_indexed() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let connection = db.get_connection().unwrap(); @@ -1331,7 +1343,7 @@ fn edit_event_removes_original() { use crate::config::SearchConfig; let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); // Add an original message @@ -1388,7 +1400,7 @@ fn original_event_skipped_if_edit_comes_first() { use crate::config::SearchConfig; let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); // Add an edit event FIRST (before the original) diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 74505af1..fd2d0786 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -175,7 +175,7 @@ fn create_db() { #[test] fn save_and_search() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile); @@ -190,7 +190,7 @@ fn save_and_search() { #[test] fn search_with_room() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile); @@ -228,7 +228,7 @@ fn search_with_room() { #[test] fn duplicate_events() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile.clone()); @@ -279,7 +279,7 @@ fn save_and_search_historic_events() { #[test] fn get_size() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); @@ -305,7 +305,7 @@ fn get_size() { #[test] fn add_differing_events() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile.clone()); @@ -324,7 +324,7 @@ fn add_differing_events() { #[test] fn search_with_specific_key() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); let searcher = db.get_searcher(); @@ -370,7 +370,7 @@ fn delete() { fn encrypted_save_and_search() { let tmpdir = tempdir().unwrap(); let db_config = Config::new().set_passphrase("wordpass"); - let mut db = Database::new_with_config(tmpdir.path(), &db_config).unwrap(); + let db = Database::new_with_config(tmpdir.path(), &db_config).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile); @@ -385,7 +385,7 @@ fn encrypted_save_and_search() { #[test] fn load_file_events() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile.clone()); @@ -430,7 +430,7 @@ fn load_file_events() { #[test] fn load_file_events_directions() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile.clone()); @@ -472,7 +472,7 @@ fn load_file_events_directions() { #[test] fn delete_events() { let tmpdir = tempdir().unwrap(); - let mut db = Database::new(tmpdir.path()).unwrap(); + let db = Database::new(tmpdir.path()).unwrap(); let profile = Profile::new("Alice", ""); db.add_event(EVENT.clone(), profile.clone());