Port some of the KVO support functionality to the object store

shared_realm.cpp

@@ -18,14 +18,296 @@
 
 #include "shared_realm.hpp"
 
+#include "realm_delegate.hpp"
+
 #include <realm/commit_log.hpp>
 #include <realm/group_shared.hpp>
 #include <realm/lang_bind_helper.hpp>
 
 #include <mutex>
+#include <numeric>
 
 using namespace realm;
 
+namespace {
+class TransactLogHandler {
+    using ColumnInfo = RealmDelegate::ColumnInfo;
+    using ObserverState = RealmDelegate::ObserverState;
+
+    size_t currentTable = 0;
+    std::vector<ObserverState> observers;
+    std::vector<void *> invalidated;
+    ColumnInfo *activeLinkList = nullptr;
+    RealmDelegate* m_delegate;
+
+    // Get the change info for the given column, creating it if needed
+    static ColumnInfo& get_change(ObserverState& state, size_t i)
+    {
+        if (state.changes.size() <= i) {
+            state.changes.resize(std::max(state.changes.size() * 2, i + 1));
+        }
+        return state.changes[i];
+    }
+
+    // Loop over the columns which were changed in an observer state
+    template<typename Func>
+    static void for_each(ObserverState& state, Func&& f)
+    {
+        for (size_t i = 0; i < state.changes.size(); ++i) {
+            auto const& change = state.changes[i];
+            if (change.changed) {
+                f(i, change);
+            }
+        }
+    }
+
+    // Mark the given row/col as needing notifications sent
+    bool mark_dirty(size_t row_ndx, size_t col_ndx)
+    {
+        auto it = lower_bound(begin(observers), end(observers), ObserverState{currentTable, row_ndx, nullptr});
+        if (it != end(observers) && it->table_ndx == currentTable && it->row_ndx == row_ndx) {
+            get_change(*it, col_ndx).changed = true;
+        }
+        return true;
+    }
+
+    // Remove the given observer from the list of observed objects and add it
+    // to the listed of invalidated objects
+    void invalidate(ObserverState *o)
+    {
+        invalidated.push_back(o->info);
+        observers.erase(observers.begin() + (o - &observers[0]));
+    }
+
+public:
+    template<typename Func>
+    TransactLogHandler(RealmDelegate* delegate, SharedGroup& sg, Func&& func)
+    : m_delegate(delegate)
+    {
+        if (!delegate) {
+            func();
+            return;
+        }
+
+        observers = delegate->get_observed_rows();
+        if (observers.empty()) {
+            auto old_version = sg.get_version_of_current_transaction();
+            func();
+            if (old_version != sg.get_version_of_current_transaction()) {
+                delegate->did_change({}, {});
+            }
+            return;
+        }
+
+        func(*this);
+        delegate->did_change(observers, invalidated);
+    }
+
+    // Called at the end of the transaction log immediately before the version
+    // is advanced
+    void parse_complete()
+    {
+        m_delegate->will_change(observers, invalidated);
+    }
+
+    // These would require having an observer before schema init
+    // Maybe do something here to throw an error when multiple processes have different schemas?
+    bool insert_group_level_table(size_t, size_t, StringData) noexcept { return false; }
+    bool erase_group_level_table(size_t, size_t) noexcept { return false; }
+    bool rename_group_level_table(size_t, StringData) noexcept { return false; }
+    bool insert_column(size_t, DataType, StringData, bool) { return false; }
+    bool insert_link_column(size_t, DataType, StringData, size_t, size_t) { return false; }
+    bool erase_column(size_t) { return false; }
+    bool erase_link_column(size_t, size_t, size_t) { return false; }
+    bool rename_column(size_t, StringData) { return false; }
+    bool add_search_index(size_t) { return false; }
+    bool remove_search_index(size_t) { return false; }
+    bool add_primary_key(size_t) { return false; }
+    bool remove_primary_key() { return false; }
+    bool set_link_type(size_t, LinkType) { return false; }
+
+    bool select_table(size_t group_level_ndx, int, const size_t*) noexcept {
+        currentTable = group_level_ndx;
+        return true;
+    }
+
+    bool insert_empty_rows(size_t, size_t, size_t, bool) {
+        // rows are only inserted at the end, so no need to do anything
+        return true;
+    }
+
+    bool erase_rows(size_t row_ndx, size_t, size_t last_row_ndx, bool unordered) noexcept {
+        for (size_t i = 0; i < observers.size(); ++i) {
+            auto& o = observers[i];
+            if (o.table_ndx == currentTable) {
+                if (o.row_ndx == row_ndx) {
+                    invalidate(&o);
+                    --i;
+                }
+                else if (unordered && o.row_ndx == last_row_ndx) {
+                    o.row_ndx = row_ndx;
+                }
+                else if (!unordered && o.row_ndx > row_ndx) {
+                    o.row_ndx -= 1;
+                }
+            }
+        }
+        return true;
+    }
+
+    bool clear_table() noexcept {
+        for (size_t i = 0; i < observers.size(); ) {
+            auto& o = observers[i];
+            if (o.table_ndx == currentTable) {
+                invalidate(&o);
+            }
+            else {
+                ++i;
+            }
+        }
+        return true;
+    }
+
+    bool select_link_list(size_t col, size_t row) {
+        activeLinkList = nullptr;
+        for (auto& o : observers) {
+            if (o.table_ndx == currentTable && o.row_ndx == row) {
+                activeLinkList = &get_change(o, col);
+                break;
+            }
+        }
+        return true;
+    }
+
+    void append_link_list_change(ColumnInfo::Kind kind, size_t index) {
+        ColumnInfo *o = activeLinkList;
+        if (!o || o->kind == ColumnInfo::Kind::SetAll) {
+            // Active LinkList isn't observed or already has multiple kinds of changes
+            return;
+        }
+
+        if (o->kind == ColumnInfo::Kind::None) {
+            o->kind = kind;
+            o->changed = true;
+            o->indices.add(index);
+        }
+        else if (o->kind == kind) {
+            if (kind == ColumnInfo::Kind::Remove) {
+                // Shift the index to compensate for already-removed indices
+                for (auto i : o->indices) {
+                    if (i <= index)
+                        ++index;
+                    else
+                        break;
+                }
+                o->indices.add(index);
+            }
+            else if (kind == ColumnInfo::Kind::Insert) {
+                o->indices.insert_at(index);
+            }
+            else {
+                o->indices.add(index);
+            }
+        }
+        else {
+            // Array KVO can only send a single kind of change at a time, so
+            // if there's multiple just give up and send "Set"
+            o->indices.set(0);
+            o->kind = ColumnInfo::Kind::SetAll;
+        }
+    }
+
+    bool link_list_set(size_t index, size_t) {
+        append_link_list_change(ColumnInfo::Kind::Set, index);
+        return true;
+    }
+
+    bool link_list_insert(size_t index, size_t) {
+        append_link_list_change(ColumnInfo::Kind::Insert, index);
+        return true;
+    }
+
+    bool link_list_erase(size_t index) {
+        append_link_list_change(ColumnInfo::Kind::Remove, index);
+        return true;
+    }
+
+    bool link_list_nullify(size_t index) {
+        append_link_list_change(ColumnInfo::Kind::Remove, index);
+        return true;
+    }
+
+    bool link_list_swap(size_t index1, size_t index2) {
+        append_link_list_change(ColumnInfo::Kind::Set, index1);
+        append_link_list_change(ColumnInfo::Kind::Set, index2);
+        return true;
+    }
+
+    bool link_list_clear(size_t old_size) {
+        ColumnInfo *o = activeLinkList;
+        if (!o || o->kind == ColumnInfo::Kind::SetAll) {
+            return true;
+        }
+
+        if (o->kind == ColumnInfo::Kind::Remove)
+            old_size += o->indices.size();
+        else if (o->kind == ColumnInfo::Kind::Insert)
+            old_size -= o->indices.size();
+
+        o->indices.set(old_size);
+
+        o->kind = ColumnInfo::Kind::Remove;
+        o->changed = true;
+        return true;
+    }
+
+    bool link_list_move(size_t from, size_t to) {
+        ColumnInfo *o = activeLinkList;
+        if (!o || o->kind == ColumnInfo::Kind::SetAll) {
+            return true;
+        }
+        if (from > to) {
+            std::swap(from, to);
+        }
+
+        if (o->kind == ColumnInfo::Kind::None) {
+            o->kind = ColumnInfo::Kind::Set;
+            o->changed = true;
+        }
+        if (o->kind == ColumnInfo::Kind::Set) {
+            for (size_t i = from; i <= to; ++i)
+                o->indices.add(i);
+        }
+        else {
+            o->indices.set(0);
+            o->kind = ColumnInfo::Kind::SetAll;
+        }
+        return true;
+    }
+
+    // Things that just mark the field as modified
+    bool set_int(size_t col, size_t row, int_fast64_t) { return mark_dirty(row, col); }
+    bool set_bool(size_t col, size_t row, bool) { return mark_dirty(row, col); }
+    bool set_float(size_t col, size_t row, float) { return mark_dirty(row, col); }
+    bool set_double(size_t col, size_t row, double) { return mark_dirty(row, col); }
+    bool set_string(size_t col, size_t row, StringData) { return mark_dirty(row, col); }
+    bool set_binary(size_t col, size_t row, BinaryData) { return mark_dirty(row, col); }
+    bool set_date_time(size_t col, size_t row, DateTime) { return mark_dirty(row, col); }
+    bool set_table(size_t col, size_t row) { return mark_dirty(row, col); }
+    bool set_mixed(size_t col, size_t row, const Mixed&) { return mark_dirty(row, col); }
+    bool set_link(size_t col, size_t row, size_t) { return mark_dirty(row, col); }
+    bool set_null(size_t col, size_t row) { return mark_dirty(row, col); }
+    bool nullify_link(size_t col, size_t row) { return mark_dirty(row, col); }
+
+    // Things we don't need to do anything for
+    bool optimize_table() { return false; }
+
+    // Things that we don't do in the binding
+    bool select_descriptor(int, const size_t*) { return true; }
+    bool add_int_to_column(size_t, int_fast64_t) { return false; }
+};
+}
+
 RealmCache Realm::s_global_cache;
 
 Realm::Config::Config(const Config& c)
@@ -214,18 +496,13 @@ void Realm::begin_transaction()
         throw InvalidTransactionException("The Realm is already in a write transaction");
     }
 
-    // if the upgrade to write will move the transaction forward, announce the change after promoting
-    bool announce = m_shared_group->has_changed();
-
     // make sure we have a read transaction
     read_group();
 
-    LangBindHelper::promote_to_write(*m_shared_group, *m_history);
+    TransactLogHandler(m_delegate.get(), *m_shared_group, [&](auto&&... args) {
+        LangBindHelper::promote_to_write(*m_shared_group, *m_history, std::move(args)...);
+    });
     m_in_transaction = true;
-
-    if (announce && m_delegate) {
-        m_delegate->did_change();
-    }
 }
 
 void Realm::commit_transaction()
@@ -242,7 +519,6 @@ void Realm::commit_transaction()
 
     if (m_delegate) {
         m_delegate->transaction_committed();
-        m_delegate->did_change();
     }
 }
 
@@ -255,7 +531,9 @@ void Realm::cancel_transaction()
         throw InvalidTransactionException("Can't cancel a non-existing write transaction");
     }
 
-    LangBindHelper::rollback_and_continue_as_read(*m_shared_group, *m_history);
+    TransactLogHandler(m_delegate.get(), *m_shared_group, [&](auto&&... args) {
+        LangBindHelper::rollback_and_continue_as_read(*m_shared_group, *m_history, std::move(args)...);
+    });
     m_in_transaction = false;
 }
 
@@ -303,10 +581,12 @@ void Realm::notify()
     if (m_shared_group->has_changed()) { // Throws
         if (m_auto_refresh) {
             if (m_group) {
-                LangBindHelper::advance_read(*m_shared_group, *m_history);
+                TransactLogHandler(m_delegate.get(), *m_shared_group, [&](auto&&... args) {
+                    LangBindHelper::advance_read(*m_shared_group, *m_history, std::move(args)...);
+                });
             }
-            if (m_delegate) {
+            else if (m_delegate) {
-                m_delegate->did_change();
+                m_delegate->did_change({}, {});
             }
         }
         else if (m_delegate) {
@@ -332,16 +612,15 @@ bool Realm::refresh()
     }
 
     if (m_group) {
-        LangBindHelper::advance_read(*m_shared_group, *m_history);
+        TransactLogHandler(m_delegate.get(), *m_shared_group, [&](auto&&... args) {
+            LangBindHelper::advance_read(*m_shared_group, *m_history, std::move(args)...);
+        });
     }
     else {
         // Create the read transaction
         read_group();
     }
 
-    if (m_delegate) {
-        m_delegate->did_change();
-    }
     return true;
 }
 

shared_realm.hpp

@@ -132,23 +132,6 @@ namespace realm {
         std::mutex m_mutex;
     };
 
-    class RealmDelegate
-    {
-    public:
-        virtual ~RealmDelegate() = default;
-
-        // The Realm has committed a write transaction, and other Realms at the
-        // same path should be notified
-        virtual void transaction_committed() = 0;
-
-        // There are now new versions available for the Realm, but it has not
-        // had its read version advanced
-        virtual void changes_available() = 0;
-
-        // The Realm's read version has advanced
-        virtual void did_change() = 0;
-    };
-
     class RealmFileException : public std::runtime_error
     {
       public: