sqlite web: Support requests to client

Author: simolus3

sqlite3_web/lib/src/client.dart

@@ -4,6 +4,7 @@ import 'dart:js_interop_unsafe';
 import 'dart:typed_data';
 
 import 'package:sqlite3/wasm.dart' hide WorkerOptions;
+import 'package:stream_channel/stream_channel.dart';
 import 'package:web/web.dart'
     hide Response, Request, FileSystem, Notification, Lock;
 
@@ -257,15 +258,21 @@ final class RemoteFileSystem implements FileSystem {
 final class WorkerConnection extends ProtocolChannel {
   final StreamController<Notification> notifications =
       StreamController.broadcast();
+  final Future<JSAny?> Function(JSAny?) handleCustomRequest;
 
-  WorkerConnection(super.channel) {
+  WorkerConnection(super.channel, this.handleCustomRequest) {
     closed.whenComplete(notifications.close);
   }
 
   @override
-  Future<Response> handleRequest(Request request) {
-    // TODO: implement handleRequest
-    throw UnimplementedError();
+  Future<Response> handleRequest(Request request) async {
+    switch (request) {
+      case CustomRequest(requestId: final id, :final payload):
+        final response = await handleCustomRequest(payload);
+        return SimpleSuccessResponse(response: response, requestId: id);
+      default:
+        throw UnimplementedError();
+    }
   }
 
   @override
@@ -278,6 +285,7 @@ final class DatabaseClient implements WebSqlite {
   final Uri workerUri;
   final Uri wasmUri;
   final DatabaseController _localController;
+  final Future<JSAny?> Function(JSAny?) _handleCustomRequest;
 
   final Lock _startWorkersLock = Lock();
   bool _startedWorkers = false;
@@ -289,7 +297,12 @@ final class DatabaseClient implements WebSqlite {
 
   final Set<MissingBrowserFeature> _missingFeatures = {};
 
-  DatabaseClient(this.workerUri, this.wasmUri, this._localController);
+  DatabaseClient(this.workerUri, this.wasmUri, this._localController,
+      Future<JSAny?> Function(JSAny?)? handleCustomRequest)
+      : _handleCustomRequest = handleCustomRequest ??
+            ((_) async {
+              throw StateError('No custom request handler installed');
+            });
 
   Future<void> startWorkers() {
     return _startWorkersLock.synchronized(() async {
@@ -303,6 +316,10 @@ final class DatabaseClient implements WebSqlite {
     });
   }
 
+  WorkerConnection _connection(StreamChannel<Message> channel) {
+    return WorkerConnection(channel, _handleCustomRequest);
+  }
+
   Future<void> _startDedicated() async {
     if (globalContext.has('Worker')) {
       final Worker dedicated;
@@ -319,8 +336,7 @@ final class DatabaseClient implements WebSqlite {
       final (endpoint, channel) = await createChannel();
       ConnectRequest(endpoint: endpoint, requestId: 0).sendToWorker(dedicated);
 
-      _connectionToDedicated =
-          WorkerConnection(channel.injectErrorsFrom(dedicated));
+      _connectionToDedicated = _connection(channel.injectErrorsFrom(dedicated));
     } else {
       _missingFeatures.add(MissingBrowserFeature.dedicatedWorkers);
     }
@@ -341,7 +357,7 @@ final class DatabaseClient implements WebSqlite {
       final (endpoint, channel) = await createChannel();
       ConnectRequest(endpoint: endpoint, requestId: 0).sendToPort(shared.port);
 
-      _connectionToShared = WorkerConnection(channel.injectErrorsFrom(shared));
+      _connectionToShared = _connection(channel.injectErrorsFrom(shared));
     } else {
       _missingFeatures.add(MissingBrowserFeature.sharedWorkers);
     }
@@ -358,7 +374,7 @@ final class DatabaseClient implements WebSqlite {
           ConnectRequest(requestId: 0, endpoint: endpoint),
           MessageType.simpleSuccessResponse);
 
-      return _connectionToDedicatedInShared = WorkerConnection(channel);
+      return _connectionToDedicatedInShared = _connection(channel);
     });
   }
 
@@ -374,7 +390,7 @@ final class DatabaseClient implements WebSqlite {
       local
           .addTopLevelMessage(ConnectRequest(requestId: 0, endpoint: endpoint));
 
-      return _connectionToLocal = WorkerConnection(channel);
+      return _connectionToLocal = _connection(channel);
     });
   }
 
@@ -512,7 +528,7 @@ final class DatabaseClient implements WebSqlite {
   }
 
   Future<Database> connectToExisting(SqliteWebEndpoint endpoint) async {
-    final channel = WorkerConnection(
+    final channel = _connection(
         WebEndpoint(port: endpoint.$1, lockName: endpoint.$2).connect());
 
     return RemoteDatabase(

sqlite3_web/lib/src/database.dart

@@ -123,8 +123,7 @@ abstract class ClientConnection {
   /// because the owning tab is closed.
   Future<void> get closed;
 
-  /// Sends a custom request __towards the client__. This is not currently
-  /// implemented.
+  /// Sends a custom request __towards the client__.
   Future<JSAny?> customRequest(JSAny? request);
 }
 
@@ -230,15 +229,21 @@ abstract class WebSqlite {
   /// The [controller] is used when connecting to a sqlite3 database without
   /// using workers. It should typically be the same implementation as the one
   /// passed to [workerEntrypoint].
+  ///
+  /// The optional [handleCustomRequest] function is invoked when the controller
+  /// sends a custom request to the client (via [ClientConnection.customRequest]).
+  /// If it's absent, the default is to throw an exception when called.
   static WebSqlite open({
     required Uri worker,
     required Uri wasmModule,
     DatabaseController? controller,
+    Future<JSAny?> Function(JSAny?)? handleCustomRequest,
   }) {
     return DatabaseClient(
       worker,
       wasmModule,
       controller ?? const _DefaultDatabaseController(),
+      handleCustomRequest,
     );
   }
 
@@ -253,9 +258,16 @@ abstract class WebSqlite {
   /// be valid as long as the original [Database] where [Database.additionalConnection]
   /// was called. This limitation does not exist for databases hosted by shared
   /// workers.
-  static Future<Database> connectToPort(SqliteWebEndpoint endpoint) {
-    final client =
-        DatabaseClient(Uri.base, Uri.base, const _DefaultDatabaseController());
+  ///
+  /// The optional [handleCustomRequest] function is invoked when the controller
+  /// sends a custom request to the client (via [ClientConnection.customRequest]).
+  /// If it's absent, the default is to throw an exception when called.
+  static Future<Database> connectToPort(
+    SqliteWebEndpoint endpoint, {
+    Future<JSAny?> Function(JSAny?)? handleCustomRequest,
+  }) {
+    final client = DatabaseClient(Uri.base, Uri.base,
+        const _DefaultDatabaseController(), handleCustomRequest);
     return client.connectToExisting(endpoint);
   }
 }

sqlite3_web/lib/src/worker.dart

@@ -354,8 +354,6 @@ final class _ClientConnection extends ProtocolChannel
     return SimpleSuccessResponse(response: null, requestId: request.requestId);
   }
 
-  void handleStreamCancelRequest() {}
-
   @override
   void handleNotification(Notification notification) {
     // There aren't supposed to be any notifications from the client.

sqlite3_web/test/integration_test.dart

@@ -209,6 +209,8 @@ final class _TestConfiguration {
         );
         await driver.assertFile(false);
 
+        expect(await driver.customRequest(), 42);
+
         await driver.execute('CREATE TABLE foo (bar TEXT);');
         var events = await driver.countEvents();
         expect(events.updates, 0);

sqlite3_web/tool/server.dart

@@ -210,6 +210,13 @@ class TestWebDriver {
     }
   }
 
+  Future<int> customRequest() async {
+    final res =
+        await driver.executeAsync('custom_request("", arguments[0])', []);
+
+    return res as int;
+  }
+
   Future<void> flush() async {
     final result = await driver.executeAsync('flush("", arguments[0])', []);
     if (result != true) {

sqlite3_web/web/controller.dart

@@ -43,6 +43,9 @@ final class ExampleDatabase extends WorkerDatabase {
   @override
   Future<JSAny?> handleCustomRequest(
       ClientConnection connection, JSAny? request) async {
-    return null;
+    final response =
+        ((await connection.customRequest(null)) as JSNumber).toDartInt;
+
+    return (2 * response).toJS;
   }
 }

sqlite3_web/web/main.dart

@@ -133,6 +133,9 @@ void main() {
 
     return null;
   });
+  _addCallbackForWebDriver('custom_request', (arg) async {
+    return database!.customRequest(null);
+  });
 
   document.getElementById('selfcheck')?.onClick.listen((event) async {
     print('starting');
@@ -173,6 +176,9 @@ WebSqlite initializeSqlite() {
     worker: workerUri,
     wasmModule: sqlite3WasmUri,
     controller: ExampleController(isInWorker: false),
+    handleCustomRequest: (request) async {
+      return 21.toJS;
+    },
   );
 }