Add notifications to RealmList

lib/src/collection_changes.dart

@@ -71,4 +71,9 @@ class RealmCollectionChanges {
 class RealmResultsChanges<T> extends RealmCollectionChanges {
   final RealmResults<T> results;
   RealmResultsChanges(this.results, RealmCollectionChanges changes) : super(changes._handle, changes.realm);
-}
+}
+
+class RealmListChanges<T> extends RealmCollectionChanges {
+  final RealmList<T> list;
+  RealmListChanges(this.list, RealmCollectionChanges changes) : super(changes._handle, changes.realm);
+}

lib/src/list.dart

@@ -66,6 +66,13 @@ class RealmList<T> extends collection.ListBase<T> {
   void clear() {
     realmCore.listClear(this);
   }
+
+  Stream<RealmListChanges<T>> get changed => realmCore
+      .listChanged(
+        this,
+        realm.scheduler.handle,
+      )
+      .map((changes) => RealmListChanges(this, changes));
 }
 
 /// @nodoc

lib/src/native/realm_core.dart

@@ -391,7 +391,6 @@ class _RealmCore {
     });
   }
 
-
   Counts getCollectionChangesCounts(RealmCollectionChangesHandle changes) {
     return using((arena) {
       final out_num_deletions = arena<IntPtr>();
@@ -480,6 +479,31 @@ class _RealmCore {
     return controller.stream;
   }
 
+  Stream<RealmCollectionChanges> listChanged(RealmList list, SchedulerHandle scheduler) {
+    late StreamController<RealmCollectionChanges> controller;
+
+    void callback(Pointer<Void> data) {
+      final changes = RealmCollectionChanges(
+        RealmCollectionChangesHandle._(_realmLib.realm_clone(data).cast()),
+        list.realm,
+      );
+      controller.add(changes);
+    }
+
+    controller = _constructRealmNotificationStreamController(
+        (userData, callback, free, error) => _realmLib.realm_list_add_notification_callback(
+              list.handle._pointer,
+              userData,
+              free,
+              callback.cast(),
+              error,
+              scheduler._pointer,
+            ),
+        callback);
+
+    return controller.stream;
+  }
+
   RealmLinkHandle _getObjectAsLink(RealmObject object) {
     final realm_link = _realmLib.realm_object_as_link(object.handle._pointer);
     return RealmLinkHandle._(realm_link);

test/realm_test.dart

@@ -836,6 +836,46 @@ Future<void> main([List<String>? args]) async {
         }
         await forceGC?.call();
       });
+
+      test('RealmList.changed', () async {
+        var config = Configuration([Team.schema, Person.schema]);
+        var realm = Realm(config);
+
+        final team = Team('Ferari');
+        realm.write(() => realm.add(team));
+
+        void write(void Function() writer) async {
+          realm.write(writer);
+          realm.write(() {}); // dummy write to raise notification from previous write
+        }
+
+        final stream = (team.players as RealmList<Person>).changed.asBroadcastStream();
+
+        var callbacks = 0;
+        final subscription = stream.listen((_) => ++callbacks);
+
+        {
+          final event = stream.skip(1).first;
+          write(() => team.players.add(Person('Niki')));
+          final change = await event;
+          expect(callbacks, 2); // first time
+          expect(change.counts, Counts(0, 1, 0, 0));
+          expect(change.changes.insertions, [0]);
+        }
+        {
+          final event = stream.first;
+          write(() {
+            team.players[0].name = 'Michael';
+            team.players.add(Person('Kimi'));
+          });
+          final change = await event;
+          expect(callbacks, 3); // once per transaction, not once per change
+          expect(change.counts, Counts(0, 1, 1, 0));
+          expect(change.changes.insertions, [1]);
+          expect(change.changes.modifications, [0]);
+        }
+        subscription.cancel();
+      });
     });
 
     test('RealmObject add with list properties', () {