[Alpha WIP] Open Factory SqliteConnections

lib/src/common/abstract_open_factory.dart

@@ -2,7 +2,11 @@ import 'dart:async';
 import 'package:meta/meta.dart';
 
 import 'package:sqlite_async/sqlite3_common.dart' as sqlite;
+import 'package:sqlite_async/src/common/mutex.dart';
+import 'package:sqlite_async/src/common/port_channel.dart';
+import 'package:sqlite_async/src/sqlite_connection.dart';
 import 'package:sqlite_async/src/sqlite_options.dart';
+import 'package:sqlite_async/src/update_notification.dart';
 
 /// Factory to create new SQLite database connections.
 ///
@@ -11,7 +15,11 @@ import 'package:sqlite_async/src/sqlite_options.dart';
 abstract class SqliteOpenFactory<Database extends sqlite.CommonDatabase> {
   String get path;
 
+  /// Opens a direct connection to the SQLite database
   FutureOr<Database> open(SqliteOpenOptions options);
+
+  /// Opens an asynchronous [SqliteConnection]
+  FutureOr<SqliteConnection> openConnection(SqliteOpenOptions options);
 }
 
 class SqliteOpenOptions {
@@ -21,8 +29,24 @@ class SqliteOpenOptions {
   /// Whether this connection is read-only.
   final bool readOnly;
 
+  /// Mutex to use in [SqliteConnection]s
+  final Mutex? mutex;
+
+  /// Name used in debug logs
+  final String? debugName;
+
+  final SerializedPortClient? upstreamPort;
+
+  /// Stream of external update notifications
+  final Stream<UpdateNotification>? updates;
+
   const SqliteOpenOptions(
-      {required this.primaryConnection, required this.readOnly});
+      {required this.primaryConnection,
+      required this.readOnly,
+      this.mutex,
+      this.debugName,
+      this.updates,
+      this.upstreamPort});
 
   sqlite.OpenMode get openMode {
     if (primaryConnection) {
@@ -55,9 +79,14 @@ abstract class AbstractDefaultSqliteOpenFactory<
   List<String> pragmaStatements(SqliteOpenOptions options);
 
   @protected
+
+  /// Opens a direct connection to a SQLite database connection
   FutureOr<Database> openDB(SqliteOpenOptions options);
 
   @override
+
+  /// Opens a direct connection to a SQLite database connection
+  /// and executes setup pragma statements to initialize the DB
   FutureOr<Database> open(SqliteOpenOptions options) async {
     var db = await openDB(options);
 
@@ -66,4 +95,10 @@ abstract class AbstractDefaultSqliteOpenFactory<
     }
     return db;
   }
+
+  @override
+
+  /// Opens an asynchronous [SqliteConnection] to a SQLite database
+  /// and executes setup pragma statements to initialize the DB
+  FutureOr<SqliteConnection> openConnection(SqliteOpenOptions options);
 }

lib/src/impl/stub_sqlite_open_factory.dart

@@ -1,5 +1,8 @@
+import 'dart:async';
+
 import 'package:sqlite_async/sqlite3_common.dart';
 import 'package:sqlite_async/src/common/abstract_open_factory.dart';
+import 'package:sqlite_async/src/sqlite_connection.dart';
 import 'package:sqlite_async/src/sqlite_options.dart';
 
 class DefaultSqliteOpenFactory extends AbstractDefaultSqliteOpenFactory {
@@ -16,4 +19,9 @@ class DefaultSqliteOpenFactory extends AbstractDefaultSqliteOpenFactory {
   List<String> pragmaStatements(SqliteOpenOptions options) {
     throw UnimplementedError();
   }
+
+  @override
+  FutureOr<SqliteConnection> openConnection(SqliteOpenOptions options) {
+    throw UnimplementedError();
+  }
 }

lib/src/native/database/connection_pool.dart

@@ -117,21 +117,15 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
 
   @override
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
-      {Duration? lockTimeout, String? debugContext}) {
+      {Duration? lockTimeout, String? debugContext}) async {
     if (closed) {
       throw AssertionError('Closed');
     }
     if (_writeConnection?.closed == true) {
       _writeConnection = null;
     }
-    _writeConnection ??= SqliteConnectionImpl(
-        upstreamPort: _upstreamPort,
-        primary: false,
-        updates: updates,
-        debugName: debugName != null ? '$debugName-writer' : null,
-        mutex: mutex,
-        readOnly: false,
-        openFactory: _factory);
+    _writeConnection ??= await _openPrimaryConnection(
+        debugName: debugName != null ? '$debugName-writer' : null);
     return _runZoned(() {
       return _writeConnection!.writeLock(callback,
           lockTimeout: lockTimeout, debugContext: debugContext);
@@ -192,4 +186,14 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
     // read-only connections first.
     await _writeConnection?.close();
   }
+
+  FutureOr<SqliteConnection> _openPrimaryConnection({String? debugName}) {
+    return _factory.openConnection(SqliteOpenOptions(
+        upstreamPort: _upstreamPort,
+        primaryConnection: true,
+        updates: updates,
+        debugName: debugName,
+        mutex: mutex,
+        readOnly: false));
+  }
 }

lib/src/native/database/native_sqlite_database.dart

@@ -5,7 +5,6 @@ import 'package:sqlite_async/src/common/abstract_open_factory.dart';
 import 'package:sqlite_async/src/common/sqlite_database.dart';
 import 'package:sqlite_async/src/common/port_channel.dart';
 import 'package:sqlite_async/src/native/database/connection_pool.dart';
-import 'package:sqlite_async/src/native/database/native_sqlite_connection_impl.dart';
 import 'package:sqlite_async/src/native/native_isolate_connection_factory.dart';
 import 'package:sqlite_async/src/native/native_isolate_mutex.dart';
 import 'package:sqlite_async/src/native/native_sqlite_open_factory.dart';
@@ -33,11 +32,11 @@ class SqliteDatabaseImpl
   int maxReaders;
 
   @override
-  late Future<void> isInitialized;
+  // Native doesn't require any asynchronous initialization
+  late Future<void> isInitialized = Future.value();
 
   late final PortServer _eventsPort;
 
-  late final SqliteConnectionImpl _internalConnection;
   late final SqliteConnectionPool _pool;
 
   /// Global lock to serialize write transactions.
@@ -77,20 +76,12 @@ class SqliteDatabaseImpl
 
     _listenForEvents();
 
-    _internalConnection = _openPrimaryConnection(debugName: 'sqlite-writer');
     _pool = SqliteConnectionPool(openFactory,
         upstreamPort: _eventsPort.client(),
         updates: updates,
-        writeConnection: _internalConnection,
         debugName: 'sqlite',
         maxReaders: maxReaders,
         mutex: mutex);
-
-    isInitialized = _init();
-  }
-
-  Future<void> _init() async {
-    await _internalConnection.ready;
   }
 
   @override
@@ -160,17 +151,6 @@ class SqliteDatabaseImpl
         upstreamPort: _eventsPort.client());
   }
 
-  SqliteConnectionImpl _openPrimaryConnection({String? debugName}) {
-    return SqliteConnectionImpl(
-        upstreamPort: _eventsPort.client(),
-        primary: true,
-        updates: updates,
-        debugName: debugName,
-        mutex: mutex,
-        readOnly: false,
-        openFactory: openFactory);
-  }
-
   @override
   Future<void> close() async {
     await _pool.close();

lib/src/native/native_sqlite_open_factory.dart

@@ -2,6 +2,8 @@ import 'package:sqlite3/sqlite3.dart';
 import 'package:sqlite_async/sqlite3_common.dart';
 
 import 'package:sqlite_async/src/common/abstract_open_factory.dart';
+import 'package:sqlite_async/src/native/database/native_sqlite_connection_impl.dart';
+import 'package:sqlite_async/src/sqlite_connection.dart';
 import 'package:sqlite_async/src/sqlite_options.dart';
 
 /// Native implementation of [AbstractDefaultSqliteOpenFactory]
@@ -37,4 +39,17 @@ class DefaultSqliteOpenFactory extends AbstractDefaultSqliteOpenFactory {
     }
     return statements;
   }
+
+  @override
+  SqliteConnection openConnection(SqliteOpenOptions options) {
+    return SqliteConnectionImpl(
+      primary: options.primaryConnection,
+      readOnly: options.readOnly,
+      mutex: options.mutex!,
+      upstreamPort: options.upstreamPort!,
+      debugName: options.debugName,
+      updates: options.updates,
+      openFactory: this,
+    );
+  }
 }

lib/src/sqlite_connection.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 
 import 'package:sqlite3/common.dart' as sqlite;
+import 'package:sqlite_async/src/update_notification.dart';
 
 /// Abstract class representing calls available in a read-only or read-write context.
 abstract class SqliteReadContext {
@@ -74,6 +75,9 @@ abstract class SqliteWriteContext extends SqliteReadContext {
 
 /// Abstract class representing a connection to the SQLite database.
 abstract class SqliteConnection extends SqliteWriteContext {
+  /// Reports table change update notification
+  Stream<UpdateNotification>? get updates;
+
   /// Open a read-only transaction.
   ///
   /// Statements within the transaction must be done on the provided