Remove SqliteConnectionFactory - single-isolate support for now.

Author: rkistner

lib/src/connection_pool.dart

@@ -1,8 +1,10 @@
 import 'dart:async';
+import 'dart:isolate';
 
+import 'mutex.dart';
 import 'sqlite_connection.dart';
-import 'sqlite_connection_factory.dart';
 import 'sqlite_connection_impl.dart';
+import 'sqlite_open_factory.dart';
 import 'sqlite_queries.dart';
 import 'update_notification.dart';
 
@@ -12,7 +14,8 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
 
   final List<SqliteConnectionImpl> _readConnections = [];
 
-  final SqliteConnectionFactory _factory;
+  final SqliteOpenFactory _factory;
+  final SendPort _upstreamPort;
 
   @override
   final Stream<UpdateNotification>? updates;
@@ -21,6 +24,8 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
 
   final String? debugName;
 
+  final Mutex mutex;
+
   /// Open a new connection pool.
   ///
   /// The provided factory is used to open connections on demand. Connections
@@ -35,8 +40,11 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       {this.updates,
       this.maxReaders = 5,
       SqliteConnection? writeConnection,
-      this.debugName})
-      : _writeConnection = writeConnection;
+      this.debugName,
+      required this.mutex,
+      required SendPort upstreamPort})
+      : _writeConnection = writeConnection,
+        _upstreamPort = upstreamPort;
 
   @override
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
@@ -92,8 +100,14 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
   @override
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
       {Duration? lockTimeout}) {
-    _writeConnection ??= _factory.openConnection(
-        debugName: debugName != null ? '$debugName-writer' : null);
+    _writeConnection ??= SqliteConnectionImpl(
+        upstreamPort: _upstreamPort,
+        primary: false,
+        updates: updates,
+        debugName: debugName != null ? '$debugName-writer' : null,
+        mutex: mutex,
+        readOnly: false,
+        openFactory: _factory);
     return _writeConnection!.writeLock(callback, lockTimeout: lockTimeout);
   }
 
@@ -106,10 +120,14 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       var name = debugName == null
           ? null
           : '$debugName-${_readConnections.length + 1}';
-      var connection = _factory.openConnection(
+      var connection = SqliteConnectionImpl(
+          upstreamPort: _upstreamPort,
+          primary: false,
           updates: updates,
           debugName: name,
-          readOnly: true) as SqliteConnectionImpl;
+          mutex: mutex,
+          readOnly: true,
+          openFactory: _factory);
       _readConnections.add(connection);
 
       // Edge case:

lib/src/port_channel.dart

@@ -2,8 +2,6 @@ import 'dart:async';
 import 'dart:collection';
 import 'dart:isolate';
 
-import 'package:sqlite_async/src/isolate_completer.dart';
-
 class PortClient {
   late Future<SendPort> sendPort;
   ReceivePort receivePort = ReceivePort();

lib/src/sqlite_connection_factory.dart

@@ -1,13 +1,13 @@
 import 'dart:async';
-import 'dart:collection';
 import 'dart:isolate';
 
 import 'package:sqlite3/sqlite3.dart' as sqlite;
-import 'package:sqlite_async/src/port_channel.dart';
 
+import 'isolate_completer.dart';
 import 'mutex.dart';
+import 'port_channel.dart';
 import 'sqlite_connection.dart';
-import 'sqlite_connection_factory.dart';
+import 'sqlite_open_factory.dart';
 import 'sqlite_queries.dart';
 import 'update_notification.dart';
 
@@ -16,31 +16,42 @@ typedef TxCallback<T> = Future<T> Function(sqlite.Database db);
 /// Implements a SqliteConnection using a separate isolate for the database
 /// operations.
 class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
-  final SqliteConnectionFactory _factory;
-
   /// Private to this connection
   final SimpleMutex _connectionMutex = SimpleMutex();
+  final Mutex _writeMutex;
 
   @override
   final Stream<UpdateNotification>? updates;
   final PortClient _dbIsolate = PortClient();
   final String? debugName;
   final bool readOnly;
 
-  SqliteConnectionImpl(this._factory,
-      {this.updates, this.debugName, this.readOnly = false}) {
-    _open();
+  SqliteConnectionImpl(
+      {required SqliteOpenFactory openFactory,
+      required Mutex mutex,
+      required SendPort upstreamPort,
+      this.updates,
+      this.debugName,
+      this.readOnly = false,
+      bool primary = false})
+      : _writeMutex = mutex {
+    _open(openFactory, primary: primary, upstreamPort: upstreamPort);
   }
 
   Future<void> get ready async {
     await _dbIsolate.ready;
   }
 
-  Future<void> _open() async {
+  Future<void> _open(SqliteOpenFactory openFactory,
+      {required bool primary, required SendPort upstreamPort}) async {
     await _connectionMutex.lock(() async {
       var isolate = await Isolate.spawn(
           _sqliteConnectionIsolate,
-          _SqliteConnectionParams(_factory, _dbIsolate.server(),
+          _SqliteConnectionParams(
+              openFactory: openFactory,
+              port: upstreamPort,
+              primary: primary,
+              portServer: _dbIsolate.server(),
               readOnly: readOnly),
           debugName: debugName,
           paused: true);
@@ -88,7 +99,7 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
         stopWatch.stop();
       }
       // DB lock so that only one write happens at a time
-      return await _factory.mutex.lock(() async {
+      return await _writeMutex.lock(() async {
         final ctx = _TransactionContext(_dbIsolate);
         try {
           return await callback(ctx);
@@ -173,7 +184,7 @@ class _TransactionContext implements SqliteWriteContext {
   @override
   Future<void> executeBatch(String sql, List<List<Object?>> parameterSets) {
     return computeWithDatabase((db) async {
-      final statement = db.prepare(sql);
+      final statement = db.prepare(sql, checkNoTail: true);
       try {
         for (var parameters in parameterSets) {
           statement.execute(parameters);
@@ -186,15 +197,24 @@ class _TransactionContext implements SqliteWriteContext {
 }
 
 void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
-  final db = await params.factory.openRawDatabase(readOnly: params.readOnly);
-  final port = params.factory.port;
+  final port = params.port;
+  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;
+  }
+  final db = await params.openFactory.open(SqliteOpenOptions(
+      primaryConnection: params.primary, readOnly: params.readOnly));
 
   final server = params.portServer;
   final commandPort = ReceivePort();
 
   Set<String> updatedTables = {};
-  int? txId = null;
-  Object? txError = null;
+  int? txId;
+  Object? txError;
 
   db.updates.listen((event) {
     updatedTables.add(event.tableName);
@@ -261,12 +281,19 @@ void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
 }
 
 class _SqliteConnectionParams {
-  SqliteConnectionFactory factory;
-  PortServer portServer;
-  bool readOnly;
+  final PortServer portServer;
+  final bool readOnly;
+
+  final SendPort port;
+  final bool primary;
+  final SqliteOpenFactory openFactory;
 
-  _SqliteConnectionParams(this.factory, this.portServer,
-      {required this.readOnly});
+  _SqliteConnectionParams(
+      {required this.openFactory,
+      required this.portServer,
+      required this.port,
+      required this.readOnly,
+      required this.primary});
 }
 
 class _SqliteIsolateStatement {

lib/src/sqlite_connection_impl.dart

@@ -1,14 +1,12 @@
 import 'dart:async';
 import 'dart:isolate';
 
-import 'package:sqlite_async/src/sqlite_open_factory.dart';
-
 import 'connection_pool.dart';
 import 'isolate_completer.dart';
 import 'mutex.dart';
 import 'sqlite_connection.dart';
-import 'sqlite_connection_factory.dart';
 import 'sqlite_connection_impl.dart';
+import 'sqlite_open_factory.dart';
 import 'sqlite_options.dart';
 import 'sqlite_queries.dart';
 import 'update_notification.dart';
@@ -24,7 +22,7 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
   final int maxReaders;
 
   /// Global lock to serialize write transactions.
-  final Mutex mutex = Mutex.shared();
+  final Mutex mutex = Mutex();
 
   /// Factory that opens a raw database connection in each isolate.
   ///
@@ -74,11 +72,13 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
   SqliteDatabase.withFactory({required this.openFactory, this.maxReaders = 5}) {
     updates = _updatesController.stream;
     _internalConnection = _openPrimaryConnection(debugName: 'sqlite-writer');
-    _pool = SqliteConnectionPool(_factory(),
+    _pool = SqliteConnectionPool(openFactory,
+        upstreamPort: _eventsPort.sendPort,
         updates: updates,
         writeConnection: _internalConnection,
         debugName: 'sqlite',
-        maxReaders: maxReaders);
+        maxReaders: maxReaders,
+        mutex: mutex);
 
     _listenForEvents();
 
@@ -129,23 +129,13 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
   }
 
   SqliteConnectionImpl _openPrimaryConnection({String? debugName}) {
-    return SqliteConnectionImpl(_factory(primary: true),
-        updates: updates, debugName: debugName);
-  }
-
-  /// Advanced: Get a connection factory.
-  ///
-  /// This factory can be passed to other isolates, to allow querying from
-  /// different isolates.
-  SqliteConnectionFactory connectionFactory() {
-    return _factory();
-  }
-
-  SqliteConnectionFactory _factory({bool primary = false}) {
-    return SqliteConnectionFactory(
-        port: _eventsPort.sendPort,
+    return SqliteConnectionImpl(
+        upstreamPort: _eventsPort.sendPort,
+        primary: true,
+        updates: updates,
+        debugName: debugName,
         mutex: mutex,
-        primary: primary,
+        readOnly: false,
         openFactory: openFactory);
   }
 

lib/src/sqlite_database.dart

@@ -1,8 +1,9 @@
-import 'sqlite_connection.dart';
-import 'update_notification.dart';
+import 'package:sqlite3/sqlite3.dart' as sqlite;
+
 import 'database_utils.dart';
+import 'sqlite_connection.dart';
 import 'throttle.dart';
-import 'package:sqlite3/sqlite3.dart' as sqlite;
+import 'update_notification.dart';
 
 /// Mixin to provide default query functionality.
 ///