Add tests to sqlite3_web

Author: simolus3

sqlite3_web/lib/src/client.dart

@@ -14,6 +14,11 @@ import 'protocol.dart';
 import 'shared.dart';
 import 'worker.dart';
 
+final class _CommitOrRollbackStream {
+  StreamSubscription<Notification>? workerSubscription;
+  final StreamController<void> controller = StreamController.broadcast();
+}
+
 final class RemoteDatabase implements Database {
   final WorkerConnection connection;
   final int databaseId;
@@ -22,10 +27,9 @@ final class RemoteDatabase implements Database {
 
   StreamSubscription<Notification>? _updateNotificationSubscription;
   final StreamController<SqliteUpdate> _updates = StreamController.broadcast();
-  StreamSubscription<Notification>? _rollbackNotificationSubscription;
-  final StreamController<void> _rollbacks = StreamController.broadcast();
-  StreamSubscription<Notification>? _commitNotificationSubscription;
-  final StreamController<void> _commits = StreamController.broadcast();
+
+  final _CommitOrRollbackStream _commits = _CommitOrRollbackStream();
+  final _CommitOrRollbackStream _rollbacks = _CommitOrRollbackStream();
 
   RemoteDatabase({required this.connection, required this.databaseId}) {
     _updates
@@ -38,76 +42,54 @@ final class RemoteDatabase implements Database {
             }
           }
         });
-        _requestUpdates(true);
+        _requestStreamUpdates(MessageType.updateRequest, true);
       })
       ..onCancel = (() {
         _updateNotificationSubscription?.cancel();
         _updateNotificationSubscription = null;
-        _requestUpdates(false);
+        _requestStreamUpdates(MessageType.updateRequest, false);
       });
 
-    _rollbacks
-      ..onListen = (() {
-        _rollbackNotificationSubscription ??=
-            connection.notifications.stream.listen((notification) {
-          if (notification case RollbackNotification()) {
-            if (notification.databaseId == databaseId) {
-              _rollbacks.add(null);
-            }
-          }
-        });
-        _requestRollbacks(true);
-      })
-      ..onCancel = (() {
-        _rollbackNotificationSubscription?.cancel();
-        _rollbackNotificationSubscription = null;
-        _requestRollbacks(false);
-      });
+    _setupCommitOrRollbackStream(
+        _commits, MessageType.commitRequest, MessageType.notifyCommit);
+    _setupCommitOrRollbackStream(
+        _rollbacks, MessageType.rollbackRequest, MessageType.notifyRollback);
+  }
 
-    _commits
+  void _setupCommitOrRollbackStream(
+    _CommitOrRollbackStream stream,
+    MessageType requestSubscription,
+    MessageType notificationType,
+  ) {
+    stream.controller
       ..onListen = (() {
-        _commitNotificationSubscription ??=
+        stream.workerSubscription ??=
             connection.notifications.stream.listen((notification) {
-          if (notification case CommitNotification()) {
-            if (notification.databaseId == databaseId) {
-              _commits.add(null);
+          if (notification case EmptyNotification(type: final type)) {
+            if (notification.databaseId == databaseId &&
+                type == notificationType) {
+              stream.controller.add(null);
             }
           }
         });
-        _requestCommits(true);
+        _requestStreamUpdates(requestSubscription, true);
       })
       ..onCancel = (() {
-        _commitNotificationSubscription?.cancel();
-        _commitNotificationSubscription = null;
-        _requestCommits(false);
+        stream.workerSubscription?.cancel();
+        stream.workerSubscription = null;
+        _requestStreamUpdates(requestSubscription, false);
       });
   }
 
-  void _requestUpdates(bool sendUpdates) {
+  void _requestStreamUpdates(MessageType streamType, bool subscribe) {
     if (!_isClosed) {
       connection.sendRequest(
-        UpdateStreamRequest(
-            action: sendUpdates, requestId: 0, databaseId: databaseId),
-        MessageType.simpleSuccessResponse,
-      );
-    }
-  }
-
-  void _requestRollbacks(bool sendRollbacks) {
-    if (!_isClosed) {
-      connection.sendRequest(
-        RollbackStreamRequest(
-            action: sendRollbacks, requestId: 1, databaseId: databaseId),
-        MessageType.simpleSuccessResponse,
-      );
-    }
-  }
-
-  void _requestCommits(bool sendCommits) {
-    if (!_isClosed) {
-      connection.sendRequest(
-        CommitStreamRequest(
-            action: sendCommits, requestId: 2, databaseId: databaseId),
+        StreamRequest(
+          type: streamType,
+          action: subscribe,
+          requestId: 0, // filled out in sendRequest
+          databaseId: databaseId,
+        ),
         MessageType.simpleSuccessResponse,
       );
     }
@@ -123,8 +105,8 @@ final class RemoteDatabase implements Database {
     _isClosed = true;
     await (
       _updates.close(),
-      _rollbacks.close(),
-      _commits.close(),
+      _rollbacks.controller.close(),
+      _commits.controller.close(),
       connection.sendRequest(
           CloseDatabase(requestId: 0, databaseId: databaseId),
           MessageType.simpleSuccessResponse)
@@ -190,10 +172,10 @@ final class RemoteDatabase implements Database {
   Stream<SqliteUpdate> get updates => _updates.stream;
 
   @override
-  Stream<void> get rollbacks => _rollbacks.stream;
+  Stream<void> get rollbacks => _rollbacks.controller.stream;
 
   @override
-  Stream<void> get commits => _commits.stream;
+  Stream<void> get commits => _commits.controller.stream;
 
   @override
   Future<int> get userVersion async {

sqlite3_web/lib/src/protocol.dart

@@ -25,18 +25,18 @@ enum MessageType<T extends Message> {
   fileSystemFlush<FileSystemFlushRequest>(),
   connect<ConnectRequest>(),
   startFileSystemServer<StartFileSystemServer>(),
-  updateRequest<UpdateStreamRequest>(),
-  rollbackRequest<RollbackStreamRequest>(),
-  commitRequest<CommitStreamRequest>(),
+  updateRequest<StreamRequest>(),
+  rollbackRequest<StreamRequest>(),
+  commitRequest<StreamRequest>(),
   simpleSuccessResponse<SimpleSuccessResponse>(),
   rowsResponse<RowsResponse>(),
   errorResponse<ErrorResponse>(),
   endpointResponse<EndpointResponse>(),
   closeDatabase<CloseDatabase>(),
   openAdditionalConnection<OpenAdditonalConnection>(),
   notifyUpdate<UpdateNotification>(),
-  notifyRollback<RollbackNotification>(),
-  notifyCommit<CommitNotification>(),
+  notifyRollback<EmptyNotification>(),
+  notifyCommit<EmptyNotification>(),
   ;
 
   static final Map<String, MessageType> byName = values.asNameMap();
@@ -97,17 +97,19 @@ sealed class Message {
       MessageType.closeDatabase => CloseDatabase.deserialize(object),
       MessageType.openAdditionalConnection =>
         OpenAdditonalConnection.deserialize(object),
-      MessageType.updateRequest => UpdateStreamRequest.deserialize(object),
-      MessageType.rollbackRequest => RollbackStreamRequest.deserialize(object),
-      MessageType.commitRequest => CommitStreamRequest.deserialize(object),
+      MessageType.updateRequest ||
+      MessageType.rollbackRequest ||
+      MessageType.commitRequest =>
+        StreamRequest.deserialize(type, object),
       MessageType.simpleSuccessResponse =>
         SimpleSuccessResponse.deserialize(object),
       MessageType.endpointResponse => EndpointResponse.deserialize(object),
       MessageType.rowsResponse => RowsResponse.deserialize(object),
       MessageType.errorResponse => ErrorResponse.deserialize(object),
       MessageType.notifyUpdate => UpdateNotification.deserialize(object),
-      MessageType.notifyRollback => RollbackNotification.deserialize(object),
-      MessageType.notifyCommit => CommitNotification.deserialize(object),
+      MessageType.notifyRollback ||
+      MessageType.notifyCommit =>
+        EmptyNotification.deserialize(type, object),
     };
   }
 
@@ -637,92 +639,33 @@ final class ErrorResponse extends Response {
   }
 }
 
-final class UpdateStreamRequest extends Request {
+final class StreamRequest extends Request {
   /// When true, the client is requesting to be informed about updates happening
   /// on the database identified by this request.
   ///
   /// When false, the client is requesting to no longer be informed about these
   /// updates.
   final bool action;
 
-  UpdateStreamRequest(
-      {required this.action,
-      required super.requestId,
-      required super.databaseId});
+  final MessageType<Message> type;
 
-  factory UpdateStreamRequest.deserialize(JSObject object) {
-    return UpdateStreamRequest(
-      action: (object[_UniqueFieldNames.action] as JSBoolean).toDart,
-      requestId: object.requestId,
-      databaseId: object.databaseId,
-    );
-  }
-
-  @override
-  MessageType<Message> get type => MessageType.updateRequest;
-
-  @override
-  void serialize(JSObject object, List<JSObject> transferred) {
-    super.serialize(object, transferred);
-    object[_UniqueFieldNames.action] = action.toJS;
-  }
-}
-
-final class RollbackStreamRequest extends Request {
-  /// When true, the client is requesting to be informed about rollbacks
-  /// happening on the database identified by this request.
-  ///
-  /// When false, the client is requesting to no longer be informed about these
-  /// updates.
-  final bool action;
-
-  RollbackStreamRequest(
-      {required this.action,
-      required super.requestId,
-      required super.databaseId});
-
-  factory RollbackStreamRequest.deserialize(JSObject object) {
-    return RollbackStreamRequest(
-      action: (object[_UniqueFieldNames.action] as JSBoolean).toDart,
-      requestId: object.requestId,
-      databaseId: object.databaseId,
-    );
-  }
-
-  @override
-  MessageType<Message> get type => MessageType.rollbackRequest;
-
-  @override
-  void serialize(JSObject object, List<JSObject> transferred) {
-    super.serialize(object, transferred);
-    object[_UniqueFieldNames.action] = action.toJS;
-  }
-}
-
-final class CommitStreamRequest extends Request {
-  /// When true, the client is requesting to be informed about rollbacks
-  /// happening on the database identified by this request.
-  ///
-  /// When false, the client is requesting to no longer be informed about these
-  /// updates.
-  final bool action;
-
-  CommitStreamRequest(
-      {required this.action,
-      required super.requestId,
-      required super.databaseId});
+  StreamRequest({
+    required this.type,
+    required this.action,
+    required super.requestId,
+    required super.databaseId,
+  });
 
-  factory CommitStreamRequest.deserialize(JSObject object) {
-    return CommitStreamRequest(
+  factory StreamRequest.deserialize(
+      MessageType<Message> type, JSObject object) {
+    return StreamRequest(
+      type: type,
       action: (object[_UniqueFieldNames.action] as JSBoolean).toDart,
       requestId: object.requestId,
       databaseId: object.databaseId,
     );
   }
 
-  @override
-  MessageType<Message> get type => MessageType.rollbackRequest;
-
   @override
   void serialize(JSObject object, List<JSObject> transferred) {
     super.serialize(object, transferred);
@@ -886,41 +829,23 @@ final class UpdateNotification extends Notification {
   }
 }
 
-final class RollbackNotification extends Notification {
+/// Used as a notification without a payload, e.g. for commit or rollback
+/// events.
+final class EmptyNotification extends Notification {
   final int databaseId;
-
-  RollbackNotification({required this.databaseId});
-
-  factory RollbackNotification.deserialize(JSObject object) {
-    return RollbackNotification(
-      databaseId: object.databaseId,
-    );
-  }
-
-  @override
-  MessageType<Message> get type => MessageType.notifyUpdate;
-
   @override
-  void serialize(JSObject object, List<JSObject> transferred) {
-    super.serialize(object, transferred);
-    object[_UniqueFieldNames.databaseId] = databaseId.toJS;
-  }
-}
-
-final class CommitNotification extends Notification {
-  final int databaseId;
+  final MessageType<Message> type;
 
-  CommitNotification({required this.databaseId});
+  EmptyNotification({required this.type, required this.databaseId});
 
-  factory CommitNotification.deserialize(JSObject object) {
-    return CommitNotification(
+  factory EmptyNotification.deserialize(
+      MessageType<Message> type, JSObject object) {
+    return EmptyNotification(
+      type: type,
       databaseId: object.databaseId,
     );
   }
 
-  @override
-  MessageType<Message> get type => MessageType.notifyCommit;
-
   @override
   void serialize(JSObject object, List<JSObject> transferred) {
     super.serialize(object, transferred);