Add notifications to RealmObject

lib/src/native/realm_core.dart

@@ -27,6 +27,7 @@ import '../collection_changes.dart';
 import '../configuration.dart';
 import '../init.dart';
 import '../list.dart';
+import '../object_changes.dart';
 import '../realm_class.dart';
 import '../realm_object.dart';
 import '../results.dart';
@@ -391,6 +392,43 @@ class _RealmCore {
     });
   }
 
+  int getObjectChangesCount(RealmObjectChangesHandle changes) {
+    return _realmLib.realm_object_changes_get_num_modified_properties(changes._pointer);
+  }
+
+  List<int> getObjectChanges(RealmObjectChangesHandle changes, int max) {
+    return using((arena) {
+      final out_modified = arena<Int64>(max);
+      _realmLib.realm_object_changes_get_modified_properties(changes._pointer, out_modified, max);
+      return out_modified != nullptr ? out_modified.asTypedList(max).toList() : const [];
+    });
+  }
+
+  Stream<RealmObjectChanges> realmObjectChanged(RealmObject object, SchedulerHandle scheduler) {
+    late StreamController<RealmObjectChanges> controller;
+
+    void callback(Pointer<Void> data) {
+      final changes = RealmObjectChanges(
+        RealmObjectChangesHandle._(_realmLib.realm_clone(data).cast()),
+        object.realm!,
+      );
+      controller.add(changes);
+    }
+
+    controller = _constructRealmNotificationStreamController(
+        (userData, callback, free, error) => _realmLib.realm_object_add_notification_callback(
+              object.handle._pointer,
+              userData,
+              free,
+              callback.cast(),
+              error,
+              scheduler._pointer,
+            ),
+        callback);
+
+    return controller.stream;
+  }
+
   Counts getCollectionChangesCounts(RealmCollectionChangesHandle changes) {
     return using((arena) {
       final out_num_deletions = arena<IntPtr>();
@@ -651,6 +689,10 @@ class RealmCollectionChangesHandle extends Handle<realm_collection_changes> {
   RealmCollectionChangesHandle._(Pointer<realm_collection_changes> pointer) : super(pointer, 88); // TODO: What should gc hint be?
 }
 
+class RealmObjectChangesHandle extends Handle<realm_object_changes> {
+  RealmObjectChangesHandle._(Pointer<realm_object_changes> pointer) : super(pointer, 88); // TODO: What should gc hint be?
+}
+
 extension _StringEx on String {
   Pointer<T> toUtf8Ptr<T extends NativeType>(Allocator allocator) {
     final units = utf8.encode(this);

lib/src/object_changes.dart

@@ -0,0 +1,31 @@
+import 'realm_class.dart';
+import 'native/realm_core.dart';
+
+class RealmObjectChanges {
+  final RealmObjectChangesHandle _handle;
+  final Realm realm;
+
+  RealmObjectChanges(this._handle, this.realm);
+
+  int? _count;
+  int get count => _count ??= realmCore.getObjectChangesCount(_handle);
+
+  List<int>? _keys;
+  List<int> get keys => _keys ??= realmCore.getObjectChanges(_handle, count);
+}
+
+class ObjectChanges<T extends RealmObject> extends RealmObjectChanges {
+  T _object;
+  ObjectChanges._(this._object, RealmObjectChangesHandle handle, Realm realm) : super(handle, realm);
+
+  factory ObjectChanges(T object, RealmObjectChanges changes) {
+    return ObjectChanges._(object, changes._handle, changes.realm);
+  }
+
+  List<String>? _properties;
+  List<String> get properties => _properties ??= keys.map((k) => realm.metadata[_object.runtimeType]!.findByKey(k)).toList();
+}
+
+// TODO: Perhaps let generator generate typed RealmObjectChanges subtypes, 
+// that can have a much nicer type safe interface
+

lib/src/realm_object.dart

@@ -18,6 +18,7 @@
 
 import 'list.dart';
 import 'native/realm_core.dart';
+import 'object_changes.dart';
 import 'realm_class.dart';
 
 abstract class RealmAccessor {
@@ -222,6 +223,9 @@ class RealmObject {
     if (!isManaged || !other.isManaged) return false;
     return realmCore.equals(this, other);
   }
+
+  // TODO: Should we generate this as a Stream<ObjectChanges<T>> on each realm object class?
+  Stream<ObjectChanges> get changed => realmCore.realmObjectChanged(this, realm!.scheduler.handle).map((o) => ObjectChanges(this, o)); 
 }
 
 //RealmObject package internal members

test/realm_test.dart

@@ -826,6 +826,45 @@ Future<void> main([List<String>? args]) async {
         }
       });
 
+      test('RealmObject.changed', () async {
+        var config = Configuration([Dog.schema, Person.schema]);
+        var realm = Realm(config);
+
+        void write(void Function() writer) async {
+          realm.write(writer);
+          realm.write(() {}); // dummy write to raise notification from previous write
+        }
+
+        final dog = Dog('Fido');
+        write(() => realm.add(dog));
+
+        final stream = dog.changed.asBroadcastStream();
+
+        var callbacks = 0;
+        final subscription = stream.listen((_) => ++callbacks);
+
+        {
+          final event = stream.skip(1).first;
+          write(() => dog.age = 1);
+          final change = await event;
+          expect(callbacks, 2); // first time
+          expect(change.count, 1);
+          expect(change.properties, ['age']);
+        }
+        {
+          final event = stream.first;
+          write(() {
+            dog.owner = Person('Kasper');
+            dog.age = 2;
+          });
+          final change = await event;
+          expect(callbacks, 3); // once per transaction, not once per change
+          expect(change.count, 2);
+          expect(change.properties, ['owner', 'age']);
+        }
+        subscription.cancel();
+      });
+
       test('RealmList.changed', () async {
         var config = Configuration([Team.schema, Person.schema]);
         var realm = Realm(config);