Cleaner port logic.

rkistner · rkistner · commit 6f6f89d18cda · 2023-03-22T12:00:00.000+02:00
diff --git a/lib/src/connection_pool.dart b/lib/src/connection_pool.dart
@@ -1,7 +1,7 @@
 import 'dart:async';
-import 'dart:isolate';
 
 import 'mutex.dart';
+import 'port_channel.dart';
 import 'sqlite_connection.dart';
 import 'sqlite_connection_impl.dart';
 import 'sqlite_open_factory.dart';
@@ -15,7 +15,7 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
   final List<SqliteConnectionImpl> _readConnections = [];
 
   final SqliteOpenFactory _factory;
-  final SendPort _upstreamPort;
+  final SerializedPortClient _upstreamPort;
 
   @override
   final Stream<UpdateNotification>? updates;
@@ -42,7 +42,7 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       SqliteConnection? writeConnection,
       this.debugName,
       required this.mutex,
-      required SendPort upstreamPort})
+      required SerializedPortClient upstreamPort})
       : _writeConnection = writeConnection,
         _upstreamPort = upstreamPort;
 
diff --git a/lib/src/database_utils.dart b/lib/src/database_utils.dart
@@ -106,3 +106,7 @@ Future<Set<String>> getSourceTables(SqliteReadContext ctx, String sql) async {
   Set<String> tables = {for (var row in tableRows) row['tbl_name']};
   return tables;
 }
+
+class InitDb {
+  const InitDb();
+}
diff --git a/lib/src/port_channel.dart b/lib/src/port_channel.dart
@@ -2,15 +2,28 @@ import 'dart:async';
 import 'dart:collection';
 import 'dart:isolate';
 
-class PortClient {
+abstract class PortClient {
+  Future<T> post<T>(Object message);
+  void fire(Object message);
+
+  factory PortClient.parent() {
+    return ParentPortClient();
+  }
+
+  factory PortClient.child(SendPort upstream) {
+    return ChildPortClient(upstream);
+  }
+}
+
+class ParentPortClient implements PortClient {
   late Future<SendPort> sendPort;
   ReceivePort receivePort = ReceivePort();
   bool closed = false;
   int _nextId = 1;
 
   Map<int, Completer<Object?>> handlers = HashMap();
 
-  PortClient() {
+  ParentPortClient() {
     final initCompleter = Completer<SendPort>();
     sendPort = initCompleter.future;
     receivePort.listen((message) {
@@ -54,19 +67,28 @@ class PortClient {
     }
   }
 
+  @override
   Future<T> post<T>(Object message) async {
     if (closed) {
       throw ClosedException();
     }
     var completer = Completer<T>();
     var id = _nextId++;
     handlers[id] = completer;
-    (await sendPort).send(_RequestMessage(id, message));
+    (await sendPort).send(_RequestMessage(id, message, receivePort.sendPort));
     return await completer.future;
   }
 
-  PortServer server() {
-    return PortServer(receivePort.sendPort);
+  @override
+  void fire(Object message) async {
+    if (closed) {
+      throw ClosedException();
+    }
+    (await sendPort).send(_FireMessage(message));
+  }
+
+  RequestPortServer server() {
+    return RequestPortServer(receivePort.sendPort);
   }
 
   close() async {
@@ -83,24 +105,104 @@ class PortClient {
   }
 }
 
+class SerializedPortClient {
+  final SendPort sendPort;
+
+  SerializedPortClient(this.sendPort);
+
+  ChildPortClient open() {
+    return ChildPortClient(sendPort);
+  }
+}
+
+class ChildPortClient implements PortClient {
+  final SendPort sendPort;
+  final ReceivePort receivePort = ReceivePort();
+  int _nextId = 1;
+
+  final Map<int, Completer<Object?>> handlers = HashMap();
+
+  ChildPortClient(this.sendPort) {
+    receivePort.listen((message) {
+      if (message is _PortChannelResult) {
+        final handler = handlers.remove(message.requestId);
+        assert(handler != null);
+        if (message.success) {
+          handler!.complete(message.result);
+        } else {
+          handler!.completeError(message.error, message.stackTrace);
+        }
+      }
+    });
+  }
+
+  @override
+  Future<T> post<T>(Object message) async {
+    var completer = Completer<T>();
+    var id = _nextId++;
+    handlers[id] = completer;
+    sendPort.send(_RequestMessage(id, message, receivePort.sendPort));
+    return await completer.future;
+  }
+
+  @override
+  void fire(Object message) {
+    sendPort.send(_FireMessage(message));
+  }
+}
+
+class RequestPortServer {
+  final SendPort port;
+
+  RequestPortServer(this.port);
+
+  open(Future<Object?> Function(Object? message) handle) {
+    return PortServer.forSendPort(port, handle);
+  }
+}
+
 class PortServer {
-  SendPort port;
-  late ReceivePort receivePort;
-  late Future<Object?> Function(Object? message) handle;
-
-  PortServer(this.port);
-
-  void init(Future<Object?> Function(Object? message) handle) {
-    this.handle = handle;
-    receivePort = ReceivePort();
-    port.send(_InitMessage(receivePort.sendPort));
-    receivePort.listen((message) async {
-      final request = message as _RequestMessage;
-      try {
-        var result = await handle(request.message);
-        port.send(_PortChannelResult.success(request.id, result));
-      } catch (e, stacktrace) {
-        port.send(_PortChannelResult.error(request.id, e, stacktrace));
+  final ReceivePort _receivePort = ReceivePort();
+  final Future<Object?> Function(Object? message) handle;
+
+  PortServer(this.handle) {
+    _init();
+  }
+
+  PortServer.forSendPort(SendPort port, this.handle) {
+    port.send(_InitMessage(_receivePort.sendPort));
+    _init();
+  }
+
+  SendPort get sendPort {
+    return _receivePort.sendPort;
+  }
+
+  client() {
+    return SerializedPortClient(sendPort);
+  }
+
+  close() {
+    _receivePort.close();
+  }
+
+  _init() {
+    _receivePort.listen((request) async {
+      if (request is _FireMessage) {
+        handle(request.message);
+      } else if (request is _RequestMessage) {
+        if (request.id == 0) {
+          // Fire and forget
+          handle(request.message);
+        } else {
+          try {
+            var result = await handle(request.message);
+            request.reply.send(_PortChannelResult.success(request.id, result));
+          } catch (e, stacktrace) {
+            request.reply
+                .send(_PortChannelResult.error(request.id, e, stacktrace));
+          }
+        }
       }
     });
   }
@@ -114,11 +216,18 @@ class _InitMessage {
   _InitMessage(this.port);
 }
 
+class _FireMessage {
+  final Object message;
+
+  const _FireMessage(this.message);
+}
+
 class _RequestMessage {
   final int id;
   final Object message;
+  final SendPort reply;
 
-  _RequestMessage(this.id, this.message);
+  _RequestMessage(this.id, this.message, this.reply);
 }
 
 class ClosedException implements Exception {
diff --git a/lib/src/sqlite_connection_impl.dart b/lib/src/sqlite_connection_impl.dart
@@ -3,7 +3,7 @@ import 'dart:isolate';
 
 import 'package:sqlite3/sqlite3.dart' as sqlite;
 
-import 'isolate_completer.dart';
+import 'database_utils.dart';
 import 'mutex.dart';
 import 'port_channel.dart';
 import 'sqlite_connection.dart';
@@ -22,14 +22,14 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
 
   @override
   final Stream<UpdateNotification>? updates;
-  final PortClient _dbIsolate = PortClient();
+  final ParentPortClient _dbIsolate = ParentPortClient();
   final String? debugName;
   final bool readOnly;
 
   SqliteConnectionImpl(
       {required SqliteOpenFactory openFactory,
       required Mutex mutex,
-      required SendPort upstreamPort,
+      required SerializedPortClient upstreamPort,
       this.updates,
       this.debugName,
       this.readOnly = false,
@@ -43,7 +43,8 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
   }
 
   Future<void> _open(SqliteOpenFactory openFactory,
-      {required bool primary, required SendPort upstreamPort}) async {
+      {required bool primary,
+      required SerializedPortClient upstreamPort}) async {
     await _connectionMutex.lock(() async {
       var isolate = await Isolate.spawn(
           _sqliteConnectionIsolate,
@@ -66,11 +67,6 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
     return _connectionMutex.locked;
   }
 
-  /// For internal use only
-  Future<T> lock<T>(Future<T> Function() callback, {Duration? timeout}) async {
-    return _connectionMutex.lock(callback, timeout: timeout);
-  }
-
   @override
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
       {Duration? lockTimeout}) async {
@@ -197,14 +193,13 @@ class _TransactionContext implements SqliteWriteContext {
 }
 
 void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
-  final port = params.port;
+  final client = params.port.open();
+
   if (!params.primary) {
     // Wait until the primary connection has been initialized.
     // The primary connection is responsible for configuring journal mode,
     // running migrations, and other setup.
-    var initialized = IsolateResult<void>();
-    port.send(['init-db', initialized.completer]);
-    await initialized.future;
+    await client.post(const InitDb());
   }
   final db = await params.openFactory.open(SqliteOpenOptions(
       primaryConnection: params.primary, readOnly: params.readOnly));
@@ -219,7 +214,7 @@ void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
   db.updates.listen((event) {
     updatedTables.add(event.tableName);
   });
-  server.init((data) async {
+  server.open((data) async {
     if (data is _SqliteIsolateClose) {
       if (txId != null) {
         try {
@@ -255,7 +250,7 @@ void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
       try {
         final result = db.select(data.sql, data.args);
         if (updatedTables.isNotEmpty) {
-          port.send(['update', updatedTables]);
+          client.fire(UpdateNotification(updatedTables));
           updatedTables = {};
         }
         return result;
@@ -270,7 +265,7 @@ void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
         return await data.cb(db);
       } finally {
         if (updatedTables.isNotEmpty) {
-          port.send(['update', updatedTables]);
+          client.fire(UpdateNotification(updatedTables));
           updatedTables = {};
         }
       }
@@ -281,10 +276,10 @@ void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
 }
 
 class _SqliteConnectionParams {
-  final PortServer portServer;
+  final RequestPortServer portServer;
   final bool readOnly;
 
-  final SendPort port;
+  final SerializedPortClient port;
   final bool primary;
   final SqliteOpenFactory openFactory;
 
diff --git a/lib/src/sqlite_database.dart b/lib/src/sqlite_database.dart

Original file line number	Diff line number	Diff line change
`@@ -106,3 +106,7 @@ Future<Set<String>> getSourceTables(SqliteReadContext ctx, String sql) async {`
`106`	`106`	`Set<String> tables = {for (var row in tableRows) row['tbl_name']};`
`107`	`107`	`return tables;`
`108`	`108`	`}`
	`109`	`+`
	`110`	`+class InitDb {`
	`111`	`+ const InitDb();`
	`112`	`+}`