Implement watch in isolate connections.

Author: rkistner

lib/src/database_utils.dart

@@ -1,5 +1,6 @@
 import 'dart:async';
 import 'dart:convert';
+import 'dart:isolate';
 
 import 'package:sqlite3/sqlite3.dart' as sqlite;
 
@@ -110,3 +111,15 @@ Future<Set<String>> getSourceTables(SqliteReadContext ctx, String sql) async {
 class InitDb {
   const InitDb();
 }
+
+class SubscribeToUpdates {
+  final SendPort port;
+
+  SubscribeToUpdates(this.port);
+}
+
+class UnsubscribeToUpdates {
+  final SendPort port;
+
+  UnsubscribeToUpdates(this.port);
+}

lib/src/isolate_connection_factory.dart

@@ -1,8 +1,13 @@
-import 'sqlite_connection_impl.dart';
-import 'sqlite_connection.dart';
+import 'dart:async';
+import 'dart:isolate';
+
+import 'database_utils.dart';
 import 'mutex.dart';
 import 'port_channel.dart';
+import 'sqlite_connection.dart';
+import 'sqlite_connection_impl.dart';
 import 'sqlite_open_factory.dart';
+import 'update_notification.dart';
 
 class IsolateConnectionFactory {
   SqliteOpenFactory openFactory;
@@ -15,12 +20,70 @@ class IsolateConnectionFactory {
       required this.upstreamPort});
 
   SqliteConnection open({String? debugName, bool readOnly = false}) {
-    return SqliteConnectionImpl(
+    final updates = _IsolateUpdateListener(upstreamPort);
+
+    var openMutex = mutex.open();
+
+    return _IsolateSqliteConnection(
         openFactory: openFactory,
-        mutex: mutex.open(),
+        mutex: openMutex,
         upstreamPort: upstreamPort,
         readOnly: readOnly,
         debugName: debugName,
-        primary: false);
+        updates: updates.stream,
+        closeFunction: () {
+          openMutex.close();
+          updates.close();
+        });
+  }
+}
+
+class _IsolateUpdateListener {
+  final ChildPortClient client;
+  final ReceivePort port = ReceivePort();
+  late final StreamController<UpdateNotification> controller;
+
+  _IsolateUpdateListener(SerializedPortClient upstreamPort)
+      : client = upstreamPort.open() {
+    controller = StreamController.broadcast(onListen: () {
+      client.fire(SubscribeToUpdates(port.sendPort));
+    }, onCancel: () {
+      client.fire(UnsubscribeToUpdates(port.sendPort));
+    });
+
+    port.listen((message) {
+      if (message is UpdateNotification) {
+        controller.add(message);
+      }
+    });
+  }
+
+  Stream<UpdateNotification> get stream {
+    return controller.stream;
+  }
+
+  close() {
+    client.fire(UnsubscribeToUpdates(port.sendPort));
+    controller.close();
+    port.close();
+  }
+}
+
+class _IsolateSqliteConnection extends SqliteConnectionImpl {
+  final void Function() closeFunction;
+
+  _IsolateSqliteConnection(
+      {required super.openFactory,
+      required super.mutex,
+      required super.upstreamPort,
+      super.updates,
+      super.debugName,
+      super.readOnly = false,
+      required this.closeFunction});
+
+  @override
+  Future<void> close() async {
+    await super.close();
+    closeFunction();
   }
 }

lib/src/sqlite_connection_impl.dart

@@ -20,6 +20,7 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
   final SimpleMutex _connectionMutex = SimpleMutex();
   final Mutex _writeMutex;
 
+  /// Must be a broadcast stream
   @override
   final Stream<UpdateNotification>? updates;
   final ParentPortClient _isolateClient = ParentPortClient();

lib/src/sqlite_database.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:isolate';
 
 import 'connection_pool.dart';
 import 'database_utils.dart';
@@ -101,6 +102,8 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
   void _listenForEvents() {
     UpdateNotification? updates;
 
+    Map<SendPort, StreamSubscription> subscriptions = {};
+
     _eventsPort = PortServer((message) async {
       if (message is UpdateNotification) {
         if (updates == null) {
@@ -119,6 +122,17 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
         }
       } else if (message is InitDb) {
         await _initialized;
+      } else if (message is SubscribeToUpdates) {
+        if (subscriptions.containsKey(message.port)) {
+          return;
+        }
+        final subscription = _updatesController.stream.listen((event) {
+          message.port.send(event);
+        });
+        subscriptions[message.port] = subscription;
+      } else if (message is UnsubscribeToUpdates) {
+        final subscription = subscriptions.remove(message.port);
+        subscription?.cancel();
       }
     });
   }

lib/src/sqlite_queries.dart

@@ -10,6 +10,7 @@ import 'update_notification.dart';
 /// Classes using this need to implement [SqliteConnection.readLock]
 /// and [SqliteConnection.writeLock].
 mixin SqliteQueries implements SqliteWriteContext, SqliteConnection {
+  /// Broadcast stream that is notified of any table updates
   Stream<UpdateNotification>? get updates;
 
   @override

test/basic_test.dart

@@ -47,7 +47,7 @@ void main() {
           ]));
     });
 
-    // Manual test
+    // Manually verified
     test('Concurrency', () async {
       final db = SqliteDatabase.withFactory(
           openFactory: testFactory(path: path), maxReaders: 3);

test/watch_test.dart

@@ -1,8 +1,12 @@
 import 'dart:async';
+import 'dart:isolate';
 import 'dart:math';
 
+import 'package:async/async.dart';
+import 'package:sqlite3/sqlite3.dart';
 import 'package:sqlite_async/sqlite_async.dart';
 import 'package:sqlite_async/src/database_utils.dart';
+import 'package:sqlite_async/src/isolate_connection_factory.dart';
 import 'package:test/test.dart';
 
 import 'util.dart';
@@ -163,5 +167,84 @@ void main() {
             UpdateNotification.single('assets')
           ]));
     });
+
+    test('watch in isolate', () async {
+      final db = await setupDatabase(path: path);
+      await createTables(db);
+
+      const baseTime = 20;
+
+      const throttleDuration = Duration(milliseconds: baseTime);
+
+      final rows = await db.execute(
+          'INSERT INTO customers(name) VALUES (?) RETURNING id',
+          ['a customer']);
+      final id = rows[0]['id'];
+
+      var done = false;
+      inserts() async {
+        while (!done) {
+          await db.execute(
+              'INSERT INTO assets(make, customer_id) VALUES (?, ?)',
+              ['test', id]);
+          await Future.delayed(
+              Duration(milliseconds: Random().nextInt(baseTime * 2)));
+        }
+      }
+
+      const numberOfQueries = 10;
+
+      inserts();
+
+      final factory = db.isolateConnectionFactory();
+
+      var l = await inIsolateWatch(factory, numberOfQueries, throttleDuration);
+
+      var results = l[0] as List<ResultSet>;
+      var times = l[1] as List<DateTime>;
+      done = true;
+
+      var lastCount = 0;
+      for (var r in results) {
+        final count = r.first['count'];
+        // This is not strictly incrementing, since we can't guarantee the
+        // exact order between reads and writes.
+        // We can guarantee that there will always be a read after the last write,
+        // but the previous read may have been after the same write in some cases.
+        expect(count, greaterThanOrEqualTo(lastCount));
+        lastCount = count;
+      }
+
+      // The number of read queries must not be greater than the number of writes overall.
+      expect(numberOfQueries, lessThanOrEqualTo(results.last.first['count']));
+
+      DateTime? lastTime;
+      for (var r in times) {
+        if (lastTime != null) {
+          var diff = r.difference(lastTime);
+          expect(diff, greaterThanOrEqualTo(throttleDuration));
+        }
+        lastTime = r;
+      }
+    });
+  });
+}
+
+Future<List<Object>> inIsolateWatch(IsolateConnectionFactory factory,
+    int numberOfQueries, Duration throttleDuration) async {
+  return await Isolate.run(() async {
+    final db = factory.open();
+
+    final stream = db.watch(
+        'SELECT count() AS count FROM assets INNER JOIN customers ON customers.id = assets.customer_id',
+        throttle: throttleDuration);
+    List<DateTime> times = [];
+    final results = await stream.take(numberOfQueries).map((e) {
+      times.add(DateTime.now());
+      return e;
+    }).toList();
+
+    db.close();
+    return [results, times];
   });
 }