Create Mentat FFI and expose observers (#574)

* Tidy up and add txid at beginning of transaction

* Add ffi crate and new_store function

* Add register and unregister observer FFI, Store and Conn functions.
Also add android logging facilities

* Add function for fetching entids for attribute strings

* Add functions for iterating through TxReports

* Add sync to ffi boundary

* Move Extern types from submodule to lib in FFI.
For some reason, if these types are in a submodule, even if they are publically used, the functions inside the FFI are not found in
Android. Works for iOS though. To be investigated later....

* Return to passing TxReports to observer function.
Also, remove some debug

* Expose DateTime and Utc publically

* Use Store in observer tests

Cargo.toml

@@ -20,7 +20,7 @@ default = ["bundled_sqlite3"]
 bundled_sqlite3 = ["rusqlite/bundled"]
 
 [workspace]
-members = ["tools/cli"]
+members = ["tools/cli", "ffi"]
 
 [build-dependencies]
 rustc_version = "0.2"

ffi/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "mentat_ffi"
+version = "0.1.0"
+authors = ["Emily Toop <etoop@mozilla.com>"]
+
+[dependencies.mentat]
+path = ".."

ffi/src/android.rs

@@ -0,0 +1,24 @@
+// Copyright 2018 Mozilla
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use
+// this file except in compliance with the License. You may obtain a copy of the
+// License at http://www.apache.org/licenses/LICENSE-2.0
+// Unless required by applicable law or agreed to in writing, software distributed
+// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+// CONDITIONS OF ANY KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations under the License.
+
+// TODO just use https://github.com/tomaka/android-rs-glue somehow?
+
+use std::os::raw::c_char;
+use std::os::raw::c_int;
+
+// Logging
+pub enum LogLevel {
+    Debug = 3,
+    Info = 4,
+    Warn = 5,
+    Error = 6,
+}
+
+extern { pub fn __android_log_write(prio: c_int, tag: *const c_char, text: *const c_char) -> c_int; }

ffi/src/lib.rs

@@ -0,0 +1,180 @@
+// Copyright 2018 Mozilla
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use
+// this file except in compliance with the License. You may obtain a copy of the
+// License at http://www.apache.org/licenses/LICENSE-2.0
+// Unless required by applicable law or agreed to in writing, software distributed
+// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+// CONDITIONS OF ANY KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations under the License.
+
+extern crate mentat;
+
+use std::collections::{
+    BTreeSet,
+};
+use std::os::raw::{
+    c_char,
+    c_int,
+};
+use std::slice;
+use std::sync::{
+    Arc,
+};
+
+pub use mentat::{
+    Entid,
+    HasSchema,
+    NamespacedKeyword,
+    Store,
+    Syncable,
+    TxObserver,
+};
+
+pub use mentat::errors::{
+    Result,
+};
+
+pub mod android;
+pub mod utils;
+
+pub use utils::strings::{
+    c_char_to_string,
+    string_to_c_char,
+    str_to_c_char,
+};
+
+use utils::log;
+
+#[repr(C)]
+#[derive(Debug, Clone)]
+pub struct ExternTxReport {
+    pub txid: Entid,
+    pub changes: Box<[Entid]>,
+    pub changes_len: usize,
+}
+
+#[repr(C)]
+#[derive(Debug)]
+pub struct ExternTxReportList {
+    pub reports: Box<[ExternTxReport]>,
+    pub len: usize,
+}
+
+#[repr(C)]
+pub struct ExternResult {
+    pub error: *const c_char,
+}
+
+impl From<Result<()>> for ExternResult {
+    fn from(result: Result<()>) -> Self {
+        match result {
+            Ok(_) => {
+                ExternResult {
+                    error: std::ptr::null(),
+                }
+            },
+            Err(e) => {
+                ExternResult {
+                    error: string_to_c_char(e.description().into())
+                }
+            }
+        }
+    }
+}
+
+// A store cannot be opened twice to the same location.
+// Once created, the reference to the store is held by the caller and not Rust,
+// therefore the caller is responsible for calling `store_destroy` to release the memory
+// used by the Store in order to avoid a memory leak.
+// TODO: Start returning `ExternResult`s rather than crashing on error.
+#[no_mangle]
+pub extern "C" fn store_open(uri: *const c_char) -> *mut Store {
+    let uri = c_char_to_string(uri);
+    let store = Store::open(&uri).expect("expected a store");
+    Box::into_raw(Box::new(store))
+}
+
+// Reclaim the memory for the provided Store and drop, therefore releasing it.
+#[no_mangle]
+pub unsafe extern "C" fn store_destroy(store: *mut Store) {
+    let _ = Box::from_raw(store);
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn store_register_observer(store: *mut Store,
+                                                   key: *const c_char,
+                                            attributes: *const Entid,
+                                        attributes_len: usize,
+                                              callback: extern fn(key: *const c_char, reports: &ExternTxReportList)) {
+    let store = &mut*store;
+    let mut attribute_set = BTreeSet::new();
+    let slice = slice::from_raw_parts(attributes, attributes_len);
+    attribute_set.extend(slice.iter());
+    let key = c_char_to_string(key);
+    let tx_observer = Arc::new(TxObserver::new(attribute_set, move |obs_key, batch| {
+        log::d(&format!("Calling observer registered for {:?}, batch: {:?}", obs_key, batch));
+        let extern_reports: Vec<ExternTxReport> = batch.into_iter().map(|(tx_id, changes)| {
+            let changes: Vec<Entid> = changes.into_iter().map(|i|*i).collect();
+            let len = changes.len();
+            ExternTxReport {
+                txid: *tx_id,
+                changes: changes.into_boxed_slice(),
+                changes_len: len,
+            }
+        }).collect();
+        let len = extern_reports.len();
+        let reports = ExternTxReportList {
+            reports: extern_reports.into_boxed_slice(),
+            len: len,
+        };
+        callback(str_to_c_char(obs_key), &reports);
+    }));
+    store.register_observer(key, tx_observer);
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn store_unregister_observer(store: *mut Store, key: *const c_char) {
+    let store = &mut*store;
+    let key = c_char_to_string(key);
+    log::d(&format!("Unregistering observer for key: {:?}", key));
+    store.unregister_observer(&key);
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn store_entid_for_attribute(store: *mut Store, attr: *const c_char) -> Entid {
+    let store = &mut*store;
+    let mut keyword_string = c_char_to_string(attr);
+    let attr_name = keyword_string.split_off(1);
+    let parts: Vec<&str> = attr_name.split("/").collect();
+    let kw = NamespacedKeyword::new(parts[0], parts[1]);
+    let conn = store.conn();
+    let current_schema = conn.current_schema();
+    let got_entid = current_schema.get_entid(&kw);
+    let entid = got_entid.unwrap();
+    entid.into()
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn tx_report_list_entry_at(tx_report_list: *mut ExternTxReportList, index: c_int) -> *const ExternTxReport {
+    let tx_report_list = &*tx_report_list;
+    let index = index as usize;
+    let report = Box::new(tx_report_list.reports[index].clone());
+    Box::into_raw(report)
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn changelist_entry_at(tx_report: *mut ExternTxReport, index: c_int) -> Entid {
+    let tx_report = &*tx_report;
+    let index = index as usize;
+    tx_report.changes[index].clone()
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn store_sync(store: *mut Store, user_uuid: *const c_char, server_uri: *const c_char) -> *mut ExternResult {
+    let store = &mut*store;
+    let user_uuid = c_char_to_string(user_uuid);
+    let server_uri = c_char_to_string(server_uri);
+    let res = store.sync(&server_uri, &user_uuid);
+    Box::into_raw(Box::new(res.into()))
+}

ffi/src/utils.rs

@@ -0,0 +1,60 @@
+// Copyright 2018 Mozilla
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use
+// this file except in compliance with the License. You may obtain a copy of the
+// License at http://www.apache.org/licenses/LICENSE-2.0
+// Unless required by applicable law or agreed to in writing, software distributed
+// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+// CONDITIONS OF ANY KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations under the License.
+
+pub mod strings {
+    use std::os::raw::c_char;
+    use std::ffi::{
+        CString,
+        CStr
+    };
+
+    pub fn c_char_to_string(cchar: *const c_char) -> String {
+        let c_str = unsafe { CStr::from_ptr(cchar) };
+        let r_str = match c_str.to_str() {
+            Err(_) => "",
+            Ok(string) => string,
+        };
+        r_str.to_string()
+    }
+
+    pub fn string_to_c_char(r_string: String) -> *mut c_char {
+        CString::new(r_string).unwrap().into_raw()
+    }
+
+    pub fn str_to_c_char(r_string: &str) -> *mut c_char {
+        string_to_c_char(r_string.to_string())
+    }
+}
+
+pub mod log {
+    #[cfg(all(target_os="android", not(test)))]
+    use std::ffi::CString;
+
+    #[cfg(all(target_os="android", not(test)))]
+    use android;
+
+    // TODO far from ideal. And, we might actually want to println in tests.
+    #[cfg(all(not(target_os="android"), not(target_os="ios")))]
+    pub fn d(_: &str) {}
+
+    #[cfg(all(target_os="ios", not(test)))]
+    pub fn d(message: &str) {
+        eprintln!("{}", message);
+    }
+
+    #[cfg(all(target_os="android", not(test)))]
+    pub fn d(message: &str) {
+        let message = CString::new(message).unwrap();
+        let message = message.as_ptr();
+        let tag = CString::new("Mentat").unwrap();
+        let tag = tag.as_ptr();
+        unsafe { android::__android_log_write(android::ANDROID_LOG_DEBUG, tag, message) };
+    }
+}

src/conn.rs

@@ -538,6 +538,11 @@ impl Store {
     pub fn sqlite_mut(&mut self) -> &mut rusqlite::Connection {
         &mut self.sqlite
     }
+
+    #[cfg(test)]
+    pub fn is_registered_as_observer(&self, key: &String) -> bool {
+        self.conn.tx_observer_service.lock().unwrap().is_registered(key)
+    }
 }
 
 impl Store {
@@ -565,6 +570,14 @@ impl Store {
                         direction,
                         CacheAction::Register)
     }
+
+    pub fn register_observer(&mut self, key: String, observer: Arc<TxObserver>) {
+        self.conn.register_observer(key, observer);
+    }
+
+    pub fn unregister_observer(&mut self, key: &String) {
+        self.conn.unregister_observer(key);
+    }
 }
 
 impl Queryable for Store {
@@ -623,11 +636,6 @@ impl Conn {
         }
     }
 
-    #[cfg(test)]
-    pub fn is_registered_as_observer(&self, key: &String) -> bool {
-        self.tx_observer_service.lock().unwrap().is_registered(key)
-    }
-
     /// Prepare the provided SQLite handle for use as a Mentat store. Creates tables but
     /// _does not_ write the bootstrap schema. This constructor should only be used by
     /// consumers that expect to populate raw transaction data themselves.
@@ -1470,8 +1478,7 @@ mod tests {
     }
 
     fn test_register_observer() {
-        let mut sqlite = db::new_connection("").unwrap();
-        let mut conn = Conn::connect(&mut sqlite).unwrap();
+        let mut conn = Store::open("").unwrap();
 
         let key = "Test Observer".to_string();
         let tx_observer = TxObserver::new(BTreeSet::new(), move |_obs_key, _batch| {});
@@ -1482,8 +1489,7 @@ mod tests {
 
     #[test]
     fn test_deregister_observer() {
-        let mut sqlite = db::new_connection("").unwrap();
-        let mut conn = Conn::connect(&mut sqlite).unwrap();
+        let mut conn = Store::open("").unwrap();
 
         let key = "Test Observer".to_string();
 
@@ -1497,9 +1503,9 @@ mod tests {
         assert!(!conn.is_registered_as_observer(&key));
     }
 
-    fn add_schema(conn: &mut Conn, mut sqlite: &mut rusqlite::Connection) {
+    fn add_schema(conn: &mut Store) {
         // transact some schema
-        let mut in_progress = conn.begin_transaction(&mut sqlite).expect("expected in progress");
+        let mut in_progress = conn.begin_transaction().expect("expected in progress");
         in_progress.ensure_vocabulary(&Definition {
             name: kw!(:todo/items),
             version: 1,
@@ -1549,12 +1555,11 @@ mod tests {
 
     #[test]
     fn test_observer_notified_on_registered_change() {
-        let mut sqlite = db::new_connection("").unwrap();
-        let mut conn = Conn::connect(&mut sqlite).unwrap();
-        add_schema(&mut conn, &mut sqlite);
+        let mut conn = Store::open("").unwrap();
+        add_schema(&mut conn);
 
-        let name_entid: Entid = conn.current_schema().get_entid(&kw!(:todo/name)).expect("entid to exist for name").into();
-        let date_entid: Entid = conn.current_schema().get_entid(&kw!(:todo/completion_date)).expect("entid to exist for completion_date").into();
+        let name_entid: Entid = conn.conn().current_schema().get_entid(&kw!(:todo/name)).expect("entid to exist for name").into();
+        let date_entid: Entid = conn.conn().current_schema().get_entid(&kw!(:todo/completion_date)).expect("entid to exist for completion_date").into();
         let mut registered_attrs = BTreeSet::new();
         registered_attrs.insert(name_entid.clone());
         registered_attrs.insert(date_entid.clone());
@@ -1586,9 +1591,9 @@ mod tests {
 
         let mut tx_ids = Vec::new();
         let mut changesets = Vec::new();
-        let uuid_entid: Entid = conn.current_schema().get_entid(&kw!(:todo/uuid)).expect("entid to exist for name").into();
+        let uuid_entid: Entid = conn.conn().current_schema().get_entid(&kw!(:todo/uuid)).expect("entid to exist for name").into();
         {
-            let mut in_progress = conn.begin_transaction(&mut sqlite).expect("expected transaction");
+            let mut in_progress = conn.begin_transaction().expect("expected transaction");
             for i in 0..3 {
                 let mut changeset = BTreeSet::new();
                 let name = format!("todo{}", i);
@@ -1626,12 +1631,11 @@ mod tests {
 
     #[test]
     fn test_observer_not_notified_on_unregistered_change() {
-        let mut sqlite = db::new_connection("").unwrap();
-        let mut conn = Conn::connect(&mut sqlite).unwrap();
-        add_schema(&mut conn, &mut sqlite);
+        let mut conn = Store::open("").unwrap();
+        add_schema(&mut conn);
 
-        let name_entid: Entid = conn.current_schema().get_entid(&kw!(:todo/name)).expect("entid to exist for name").into();
-        let date_entid: Entid = conn.current_schema().get_entid(&kw!(:todo/completion_date)).expect("entid to exist for completion_date").into();
+        let name_entid: Entid = conn.conn().current_schema().get_entid(&kw!(:todo/name)).expect("entid to exist for name").into();
+        let date_entid: Entid = conn.conn().current_schema().get_entid(&kw!(:todo/completion_date)).expect("entid to exist for completion_date").into();
         let mut registered_attrs = BTreeSet::new();
         registered_attrs.insert(name_entid.clone());
         registered_attrs.insert(date_entid.clone());
@@ -1662,7 +1666,7 @@ mod tests {
         let tx_ids = Vec::<Entid>::new();
         let changesets = Vec::<BTreeSet<Entid>>::new();
         {
-            let mut in_progress = conn.begin_transaction(&mut sqlite).expect("expected transaction");
+            let mut in_progress = conn.begin_transaction().expect("expected transaction");
             for i in 0..3 {
                 let name = format!("label{}", i);
                 let mut builder = in_progress.builder().describe_tempid(&name);

src/lib.rs

@@ -36,12 +36,14 @@ extern crate mentat_tx_parser;
 pub use mentat_core::{
     Attribute,
     Entid,
+    DateTime,
     HasSchema,
     KnownEntid,
     NamespacedKeyword,
     Schema,
     TypedValue,
     Uuid,
+    Utc,
     ValueType,
 };
 
@@ -52,6 +54,7 @@ pub use mentat_query::{
 pub use mentat_db::{
     CORE_SCHEMA_VERSION,
     DB_SCHEMA_CORE,
+    TxObserver,
     TxReport,
     new_connection,
 };