test

Author: davidmartos96

packages/sqlite_async/lib/src/common/sqlite_database.dart

@@ -39,6 +39,12 @@ mixin SqliteDatabaseMixin implements SqliteConnection, SqliteQueries {
   ///
   /// Use this to access the database in background isolates.
   IsolateConnectionFactory isolateConnectionFactory();
+
+  /// Locks all underlying connections making up this database, and gives [block] access to all of them at once.
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block);
 }
 
 /// A SQLite database instance.

packages/sqlite_async/lib/src/impl/single_connection_database.dart

@@ -62,4 +62,9 @@ final class SingleConnectionDatabase
   List<SqliteConnection> get allConnections {
     return [connection];
   }
+  
+  @override
+  Future<T> withAllConnections<T>(Future<T> Function(SqliteWriteContext writer, List<SqliteReadContext> readers) block) {
+    return writeLock((_) => block(connection, []));
+  }
 }

packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart

@@ -69,4 +69,9 @@ class SqliteDatabaseImpl
   List<SqliteConnection> get allConnections {
     throw UnimplementedError();
   }
+  
+  @override
+  Future<T> withAllConnections<T>(Future<T> Function(SqliteWriteContext writer, List<SqliteReadContext> readers) block) {
+    throw UnimplementedError();
+  }
 }

packages/sqlite_async/lib/src/native/database/connection_pool.dart

@@ -284,11 +284,12 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
     }
 
     // Wait after all locks are taken
-    final contexts = await Future.wait([
+    final [writer as SqliteWriteContext, ...readers] = await Future.wait([
       writeLockedCompleter.future,
       ...readLockedCompleters.map((e) => e.future)
     ]);
-    return (contexts.first as SqliteWriteContext, contexts.sublist(1));
+
+    return (writer, readers);
   }
 }
 

packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart

@@ -176,4 +176,9 @@ class SqliteDatabaseImpl
   List<SqliteConnection> get allConnections {
     return _pool.allConnections;
   }
+  
+  @override
+  Future<T> withAllConnections<T>(Future<T> Function(SqliteWriteContext writer, List<SqliteReadContext> readers) block) {
+    return _pool.withAllConnections(block);
+  }
 }

packages/sqlite_async/lib/src/web/database.dart

@@ -176,6 +176,14 @@ class WebDatabase
   List<SqliteConnection> get allConnections {
     return [this];
   }
+
+  @override
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block) {
+    return writeLock((_) => block(this, []));
+  }
 }
 
 final class _UnscopedContext extends UnscopedContext {

packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart

@@ -183,4 +183,12 @@ class SqliteDatabaseImpl
   List<SqliteConnection> get allConnections {
     return [_connection];
   }
+
+  @override
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block) {
+    return writeLock((_) => block(_connection, []));
+  }
 }

packages/sqlite_async/test/native/basic_test.dart

@@ -4,7 +4,9 @@ library;
 import 'dart:async';
 import 'dart:math';
 
+import 'package:collection/collection.dart';
 import 'package:sqlite3/common.dart' as sqlite;
+import 'package:sqlite3/sqlite3.dart' show Row;
 import 'package:sqlite_async/sqlite_async.dart';
 import 'package:test/test.dart';
 
@@ -76,25 +78,56 @@ void main() {
       await db.initialize();
       await createTables(db);
 
-      print("${DateTime.now()} start");
 
-      final withAllConnsFut = () async {
-        await Future.delayed(const Duration(milliseconds: 20));
-        await db.withAllConnections((writer, readers) async {
-          print("${DateTime.now()} in withAllConnections");
-          await Future.delayed(const Duration(seconds: 5));
-        });
-      }();
+      Future<Row> readWithRandomDelay(SqliteReadContext ctx, int id) async {
+        return await ctx.get(
+            'SELECT ? as i, test_sleep(?) as sleep, test_connection_name() as connection',
+            [id, 5 + Random().nextInt(10)]);
+      }
 
-      var readFutures = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11].map((i) => db.get(
-          'SELECT ? as i, test_sleep(?) as sleep, test_connection_name() as connection',
-          [i, 5 + Random().nextInt(10)]));
+      // Warm up to spawn the max readers
+      await Future.wait(
+        [1, 2, 3, 4, 5, 6, 7, 8].map((i) => readWithRandomDelay(db, i)),
+      );
 
-      final futures = [...readFutures, withAllConnsFut];
+      bool finishedWithAllConns = false;
 
-      await for (var result in Stream.fromFutures(futures)) {
-        print("${DateTime.now()} $result");
-      }
+      late Future<void> readsCalledWhileWithAllConnsRunning;
+
+      print("${DateTime.now()} start");
+      await db.withAllConnections((writer, readers) async {
+        assert(readers.length == 3);
+
+        // Run some reads during the block that they should run after the block finishes and releases
+        // all locks
+        readsCalledWhileWithAllConnsRunning = Future.wait(
+          [1, 2, 3, 4, 5, 6, 7, 8].map((i) async {
+            final r = await db.readLock((c) async {
+              expect(finishedWithAllConns, isTrue);
+              return await readWithRandomDelay(c, i);
+            });
+            print(
+                "${DateTime.now()} After withAllConnections, started while running $r");
+          }),
+        );
+
+        await Future.wait([
+          writer.execute(
+              "INSERT OR REPLACE INTO test_data(id, description) SELECT ? as i, test_sleep(?) || ' ' || test_connection_name() || ' 1 ' || datetime() as connection RETURNING *",
+              [
+                123,
+                5 + Random().nextInt(20)
+              ]).then((value) =>
+              print("${DateTime.now()} withAllConnections writer done $value")),
+          ...readers
+              .mapIndexed((i, r) => readWithRandomDelay(r, i).then((results) {
+                    print(
+                        "${DateTime.now()} withAllConnections readers done $results");
+                  }))
+        ]);
+      }).then((_) => finishedWithAllConns = true);
+
+      await readsCalledWhileWithAllConnsRunning;
     });
 
     test('Concurren 2', () async {