Initial package.

Author: rkistner

.gitignore

@@ -0,0 +1,9 @@
+# https://dart.dev/guides/libraries/private-files
+# Created by `dart pub`
+.dart_tool/
+
+# Avoid committing pubspec.lock for library packages; see
+# https://dart.dev/guides/libraries/private-files#pubspeclock.
+pubspec.lock
+
+.idea

CHANGELOG.md

@@ -0,0 +1,3 @@
+## 0.1.0
+
+- Initial version.

README.md

@@ -0,0 +1,20 @@
+# sqlite_async 
+
+High-performance asynchronous interface for SQLite on Dart & Flutter.
+
+[SQLite](https://www.sqlite.org/) is small, fast, has a lot of built-in functionality, and works
+great as an in-app database. However, SQLite is designed for many different use cases, and requires
+some configuration for optimal performance as an in-app database.
+
+The [sqlite3](https://pub.dev/packages/sqlite3) Dart bindings are great for direct synchronous access
+to a SQLite database, but leaves the configuration the developer.
+
+This library wraps the bindings and configures the database with a good set of defaults, with
+all database calls being asynchronous to avoid blocking the UI, while still providing direct SQL
+query access.
+
+## Features
+
+ * Fast.
+ * Direct SQL query access.
+ * Uses a connection pool to allow concurrent queries.

analysis_options.yaml

@@ -0,0 +1 @@
+include: package:lints/recommended.yaml

lib/sqlite_async.dart

@@ -0,0 +1,10 @@
+/// Support for doing something awesome.
+///
+/// More dartdocs go here.
+library sqlite_dart;
+
+export 'src/sqlite_connection.dart';
+export 'src/update_notification.dart';
+export 'src/sqlite_database.dart';
+export 'src/sqlite_options.dart';
+export 'src/sqlite_open_factory.dart';

lib/src/connection_pool.dart

@@ -0,0 +1,123 @@
+import 'dart:async';
+
+import 'sqlite_connection.dart';
+import 'sqlite_connection_factory.dart';
+import 'sqlite_connection_impl.dart';
+import 'sqlite_queries.dart';
+import 'update_notification.dart';
+
+/// A connection pool with a single write connection and multiple read connections.
+class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
+  SqliteConnection? _writeConnection;
+
+  final List<SqliteConnectionImpl> _readConnections = [];
+
+  final SqliteConnectionFactory _factory;
+
+  @override
+  final Stream<UpdateNotification>? updates;
+
+  final int maxReaders;
+
+  final String? debugName;
+
+  /// Open a new connection pool.
+  ///
+  /// The provided factory is used to open connections on demand. Connections
+  /// are only opened when requested for the first time.
+  ///
+  /// [maxReaders] specifies the maximum number of read connections.
+  /// A maximum of one write connection will be opened.
+  ///
+  /// Read connections are opened in read-only mode, and will reject any statements
+  /// that modify the database.
+  SqliteConnectionPool(this._factory,
+      {this.updates,
+      this.maxReaders = 5,
+      SqliteConnection? writeConnection,
+      this.debugName})
+      : _writeConnection = writeConnection;
+
+  @override
+  Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
+      {Duration? lockTimeout}) async {
+    await _expandPool();
+
+    bool haveLock = false;
+    var completer = Completer<T>();
+
+    var futures = _readConnections.map((connection) async {
+      try {
+        return await connection.readLock((ctx) async {
+          if (haveLock) {
+            // Already have a different lock - release this one.
+            return false;
+          }
+          haveLock = true;
+
+          var future = callback(ctx);
+          completer.complete(future);
+
+          // We have to wait for the future to complete before we can release the
+          // lock.
+          try {
+            await future;
+          } catch (_) {
+            // Ignore
+          }
+
+          return true;
+        }, lockTimeout: lockTimeout);
+      } on TimeoutException {
+        return false;
+      }
+    });
+
+    final stream = Stream<bool>.fromFutures(futures);
+    var gotAny = await stream.any((element) => element);
+
+    if (!gotAny) {
+      // All TimeoutExceptions
+      throw TimeoutException('Failed to get a read connection', lockTimeout);
+    }
+
+    try {
+      return await completer.future;
+    } catch (e) {
+      // throw e;
+      rethrow;
+    }
+  }
+
+  @override
+  Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
+      {Duration? lockTimeout}) {
+    _writeConnection ??= _factory.openConnection(
+        debugName: debugName != null ? '$debugName-writer' : null);
+    return _writeConnection!.writeLock(callback, lockTimeout: lockTimeout);
+  }
+
+  Future<void> _expandPool() async {
+    if (_readConnections.length >= maxReaders) {
+      return;
+    }
+    bool hasCapacity = _readConnections.any((connection) => !connection.locked);
+    if (!hasCapacity) {
+      var name = debugName == null
+          ? null
+          : '$debugName-${_readConnections.length + 1}';
+      var connection = _factory.openConnection(
+          updates: updates,
+          debugName: name,
+          readOnly: true) as SqliteConnectionImpl;
+      _readConnections.add(connection);
+
+      // Edge case:
+      // If we don't await here, there is a chance that a different connection
+      // is used for the transaction, and that it finishes and deletes the database
+      // while this one is still opening. This is specifically triggered in tests.
+      // To avoid that, we wait for the connection to be ready.
+      await connection.ready;
+    }
+  }
+}

lib/src/database_utils.dart

@@ -0,0 +1,89 @@
+import 'dart:async';
+
+import 'package:sqlite3/sqlite3.dart' as sqlite;
+
+import 'sqlite_connection.dart';
+
+Future<T> internalReadTransaction<T>(SqliteReadContext ctx,
+    Future<T> Function(SqliteReadContext tx) callback) async {
+  try {
+    await ctx.getAll('BEGIN');
+    final result = await callback(ctx);
+    await ctx.getAll('END TRANSACTION');
+    return result;
+  } catch (e) {
+    try {
+      await ctx.getAll('ROLLBACK');
+    } catch (e) {
+      // In rare cases, a ROLLBACK may fail.
+      // Safe to ignore.
+    }
+    rethrow;
+  }
+}
+
+Future<T> internalWriteTransaction<T>(SqliteWriteContext ctx,
+    Future<T> Function(SqliteWriteContext tx) callback) async {
+  try {
+    await ctx.execute('BEGIN IMMEDIATE');
+    final result = await callback(ctx);
+    await ctx.execute('COMMIT');
+    return result;
+  } catch (e) {
+    try {
+      await ctx.execute('ROLLBACK');
+    } catch (e) {
+      // In rare cases, a ROLLBACK may fail.
+      // Safe to ignore.
+    }
+    rethrow;
+  }
+}
+
+Future<T> asyncDirectTransaction<T>(sqlite.Database db,
+    FutureOr<T> Function(sqlite.Database db) callback) async {
+  for (var i = 50; i >= 0; i--) {
+    try {
+      db.execute('BEGIN IMMEDIATE');
+      late T result;
+      try {
+        result = await callback(db);
+        db.execute('COMMIT');
+      } catch (e) {
+        try {
+          db.execute('ROLLBACK');
+        } catch (e2) {
+          // Safe to ignore
+        }
+        rethrow;
+      }
+
+      return result;
+    } catch (e) {
+      if (e is sqlite.SqliteException) {
+        if (e.resultCode == 5 && i != 0) {
+          // SQLITE_BUSY
+          await Future.delayed(const Duration(milliseconds: 50));
+          continue;
+        }
+      }
+      rethrow;
+    }
+  }
+  throw AssertionError('Should not reach this');
+}
+
+/// Given a SELECT query, return the tables that the query depends on.
+Future<Set<String>> getSourceTables(SqliteReadContext ctx, String sql) async {
+  final rows = await ctx.getAll('EXPLAIN QUERY PLAN $sql');
+  Set<String> tables = {};
+  final re = RegExp(r'^(SCAN|SEARCH)( TABLE)? (.+?)( USING .+)?$');
+  for (var row in rows) {
+    final detail = row['detail'];
+    final match = re.firstMatch(detail);
+    if (match != null) {
+      tables.add(match.group(3)!);
+    }
+  }
+  return tables;
+}

lib/src/isolate_completer.dart

@@ -0,0 +1,115 @@
+import 'dart:async';
+import 'dart:isolate';
+
+class IsolateResult<T> {
+  final ReceivePort receivePort = ReceivePort();
+  late Future<T> future;
+
+  IsolateResult() {
+    final sendResult = receivePort.first;
+    sendResult.whenComplete(() {
+      receivePort.close();
+    });
+
+    future = sendResult.then((response) {
+      if (response is PortResult) {
+        return response.value as T;
+      } else if (response == abortedResponse) {
+        throw const IsolateTerminatedError();
+      } else {
+        throw AssertionError('Invalid response: $response');
+      }
+    });
+  }
+
+  PortCompleter<T> get completer {
+    return PortCompleter(receivePort.sendPort);
+  }
+
+  close() {
+    receivePort.close();
+  }
+}
+
+const abortedResponse = 'aborted';
+
+class PortCompleter<T> {
+  final SendPort sendPort;
+
+  PortCompleter(this.sendPort);
+
+  void complete([FutureOr<T>? value]) {
+    sendPort.send(PortResult.success(value));
+  }
+
+  void completeError(Object error, [StackTrace? stackTrace]) {
+    sendPort.send(PortResult.error(error, stackTrace));
+  }
+
+  addExitHandler() {
+    Isolate.current.addOnExitListener(sendPort, response: abortedResponse);
+  }
+
+  Future<void> handle(FutureOr<T> Function() callback,
+      {bool ignoreStackTrace = false}) async {
+    addExitHandler();
+    try {
+      final result = await callback();
+      complete(result);
+    } catch (error, stacktrace) {
+      if (ignoreStackTrace) {
+        completeError(error);
+      } else {
+        completeError(error, stacktrace);
+      }
+    }
+  }
+}
+
+class PortResult<T> {
+  final bool success;
+  final T? _result;
+  final Object? _error;
+  final StackTrace? stackTrace;
+
+  const PortResult.success(T result)
+      : success = true,
+        _error = null,
+        stackTrace = null,
+        _result = result;
+  const PortResult.error(Object error, [this.stackTrace])
+      : success = false,
+        _result = null,
+        _error = error;
+
+  T get value {
+    if (success) {
+      return _result as T;
+    } else {
+      if (_error != null && stackTrace != null) {
+        Error.throwWithStackTrace(_error!, stackTrace!);
+      } else {
+        throw _error!;
+      }
+    }
+  }
+
+  T get result {
+    assert(success);
+    return _result as T;
+  }
+
+  Object get error {
+    assert(!success);
+    return _error!;
+  }
+}
+
+class IsolateTerminatedError implements Error {
+  const IsolateTerminatedError();
+
+  @override
+  StackTrace? get stackTrace {
+    return null;
+  }
+}

lib/src/log.dart

@@ -0,0 +1,3 @@
+import 'package:logging/logging.dart';
+
+final log = Logger('SQLite');