add an experimental websockets api between client and server

pkgs/dart_services/lib/server.dart

@@ -146,6 +146,10 @@ Middleware exceptionResponse() {
     return (Request request) async {
       try {
         return await handler(request);
+      } on HijackException {
+        // We ignore hijack exceptions as they are not error conditions; they're
+        // used used for control flow when upgrading websocket connections.
+        rethrow;
       } catch (e, st) {
         if (e is BadRequest) {
           return Response.badRequest(body: e.message);
@@ -164,8 +168,11 @@ Middleware exceptionResponse() {
 @visibleForTesting
 class TestServerRunner {
   static const _port = 8080;
+
   late final DartServicesClient client;
-  final sdk = Sdk.fromLocalFlutter();
+  late final WebsocketServicesClient websocketClient;
+
+  final Sdk sdk = Sdk.fromLocalFlutter();
 
   Completer<void>? _started;
 
@@ -184,10 +191,15 @@ class TestServerRunner {
     } on SocketException {
       // This is expected if the server is already running.
     }
-    client = DartServicesClient(
-      DartServicesHttpClient(),
-      rootUrl: 'http://$localhostIp:$_port/',
-    );
+
+    final rootUrl = 'http://$localhostIp:$_port/';
+
+    // connect the regular client
+    client = DartServicesClient(DartServicesHttpClient(), rootUrl: rootUrl);
+
+    // connect the websocket client
+    websocketClient = await WebsocketServicesClient.connect(rootUrl);
+
     _started!.complete();
     return client;
   }

pkgs/dart_services/lib/src/common_server.dart

@@ -7,11 +7,14 @@ import 'dart:convert';
 import 'dart:io';
 
 import 'package:dartpad_shared/model.dart' as api;
+import 'package:dartpad_shared/ws.dart';
 import 'package:http/http.dart' as http;
 import 'package:logging/logging.dart';
 import 'package:shelf/shelf.dart';
 import 'package:shelf_router/shelf_router.dart';
 import 'package:shelf_static/shelf_static.dart';
+import 'package:shelf_web_socket/shelf_web_socket.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
 
 import 'analysis.dart';
 import 'caching.dart';
@@ -37,9 +40,9 @@ class CommonServerImpl {
   final Sdk sdk;
   final ServerCache cache;
 
-  late Analyzer analyzer;
-  late Compiler compiler;
-  final ai = GenerativeAI();
+  late final Analyzer analyzer;
+  late final Compiler compiler;
+  final GenerativeAI ai = GenerativeAI();
 
   CommonServerImpl(this.sdk, this.cache);
 
@@ -73,6 +76,9 @@ class CommonServerApi {
     // general requests (GET)
     router.get(r'/api/<apiVersion>/version', handleVersion);
 
+    // websocket requests
+    router.get(r'/ws', webSocketHandler(handleWebSocket));
+
     // serve the compiled artifacts
     final artifactsDir = Directory('artifacts');
     if (artifactsDir.existsSync()) {
@@ -115,6 +121,56 @@ class CommonServerApi {
     return ok(version().toJson());
   }
 
+  /// Handle a new websocket connection request.
+  ///
+  /// Handle incoming requests, convert them to exising commands and dispatch
+  /// them appropriately. The commands and responses mirror the existing REST
+  /// protocol.
+  ///
+  /// This will be a long-running conneciton to the client.
+  void handleWebSocket(WebSocketChannel webSocket, String? subprotocol) {
+    StreamSubscription<dynamic>? subscription;
+
+    subscription = webSocket.stream.listen(
+      (message) {
+        try {
+          // Handle incoming WebSocket messages
+          final request = JsonRpcRequest.fromJson(message as String);
+          log.genericInfo('ws request: ${request.method}');
+          JsonRpcResponse? response;
+
+          switch (request.method) {
+            case 'version':
+              final v = version();
+              response = request.createResultResponse(v.toJson());
+              break;
+            default:
+              response = request.createErrorResponse(
+                'unknown command: ${request.method}',
+              );
+              break;
+          }
+
+          webSocket.sink.add(jsonEncode(response.toJson()));
+          log.genericInfo(
+            'ws response: '
+            '${request.method} ${response.error != null ? '500' : '200'}',
+          );
+        } catch (e) {
+          log.genericSevere('error handling websocket request', error: e);
+        }
+      },
+      onDone: () {
+        // Clean up any stream subscription.
+        subscription?.cancel();
+        subscription = null;
+      },
+      onError: (Object error) {
+        log.genericSevere('error from websocket connection', error: error);
+      },
+    );
+  }
+
   Future<Response> handleAnalyze(Request request, String apiVersion) async {
     if (apiVersion != api3) return unhandledVersion(apiVersion);
 
@@ -512,7 +568,6 @@ Middleware logRequestsToLogger(DartPadLogger log) {
       final watch = Stopwatch()..start();
 
       final ctx = DartPadRequestContext.fromRequest(request);
-      log.genericInfo('received request, enableLogging=${ctx.enableLogging}');
 
       return Future.sync(() => innerHandler(request)).then(
         (response) {
@@ -524,7 +579,11 @@ Middleware logRequestsToLogger(DartPadLogger log) {
           return response;
         },
         onError: (Object error, StackTrace stackTrace) {
-          if (error is HijackException) throw error;
+          if (error is HijackException) {
+            log.info(_formatMessage(request, watch.elapsed), ctx);
+
+            throw error;
+          }
 
           log.info(_formatMessage(request, watch.elapsed, error: error), ctx);
 

pkgs/dart_services/pubspec.yaml

@@ -26,6 +26,8 @@ dependencies:
   shelf_gzip: ^4.1.0
   shelf_router: ^1.1.4
   shelf_static: ^1.1.0
+  shelf_web_socket: ^3.0.0
+  web_socket_channel: ^3.0.0
   yaml: ^3.1.3
 
 dev_dependencies:

pkgs/dart_services/test/presubmit/server_test.dart

@@ -11,13 +11,14 @@ import '../probes_and_presubmit/server_testing.dart';
 void main() async {
   final runner = TestServerRunner();
   await runner.maybeStart();
-  final client = runner.client;
 
   group('server', () {
-    testServer(client);
+    testServer(runner.client);
 
     if (runner.sdk.dartMajorVersion >= 3 && runner.sdk.dartMinorVersion >= 8) {
-      testReload(client);
+      testReload(runner.client);
     }
+
+    testServerWebsocket(runner.websocketClient);
   });
 }

pkgs/dart_services/test/probes_and_presubmit/server_testing.dart

@@ -9,7 +9,7 @@ import 'package:test/test.dart';
 import 'ddc_testing.dart';
 
 void testServer(DartServicesClient client, {int? retry}) {
-  group('server at ${client.rootUrl}', () {
+  group('DartServicesClient', () {
     testDDCEndpoint(
       'compileDDC',
       (request) => client.compileDDC(request),
@@ -352,3 +352,15 @@ void main() => print('hello world');
     });
   }, retry: retry);
 }
+
+void testServerWebsocket(WebsocketServicesClient client, {int? retry}) {
+  group('WebsocketServicesClient', () {
+    test('version', () async {
+      final result = await client.version();
+      expect(result.dartVersion, startsWith('3.'));
+      expect(result.flutterVersion, startsWith('3.'));
+      expect(result.engineVersion, isNotEmpty);
+      expect(result.packages, isNotEmpty);
+    });
+  }, retry: retry);
+}

pkgs/dart_services/tool/grind.dart

@@ -33,7 +33,7 @@ final List<String> compilationArtifactsNew = [
 ///
 /// * the Dart project template
 /// * the Flutter project template
-@Task('build the project templates')
+@Task('Build the project templates')
 void buildProjectTemplates() async {
   final templatesPath = path.join(Directory.current.path, 'project_templates');
   final templatesDirectory = Directory(templatesPath);
@@ -54,7 +54,7 @@ void buildProjectTemplates() async {
   await projectCreator.buildFlutterProjectTemplate();
 }
 
-@Task('build the sdk compilation artifacts for upload to google storage')
+@DefaultTask('Build the sdk compilation artifacts')
 void buildStorageArtifacts() async {
   final sdk = Sdk.fromLocalFlutter();
   delete(getDir('artifacts'));

pkgs/dartpad_shared/lib/backend_client.dart

@@ -9,7 +9,6 @@ import 'package:http/http.dart';
 import 'headers.dart';
 
 class DartServicesHttpClient {
-  final Client _client = Client();
   static Map<String, String> _headers = DartPadRequestHeaders(
     enableLogging: true,
   ).encoded;
@@ -19,7 +18,7 @@ class DartServicesHttpClient {
     _headers = DartPadRequestHeaders(enableLogging: false).encoded;
   }
 
-  void close() => _client.close();
+  final Client _client = Client();
 
   Future<Response> get(String url) async {
     return await _client.get(Uri.parse(url), headers: _headers);
@@ -46,4 +45,6 @@ class DartServicesHttpClient {
 
     return await _client.send(httpRequest);
   }
+
+  void close() => _client.close();
 }

pkgs/dartpad_shared/lib/services.dart

@@ -2,10 +2,14 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'dart:async';
 import 'dart:convert';
 
+import 'package:web_socket/web_socket.dart';
+
 import 'backend_client.dart';
 import 'model.dart';
+import 'ws.dart';
 
 export 'model.dart';
 
@@ -131,6 +135,81 @@ class DartServicesClient {
   }
 }
 
+/// A websocket analog to [DartServicesClient].
+class WebsocketServicesClient {
+  final Uri wsUrl;
+  final WebSocket socket;
+  final IDFactory idFactory = IDFactory();
+
+  final Map<int, Completer<Object>> responseCompleters = {};
+  final Map<int, Object Function(Map<String, Object?>)> responseDecoders = {};
+
+  final Completer<void> _closedCompleter = Completer();
+
+  WebsocketServicesClient._(this.wsUrl, this.socket);
+
+  static Future<WebsocketServicesClient> connect(String rootUrl) async {
+    final url = Uri.parse(rootUrl);
+    final wsUrl = url.replace(
+      scheme: url.scheme == 'https' ? 'wss' : 'ws',
+      path: 'ws',
+    );
+    final socket = await WebSocket.connect(wsUrl);
+    final client = WebsocketServicesClient._(wsUrl, socket);
+    client._init();
+    return client;
+  }
+
+  void _init() {
+    socket.events.listen((e) async {
+      switch (e) {
+        case TextDataReceived(text: final text):
+          _dispatch(JsonRpcResponse.fromJson(text));
+          break;
+        case BinaryDataReceived(data: final _):
+          // Ignore - binary data is unsupported.
+          break;
+        case CloseReceived(code: final _, reason: final _):
+          // Notify that the server connection has closed.
+          _closedCompleter.complete();
+          break;
+      }
+    });
+  }
+
+  Future<void> get onClosed => _closedCompleter.future;
+
+  Future<VersionResponse> version() {
+    final requestId = idFactory.generateNextId();
+    final completer = Completer<VersionResponse>();
+
+    responseCompleters[requestId] = completer;
+    responseDecoders[requestId] = VersionResponse.fromJson;
+
+    socket.sendText(
+      jsonEncode(JsonRpcRequest(method: 'version', id: requestId).toJson()),
+    );
+
+    return completer.future;
+  }
+
+  Future<void> dispose() => socket.close();
+
+  void _dispatch(JsonRpcResponse response) {
+    final id = response.id;
+
+    final completer = responseCompleters[id]!;
+    final decoder = responseDecoders[id]!;
+
+    if (response.error != null) {
+      completer.completeError(response.error!);
+    } else {
+      final result = decoder((response.result! as Map).cast());
+      completer.complete(result);
+    }
+  }
+}
+
 class ApiRequestError implements Exception {
   ApiRequestError(this.message, this.body);