Simplify.

This has a watcher collect txid -> AttributeSet mappings each time a
transact occurs. On commit we retrieve those mappings and hand them over
to the observer service, which filters them and packages them up for
dispatch.

Cargo.toml

@@ -29,6 +29,7 @@ rustc_version = "0.2"
 chrono = "0.4"
 error-chain = { git = "https://github.com/rnewman/error-chain", branch = "rnewman/sync" }
 lazy_static = "0.2"
+smallvec = "0.6"
 time = "0.1"
 uuid = "0.5"
 

db/Cargo.toml

@@ -5,10 +5,12 @@ workspace = ".."
 
 [dependencies]
 error-chain = { git = "https://github.com/rnewman/error-chain", branch = "rnewman/sync" }
+indexmap = "0.4"
 itertools = "0.7"
 lazy_static = "0.2"
 num = "0.1"
 ordered-float = "0.5"
+smallvec = "0.6"
 time = "0.1"
 
 [dependencies.rusqlite]

db/src/cache.rs

@@ -1402,7 +1402,7 @@ impl<'a> TransactWatcher for InProgressCacheTransactWatcher<'a> {
         }
     }
 
-    fn done(&mut self, schema: &Schema) -> Result<()> {
+    fn done(&mut self, _t: &Entid, schema: &Schema) -> Result<()> {
         // Oh, I wish we had impl trait. Without it we have a six-line type signature if we
         // try to break this out as a helper function.
         let collected_retractions = mem::replace(&mut self.collected_retractions, Default::default());

db/src/lib.rs

@@ -13,6 +13,7 @@
 
 #[macro_use]
 extern crate error_chain;
+extern crate indexmap;
 extern crate itertools;
 
 #[macro_use]
@@ -20,6 +21,7 @@ extern crate lazy_static;
 
 extern crate num;
 extern crate rusqlite;
+extern crate smallvec;
 extern crate tabwriter;
 extern crate time;
 
@@ -45,6 +47,7 @@ pub mod errors;
 pub mod internal_types;    // pub because we need them for building entities programmatically.
 mod metadata;
 mod schema;
+pub mod tx_observer;
 mod watcher;
 mod tx;
 pub mod types;
@@ -85,7 +88,14 @@ pub use tx::{
     transact_terms,
 };
 
+pub use tx_observer::{
+    InProgressObserverTransactWatcher,
+    TxObservationService,
+    TxObserver,
+};
+
 pub use types::{
+    AttributeSet,
     DB,
     PartitionMap,
     TxReport,

db/src/tx.rs

@@ -51,8 +51,9 @@ use std::collections::{
     BTreeSet,
     VecDeque,
 };
-
+use std::rc::{
-use std::rc::Rc;
+    Rc,
+};
 
 use db;
 use db::{
@@ -106,6 +107,7 @@ use schema::{
 };
 use types::{
     Attribute,
+    AttributeSet,
     AVPair,
     AVMap,
     Entid,
@@ -527,8 +529,7 @@ impl<'conn, 'a, W> Tx<'conn, 'a, W> where W: TransactWatcher {
     /// This approach is explained in https://github.com/mozilla/mentat/wiki/Transacting.
     // TODO: move this to the transactor layer.
     pub fn transact_entities<I>(&mut self, entities: I) -> Result<TxReport>
-    where I: IntoIterator<Item=Entity>,
+    where I: IntoIterator<Item=Entity> {
-          W: TransactWatcher {
         // Pipeline stage 1: entities -> terms with tempids and lookup refs.
         let (terms_with_temp_ids_and_lookup_refs, tempid_set, lookup_ref_set) = self.entities_into_terms_with_temp_ids_and_lookup_refs(entities)?;
 
@@ -542,8 +543,7 @@ impl<'conn, 'a, W> Tx<'conn, 'a, W> where W: TransactWatcher {
     }
 
     pub fn transact_simple_terms<I>(&mut self, terms: I, tempid_set: InternSet<TempId>) -> Result<TxReport>
-    where I: IntoIterator<Item=TermWithTempIds>,
+    where I: IntoIterator<Item=TermWithTempIds> {
-          W: TransactWatcher {
         // TODO: push these into an internal transaction report?
         let mut tempids: BTreeMap<TempId, KnownEntid> = BTreeMap::default();
 
@@ -614,7 +614,9 @@ impl<'conn, 'a, W> Tx<'conn, 'a, W> where W: TransactWatcher {
                                                     final_populations.allocated,
                                                     inert_terms.into_iter().map(|term| term.unwrap()).collect()].concat();
 
+
         let tx_instant;
+        let mut affected_attrs = AttributeSet::new();
 
         { // TODO: Don't use this block to scope borrowing the schema; instead, extract a helper function.
 
@@ -669,6 +671,8 @@ impl<'conn, 'a, W> Tx<'conn, 'a, W> where W: TransactWatcher {
                     }
 
                     self.watcher.datom(op, e, a, &v);
+                    // TODO: Create something like a watcher to do this for us.
+                    affected_attrs.insert(a);
 
                     let reduced = (e, a, attribute, v, added);
                     match (attribute.fulltext, attribute.multival) {
@@ -710,7 +714,7 @@ impl<'conn, 'a, W> Tx<'conn, 'a, W> where W: TransactWatcher {
         }
 
         db::update_partition_map(self.store, &self.partition_map)?;
-        self.watcher.done(self.schema)?;
+        self.watcher.done(&self.tx_id, self.schema)?;
 
         if tx_might_update_metadata {
             // Extract changes to metadata from the store.
@@ -747,7 +751,6 @@ fn start_tx<'conn, 'a, W>(conn: &'conn rusqlite::Connection,
                        watcher: W) -> Result<Tx<'conn, 'a, W>>
     where W: TransactWatcher {
     let tx_id = partition_map.allocate_entid(":db.part/tx");
-
     conn.begin_tx_application()?;
 
     Ok(Tx::new(conn, partition_map, schema_for_mutation, schema, watcher, tx_id))

db/src/tx_observer.rs

@@ -0,0 +1,208 @@
+// 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.
+
+use std::sync::{
+    Arc,
+    Weak,
+};
+
+use std::sync::mpsc::{
+    channel,
+    Receiver,
+    RecvError,
+    Sender,
+};
+
+use std::thread;
+
+use indexmap::{
+    IndexMap,
+};
+
+use mentat_core::{
+    Entid,
+    Schema,
+    TypedValue,
+};
+
+use mentat_tx::entities::{
+    OpType,
+};
+
+use errors::{
+    Result,
+};
+
+use types::{
+    AttributeSet,
+};
+
+use watcher::TransactWatcher;
+
+pub struct TxObserver {
+    notify_fn: Arc<Box<Fn(&str, IndexMap<&Entid, &AttributeSet>) + Send + Sync>>,
+    attributes: AttributeSet,
+}
+
+impl TxObserver {
+    pub fn new<F>(attributes: AttributeSet, notify_fn: F) -> TxObserver where F: Fn(&str, IndexMap<&Entid, &AttributeSet>) + 'static + Send + Sync {
+        TxObserver {
+            notify_fn: Arc::new(Box::new(notify_fn)),
+            attributes,
+        }
+    }
+
+    pub fn applicable_reports<'r>(&self, reports: &'r IndexMap<Entid, AttributeSet>) -> IndexMap<&'r Entid, &'r AttributeSet> {
+        reports.into_iter()
+               .filter(|&(_txid, attrs)| !self.attributes.is_disjoint(attrs))
+               .collect()
+    }
+
+    fn notify(&self, key: &str, reports: IndexMap<&Entid, &AttributeSet>) {
+        (*self.notify_fn)(key, reports);
+    }
+}
+
+pub trait Command {
+    fn execute(&mut self);
+}
+
+pub struct TxCommand {
+    reports: IndexMap<Entid, AttributeSet>,
+    observers: Weak<IndexMap<String, Arc<TxObserver>>>,
+}
+
+impl TxCommand {
+    fn new(observers: &Arc<IndexMap<String, Arc<TxObserver>>>, reports: IndexMap<Entid, AttributeSet>) -> Self {
+        TxCommand {
+            reports,
+            observers: Arc::downgrade(observers),
+        }
+    }
+}
+
+impl Command for TxCommand {
+    fn execute(&mut self) {
+        self.observers.upgrade().map(|observers| {
+            for (key, observer) in observers.iter() {
+                let applicable_reports = observer.applicable_reports(&self.reports);
+                if !applicable_reports.is_empty() {
+                    observer.notify(&key, applicable_reports);
+                }
+            }
+        });
+    }
+}
+
+pub struct TxObservationService {
+    observers: Arc<IndexMap<String, Arc<TxObserver>>>,
+    executor: Option<Sender<Box<Command + Send>>>,
+}
+
+impl TxObservationService {
+    pub fn new() -> Self {
+        TxObservationService {
+            observers: Arc::new(IndexMap::new()),
+            executor: None,
+        }
+    }
+
+    // For testing purposes
+    pub fn is_registered(&self, key: &String) -> bool {
+        self.observers.contains_key(key)
+    }
+
+    pub fn register(&mut self, key: String, observer: Arc<TxObserver>) {
+        Arc::make_mut(&mut self.observers).insert(key, observer);
+    }
+
+    pub fn deregister(&mut self, key: &String) {
+        Arc::make_mut(&mut self.observers).remove(key);
+    }
+
+    pub fn has_observers(&self) -> bool {
+        !self.observers.is_empty()
+    }
+
+    pub fn in_progress_did_commit(&mut self, txes: IndexMap<Entid, AttributeSet>) {
+        let executor = self.executor.get_or_insert_with(|| {
+            let (tx, rx): (Sender<Box<Command + Send>>, Receiver<Box<Command + Send>>) = channel();
+            let mut worker = CommandExecutor::new(rx);
+
+            thread::spawn(move || {
+                worker.main();
+            });
+
+            tx
+        });
+
+        let cmd = Box::new(TxCommand::new(&self.observers, txes));
+        executor.send(cmd).unwrap();
+    }
+}
+
+impl Drop for TxObservationService {
+    fn drop(&mut self) {
+        self.executor = None;
+    }
+}
+
+pub struct InProgressObserverTransactWatcher {
+    collected_attributes: AttributeSet,
+    pub txes: IndexMap<Entid, AttributeSet>,
+}
+
+impl InProgressObserverTransactWatcher {
+    pub fn new() -> InProgressObserverTransactWatcher {
+        InProgressObserverTransactWatcher {
+            collected_attributes: Default::default(),
+            txes: Default::default(),
+        }
+    }
+}
+
+impl TransactWatcher for InProgressObserverTransactWatcher {
+    fn datom(&mut self, _op: OpType, _e: Entid, a: Entid, _v: &TypedValue) {
+        self.collected_attributes.insert(a);
+    }
+
+    fn done(&mut self, t: &Entid, _schema: &Schema) -> Result<()> {
+        let collected_attributes = ::std::mem::replace(&mut self.collected_attributes, Default::default());
+        self.txes.insert(*t, collected_attributes);
+        Ok(())
+    }
+}
+
+struct CommandExecutor {
+    receiver: Receiver<Box<Command + Send>>,
+}
+
+impl CommandExecutor {
+    fn new(rx: Receiver<Box<Command + Send>>) -> Self {
+        CommandExecutor {
+            receiver: rx,
+        }
+    }
+
+    fn main(&mut self) {
+        loop {
+            match self.receiver.recv() {
+                Err(RecvError) => {
+                    eprintln!("Disconnected, terminating CommandExecutor");
+                    return
+                },
+
+                Ok(mut cmd) => {
+                    cmd.execute()
+                },
+            }
+        }
+    }
+}

db/src/types.rs

@@ -11,19 +11,24 @@
 #![allow(dead_code)]
 
 use std::collections::HashMap;
-use std::collections::BTreeMap;
+use std::collections::{
+    BTreeMap,
+    BTreeSet,
+};
+
+use smallvec::SmallVec;
 
 extern crate mentat_core;
 
 pub use self::mentat_core::{
-    DateTime,
-    Entid,
-    ValueType,
-    TypedValue,
     Attribute,
     AttributeBitFlags,
+    DateTime,
+    Entid,
     Schema,
+    TypedValue,
     Utc,
+    ValueType,
 };
 
 /// Represents one partition of the entid space.
@@ -82,6 +87,11 @@ pub type AVPair = (Entid, TypedValue);
 /// Used to resolve lookup-refs and upserts.
 pub type AVMap<'a> = HashMap<&'a AVPair, Entid>;
 
+// represents a set of entids that are correspond to attributes
+pub type AttributeSet = BTreeSet<Entid>;
+
+pub type AccumulatedTxids = SmallVec<[Entid; 4]>;
+
 /// A transaction report summarizes an applied transaction.
 #[derive(Clone, Debug, Eq, Hash, Ord, PartialOrd, PartialEq)]
 pub struct TxReport {

db/src/watcher.rs

@@ -38,7 +38,7 @@ pub trait TransactWatcher {
     /// Called with the schema _prior to_ the transact -- any attributes or
     /// attribute changes transacted during this transact are not reflected in
     /// the schema.
-    fn done(&mut self, schema: &Schema) -> Result<()>;
+    fn done(&mut self, t: &Entid, schema: &Schema) -> Result<()>;
 }
 
 pub struct NullWatcher();
@@ -47,7 +47,7 @@ impl TransactWatcher for NullWatcher {
     fn datom(&mut self, _op: OpType, _e: Entid, _a: Entid, _v: &TypedValue) {
     }
 
-    fn done(&mut self, _schema: &Schema) -> Result<()> {
+    fn done(&mut self, _t: &Entid, _schema: &Schema) -> Result<()> {
         Ok(())
     }
 }

src/conn.rs

@@ -48,6 +48,7 @@ use mentat_core::{
 use mentat_core::intern_set::InternSet;
 
 use mentat_db::cache::{
+    InProgressCacheTransactWatcher,
     InProgressSQLiteAttributeCache,
     SQLiteAttributeCache,
 };
@@ -56,15 +57,26 @@ use mentat_db::db;
 use mentat_db::{
     transact,
     transact_terms,
+    InProgressObserverTransactWatcher,
     PartitionMap,
+    TransactWatcher,
+    TxObservationService,
+    TxObserver,
     TxReport,
 };
 
+use mentat_db::types::{
+    AccumulatedTxids,
+};
+
 use mentat_db::internal_types::TermWithTempIds;
 
 use mentat_tx;
 
-use mentat_tx::entities::TempId;
+use mentat_tx::entities::{
+    TempId,
+    OpType,
+};
 
 use mentat_tx_parser;
 
@@ -140,6 +152,7 @@ pub struct Conn {
 
     // TODO: maintain cache of query plans that could be shared across threads and invalidated when
     // the schema changes. #315.
+    tx_observer_service: Mutex<TxObservationService>,
 }
 
 /// A convenience wrapper around a single SQLite connection and a Conn. This is suitable
@@ -202,10 +215,11 @@ pub struct InProgress<'a, 'c> {
     generation: u64,
     partition_map: PartitionMap,
     schema: Schema,
-
     cache: InProgressSQLiteAttributeCache,
-
     use_caching: bool,
+    tx_ids: AccumulatedTxids,
+    tx_observer: &'a Mutex<TxObservationService>,
+    tx_observer_watcher: InProgressObserverTransactWatcher,
 }
 
 /// Represents an in-progress set of reads to the store. Just like `InProgress`,
@@ -366,12 +380,15 @@ impl<'a, 'c> InProgress<'a, 'c> {
     }
 
     pub fn transact_terms<I>(&mut self, terms: I, tempid_set: InternSet<TempId>) -> Result<TxReport> where I: IntoIterator<Item=TermWithTempIds> {
+        let w = InProgressTransactWatcher::new(
+                &mut self.tx_observer_watcher,
+                self.cache.transact_watcher());
         let (report, next_partition_map, next_schema, _watcher) =
             transact_terms(&self.transaction,
                            self.partition_map.clone(),
                            &self.schema,
                            &self.schema,
-                           self.cache.transact_watcher(),
+                           w,
                            terms,
                            tempid_set)?;
         self.partition_map = next_partition_map;
@@ -390,12 +407,16 @@ impl<'a, 'c> InProgress<'a, 'c> {
         //    `Metadata` on return. If we used `Cell` or other mechanisms, we'd be using
         //    `Default::default` in those situations to extract the partition map, and so there
         //    would still be some cost.
+        let w = InProgressTransactWatcher::new(
+                &mut self.tx_observer_watcher,
+                self.cache.transact_watcher());
         let (report, next_partition_map, next_schema, _watcher) =
             transact(&self.transaction,
                      self.partition_map.clone(),
                      &self.schema,
-                     &self.schema,
+                     &
-                     self.cache.transact_watcher(),
+                     self.schema,
+                     w,
                      entities)?;
         self.partition_map = next_partition_map;
         if let Some(schema) = next_schema {
@@ -451,6 +472,9 @@ impl<'a, 'c> InProgress<'a, 'c> {
             // TODO: consider making vocabulary lookup lazy -- we won't need it much of the time.
         }
 
+        let txes = self.tx_observer_watcher.txes;
+        self.tx_observer.lock().unwrap().in_progress_did_commit(txes);
+
         Ok(())
     }
 
@@ -478,6 +502,36 @@ impl<'a, 'c> InProgress<'a, 'c> {
     }
 }
 
+struct InProgressTransactWatcher<'a, 'o> {
+    cache_watcher: InProgressCacheTransactWatcher<'a>,
+    observer_watcher: &'o mut InProgressObserverTransactWatcher,
+    tx_id: Option<Entid>,
+}
+
+impl<'a, 'o> InProgressTransactWatcher<'a, 'o> {
+    fn new(observer_watcher: &'o mut InProgressObserverTransactWatcher, cache_watcher: InProgressCacheTransactWatcher<'a>) -> Self {
+        InProgressTransactWatcher {
+            cache_watcher: cache_watcher,
+            observer_watcher: observer_watcher,
+            tx_id: None,
+        }
+    }
+}
+
+impl<'a, 'o> TransactWatcher for InProgressTransactWatcher<'a, 'o> {
+    fn datom(&mut self, op: OpType, e: Entid, a: Entid, v: &TypedValue) {
+        self.cache_watcher.datom(op.clone(), e.clone(), a.clone(), v);
+        self.observer_watcher.datom(op.clone(), e.clone(), a.clone(), v);
+    }
+
+    fn done(&mut self, t: &Entid, schema: &Schema) -> ::mentat_db::errors::Result<()> {
+        self.cache_watcher.done(t, schema)?;
+        self.observer_watcher.done(t, schema)?;
+        self.tx_id = Some(t.clone());
+        Ok(())
+    }
+}
+
 impl Store {
     /// Intended for use from tests.
     pub fn sqlite_mut(&mut self) -> &mut rusqlite::Connection {
@@ -564,9 +618,15 @@ impl Conn {
     fn new(partition_map: PartitionMap, schema: Schema) -> Conn {
         Conn {
             metadata: Mutex::new(Metadata::new(0, partition_map, Arc::new(schema), Default::default())),
+            tx_observer_service: Mutex::new(TxObservationService::new()),
         }
     }
 
+    #[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.
@@ -705,6 +765,9 @@ impl Conn {
             schema: (*current_schema).clone(),
             cache: InProgressSQLiteAttributeCache::from_cache(cache_cow),
             use_caching: true,
+            tx_ids: Default::default(),
+            tx_observer: &self.tx_observer_service,
+            tx_observer_watcher: InProgressObserverTransactWatcher::new(),
         })
     }
 
@@ -786,6 +849,14 @@ impl Conn {
             },
         }
     }
+
+    pub fn register_observer(&mut self, key: String, observer: Arc<TxObserver>) {
+        self.tx_observer_service.lock().unwrap().register(key, observer);
+    }
+
+    pub fn unregister_observer(&mut self, key: &String) {
+        self.tx_observer_service.lock().unwrap().deregister(key);
+    }
 }
 
 #[cfg(test)]
@@ -798,19 +869,25 @@ mod tests {
     use std::collections::{
         BTreeSet,
     };
-
     use std::path::{
         PathBuf,
     };
-
+    use std::sync::mpsc;
-    use std::time::Instant;
+    use std::time::{
+        Duration,
+        Instant,
+    };
 
     use mentat_core::{
         CachedAttributes,
         TypedValue,
     };
 
-    use query::{
+    use ::entity_builder::{
+        BuildTerms,
+    };
+
+    use ::query::{
         PreparedQuery,
         Variable,
     };
@@ -821,6 +898,16 @@ mod tests {
         QueryResults,
     };
 
+    use ::vocabulary::{
+        AttributeBuilder,
+        Definition,
+        VersionedStore,
+    };
+
+    use ::vocabulary::attribute::{
+        Unique,
+    };
+
     use mentat_db::USER0;
 
     #[test]
@@ -1381,4 +1468,218 @@ mod tests {
             assert_eq!(only_two, vec![1, 3].into_iter().map(TypedValue::Long).collect());
         }
     }
+
+    fn test_register_observer() {
+        let mut sqlite = db::new_connection("").unwrap();
+        let mut conn = Conn::connect(&mut sqlite).unwrap();
+
+        let key = "Test Observer".to_string();
+        let tx_observer = TxObserver::new(BTreeSet::new(), move |_obs_key, _batch| {});
+
+        conn.register_observer(key.clone(), Arc::new(tx_observer));
+        assert!(conn.is_registered_as_observer(&key));
+    }
+
+    #[test]
+    fn test_deregister_observer() {
+        let mut sqlite = db::new_connection("").unwrap();
+        let mut conn = Conn::connect(&mut sqlite).unwrap();
+
+        let key = "Test Observer".to_string();
+
+        let tx_observer = TxObserver::new(BTreeSet::new(), move |_obs_key, _batch| {});
+
+        conn.register_observer(key.clone(), Arc::new(tx_observer));
+        assert!(conn.is_registered_as_observer(&key));
+
+        conn.unregister_observer(&key);
+
+        assert!(!conn.is_registered_as_observer(&key));
+    }
+
+    fn add_schema(conn: &mut Conn, mut sqlite: &mut rusqlite::Connection) {
+        // transact some schema
+        let mut in_progress = conn.begin_transaction(&mut sqlite).expect("expected in progress");
+        in_progress.ensure_vocabulary(&Definition {
+            name: kw!(:todo/items),
+            version: 1,
+            attributes: vec![
+                (kw!(:todo/uuid),
+                AttributeBuilder::new()
+                    .value_type(ValueType::Uuid)
+                    .multival(false)
+                    .unique(Unique::Value)
+                    .index(true)
+                    .build()),
+                (kw!(:todo/name),
+                AttributeBuilder::new()
+                    .value_type(ValueType::String)
+                    .multival(false)
+                    .fulltext(true)
+                    .build()),
+                (kw!(:todo/completion_date),
+                AttributeBuilder::new()
+                    .value_type(ValueType::Instant)
+                    .multival(false)
+                    .build()),
+                (kw!(:label/name),
+                AttributeBuilder::new()
+                    .value_type(ValueType::String)
+                    .multival(false)
+                    .unique(Unique::Value)
+                    .fulltext(true)
+                    .index(true)
+                    .build()),
+                (kw!(:label/color),
+                AttributeBuilder::new()
+                    .value_type(ValueType::String)
+                    .multival(false)
+                    .build()),
+            ],
+        }).expect("expected vocubulary");
+        in_progress.commit().expect("Expected vocabulary committed");
+    }
+
+    #[derive(Default, Debug)]
+    struct ObserverOutput {
+        txids: Vec<i64>,
+        changes: Vec<BTreeSet<i64>>,
+        called_key: Option<String>,
+    }
+
+    #[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 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 mut registered_attrs = BTreeSet::new();
+        registered_attrs.insert(name_entid.clone());
+        registered_attrs.insert(date_entid.clone());
+
+        let key = "Test Observing".to_string();
+
+        let output = Arc::new(Mutex::new(ObserverOutput::default()));
+
+        let mut_output = Arc::downgrade(&output);
+        let (tx, rx): (mpsc::Sender<()>, mpsc::Receiver<()>) = mpsc::channel();
+        // because the TxObserver is in an Arc and is therefore Sync, we have to wrap the Sender in a Mutex to also
+        // make it Sync.
+        let thread_tx = Mutex::new(tx);
+        let tx_observer = Arc::new(TxObserver::new(registered_attrs, move |obs_key, batch| {
+            if let Some(out) = mut_output.upgrade() {
+                let mut o = out.lock().unwrap();
+                o.called_key = Some(obs_key.to_string());
+                for (tx_id, changes) in batch.into_iter() {
+                    o.txids.push(*tx_id);
+                    o.changes.push(changes.clone());
+                }
+                o.txids.sort();
+            }
+            thread_tx.lock().unwrap().send(()).unwrap();
+        }));
+
+        conn.register_observer(key.clone(), Arc::clone(&tx_observer));
+        assert!(conn.is_registered_as_observer(&key));
+
+        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 mut in_progress = conn.begin_transaction(&mut sqlite).expect("expected transaction");
+            for i in 0..3 {
+                let mut changeset = BTreeSet::new();
+                let name = format!("todo{}", i);
+                let uuid = Uuid::new_v4();
+                let mut builder = in_progress.builder().describe_tempid(&name);
+                builder.add_kw(&kw!(:todo/uuid), TypedValue::Uuid(uuid)).expect("Expected added uuid");
+                changeset.insert(uuid_entid.clone());
+                builder.add_kw(&kw!(:todo/name), TypedValue::typed_string(&name)).expect("Expected added name");
+                changeset.insert(name_entid.clone());
+                if i % 2 == 0 {
+                    builder.add_kw(&kw!(:todo/completion_date), TypedValue::current_instant()).expect("Expected added date");
+                    changeset.insert(date_entid.clone());
+                }
+                let (ip, r) = builder.transact();
+                let report = r.expect("expected a report");
+                tx_ids.push(report.tx_id.clone());
+                changesets.push(changeset);
+                in_progress = ip;
+            }
+            let mut builder = in_progress.builder().describe_tempid("Label");
+            builder.add_kw(&kw!(:label/name), TypedValue::typed_string("Label 1")).expect("Expected added name");
+            builder.add_kw(&kw!(:label/color), TypedValue::typed_string("blue")).expect("Expected added color");
+            builder.commit().expect("expect transaction to occur");
+        }
+
+        let delay = Duration::from_millis(100);
+        let _ = rx.recv_timeout(delay);
+
+        let out = Arc::try_unwrap(output).expect("unwrapped");
+        let o = out.into_inner().expect("Expected an Output");
+        assert_eq!(o.called_key, Some(key.clone()));
+        assert_eq!(o.txids, tx_ids);
+        assert_eq!(o.changes, changesets);
+    }
+
+    #[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 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 mut registered_attrs = BTreeSet::new();
+        registered_attrs.insert(name_entid.clone());
+        registered_attrs.insert(date_entid.clone());
+
+        let key = "Test Observing".to_string();
+
+        let output = Arc::new(Mutex::new(ObserverOutput::default()));
+
+        let mut_output = Arc::downgrade(&output);
+        let (tx, rx): (mpsc::Sender<()>, mpsc::Receiver<()>) = mpsc::channel();
+        let thread_tx = Mutex::new(tx);
+        let tx_observer = Arc::new(TxObserver::new(registered_attrs, move |obs_key, batch| {
+            if let Some(out) = mut_output.upgrade() {
+                let mut o = out.lock().unwrap();
+                o.called_key = Some(obs_key.to_string());
+                for (tx_id, changes) in batch.into_iter() {
+                    o.txids.push(*tx_id);
+                    o.changes.push(changes.clone());
+                }
+                o.txids.sort();
+            }
+            thread_tx.lock().unwrap().send(()).unwrap();
+        }));
+
+        conn.register_observer(key.clone(), Arc::clone(&tx_observer));
+        assert!(conn.is_registered_as_observer(&key));
+
+        let tx_ids = Vec::<Entid>::new();
+        let changesets = Vec::<BTreeSet<Entid>>::new();
+        {
+            let mut in_progress = conn.begin_transaction(&mut sqlite).expect("expected transaction");
+            for i in 0..3 {
+                let name = format!("label{}", i);
+                let mut builder = in_progress.builder().describe_tempid(&name);
+                builder.add_kw(&kw!(:label/name), TypedValue::typed_string(&name)).expect("Expected added name");
+                builder.add_kw(&kw!(:label/color), TypedValue::typed_string("blue")).expect("Expected added color");
+                let (ip, _) = builder.transact();
+                in_progress = ip;
+            }
+        }
+
+        let delay = Duration::from_millis(100);
+        let _ = rx.recv_timeout(delay);
+
+        let out = Arc::try_unwrap(output).expect("unwrapped");
+        let o = out.into_inner().expect("Expected an Output");
+        assert_eq!(o.called_key, None);
+        assert_eq!(o.txids, tx_ids);
+        assert_eq!(o.changes, changesets);
+    }
 }

src/lib.rs

@@ -18,6 +18,8 @@ extern crate lazy_static;
 
 extern crate rusqlite;
 
+extern crate smallvec;
+
 extern crate uuid;
 
 pub extern crate edn;