Make table updates a multi-subscription stream

Author: simolus3

packages/sqlite_async/lib/src/native/database/native_sqlite_connection_impl.dart

@@ -303,7 +303,7 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
   final server = params.portServer;
   final commandPort = ReceivePort();
 
-  db.throttledUpdatedTables.listen((changedTables) {
+  db.updatedTables.listen((changedTables) {
     client.fire(UpdateNotification(changedTables));
   });
 

packages/sqlite_async/lib/src/utils/shared_utils.dart

@@ -79,68 +79,87 @@ List<Object?> mapParameters(List<Object?> parameters) {
 }
 
 extension ThrottledUpdates on CommonDatabase {
-  /// Wraps [updatesSync] to:
+  /// An unthrottled stream of updated tables that emits on every commit.
   ///
-  ///   - Not fire in transactions.
-  ///   - Fire asynchronously.
-  ///   - Only report table names, which are buffered to avoid duplicates.
-  Stream<Set<String>> get throttledUpdatedTables {
-    StreamController<Set<String>>? controller;
-    var pendingUpdates = <String>{};
-    var paused = false;
-
-    Timer? updateDebouncer;
-
-    void maybeFireUpdates() {
-      updateDebouncer?.cancel();
-      updateDebouncer = null;
-
-      if (paused) {
-        // Continue collecting updates, but don't fire any
-        return;
+  /// A paused subscription on this stream will buffer changed tables into a
+  /// growing set instead of losing events, so this stream is simple to throttle
+  /// downstream.
+  Stream<Set<String>> get updatedTables {
+    final listeners = <_UpdateListener>[];
+    var uncommitedUpdates = <String>{};
+    var underlyingSubscriptions = <StreamSubscription<void>>[];
+
+    void handleUpdate(SqliteUpdate update) {
+      uncommitedUpdates.add(update.tableName);
+    }
+
+    void afterCommit() {
+      for (final listener in listeners) {
+        listener.notify(uncommitedUpdates);
       }
 
-      if (!autocommit) {
-        // Inside a transaction - do not fire updates
-        return;
+      uncommitedUpdates.clear();
+    }
+
+    void afterRollback() {
+      uncommitedUpdates.clear();
+    }
+
+    void addListener(_UpdateListener listener) {
+      listeners.add(listener);
+
+      if (listeners.length == 1) {
+        // First listener, start listening for raw updates on underlying
+        // database.
+        underlyingSubscriptions = [
+          updatesSync.listen(handleUpdate),
+          commits.listen((_) => afterCommit()),
+          commits.listen((_) => afterRollback())
+        ];
       }
+    }
 
-      if (pendingUpdates.isNotEmpty) {
-        controller!.add(pendingUpdates);
-        pendingUpdates = {};
+    void removeListener(_UpdateListener listener) {
+      listeners.remove(listener);
+      if (listeners.isEmpty) {
+        for (final sub in underlyingSubscriptions) {
+          sub.cancel();
+        }
       }
     }
 
-    void collectUpdate(SqliteUpdate event) {
-      pendingUpdates.add(event.tableName);
+    return Stream.multi(
+      (listener) {
+        final wrapped = _UpdateListener(listener);
+        addListener(wrapped);
 
-      updateDebouncer ??=
-          Timer(const Duration(milliseconds: 1), maybeFireUpdates);
+        listener.onCancel = () => removeListener(wrapped);
+      },
+      isBroadcast: true,
+    );
+  }
+}
+
+class _UpdateListener {
+  final MultiStreamController<Set<String>> downstream;
+  Set<String> buffered = {};
+
+  _UpdateListener(this.downstream);
+
+  void notify(Set<String> pendingUpdates) {
+    buffered.addAll(pendingUpdates);
+    if (!downstream.isPaused) {
+      downstream.add(buffered);
+      buffered = {};
     }
+  }
+}
 
-    StreamSubscription? txSubscription;
-    StreamSubscription? sourceSubscription;
-
-    controller = StreamController(onListen: () {
-      txSubscription = commits.listen((_) {
-        maybeFireUpdates();
-      }, onError: (error) {
-        controller?.addError(error);
-      });
-
-      sourceSubscription = updatesSync.listen(collectUpdate, onError: (error) {
-        controller?.addError(error);
-      });
-    }, onPause: () {
-      paused = true;
-    }, onResume: () {
-      paused = false;
-      maybeFireUpdates();
-    }, onCancel: () {
-      txSubscription?.cancel();
-      sourceSubscription?.cancel();
-    });
-
-    return controller.stream;
+extension StreamUtils<T> on Stream<T> {
+  Stream<T> pauseAfterEvent(Duration duration) async* {
+    await for (final event in this) {
+      yield event;
+      await Future.delayed(duration);
+    }
   }
 }

packages/sqlite_async/lib/src/web/worker/worker_utils.dart

@@ -56,7 +56,8 @@ class AsyncSqliteDatabase extends WorkerDatabase {
   final Map<ClientConnection, _ConnectionState> _state = {};
 
   AsyncSqliteDatabase({required this.database})
-      : _updates = database.throttledUpdatedTables;
+      : _updates = database.updatedTables
+            .pauseAfterEvent(const Duration(milliseconds: 1));
 
   _ConnectionState _findState(ClientConnection connection) {
     return _state.putIfAbsent(connection, _ConnectionState.new);