[ package:dds ] Add server-sent event (SSE) support to DDS

This support is required for web clients of dwds within google3

Change-Id: Ia1ecbf8f5ba79d53cb340c83a579dc0810ec0065
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/150183
Reviewed-by: Gary Roumanis <grouma@google.com>

Author: bkonyi

.dart_tool/package_config.json

@@ -556,6 +556,12 @@
       "packageUri": "lib/",
       "languageVersion": "2.1"
     },
+    {
+      "name": "sse",
+      "rootUri": "../third_party/pkg/sse",
+      "packageUri": "lib/",
+      "languageVersion": "2.6"
+    },
     {
       "name": "stack_trace",
       "rootUri": "../third_party/pkg/stack_trace",
@@ -586,6 +592,12 @@
       "packageUri": "lib/",
       "languageVersion": "2.0"
     },
+    {
+      "name": "sync_http",
+      "rootUri": "../third_party/pkg/sync_http",
+      "packageUri": "lib/",
+      "languageVersion": "2.6"
+    },
     {
       "name": "telemetry",
       "rootUri": "../pkg/telemetry",
@@ -694,6 +706,12 @@
       "packageUri": "lib/",
       "languageVersion": "2.2"
     },
+    {
+      "name": "webdriver",
+      "rootUri": "../third_party/pkg/webdriver",
+      "packageUri": "lib/",
+      "languageVersion": "2.6"
+    },
     {
       "name": "web_components",
       "rootUri": "../third_party/pkg/web_components",
@@ -713,4 +731,4 @@
       "languageVersion": "2.4"
     }
   ]
-}
+}

.packages

@@ -92,11 +92,13 @@ source_map_stack_trace:third_party/pkg/source_map_stack_trace/lib
 sourcemap_testing:pkg/sourcemap_testing/lib
 source_maps:third_party/pkg/source_maps/lib
 source_span:third_party/pkg/source_span/lib
+sse:third_party/pkg/sse/lib
 stack_trace:third_party/pkg/stack_trace/lib
 stagehand:third_party/pkg/stagehand/lib
 status_file:pkg/status_file/lib
 stream_channel:third_party/pkg/stream_channel/lib
 string_scanner:third_party/pkg/string_scanner/lib
+sync_http:third_party/pkg/sync_http/lib
 telemetry:pkg/telemetry/lib
 term_glyph:third_party/pkg/term_glyph/lib
 test:third_party/pkg/test/pkgs/test/lib
@@ -115,6 +117,7 @@ vm:pkg/vm/lib
 vm_service:pkg/vm_service/lib
 vm_snapshot_analysis:pkg/vm_snapshot_analysis/lib
 watcher:third_party/pkg/watcher/lib
+webdriver:third_party/pkg/webdriver/lib
 web_components:third_party/pkg/web_components/lib
 web_socket_channel:third_party/pkg/web_socket_channel/lib
 yaml:third_party/pkg/yaml/lib

DEPS

@@ -96,6 +96,7 @@ vars = {
   # For more details, see https://github.com/dart-lang/sdk/issues/30164
   "dart_style_tag": "1.3.6",  # Please see the note above before updating.
 
+  "chromedriver_tag": "83.0.4103.39",
   "dartdoc_tag" : "v0.32.2",
   "ffi_rev": "454ab0f9ea6bd06942a983238d8a6818b1357edb",
   "fixnum_rev": "9b38f49f6679654d66a363e69e48173cca07e882",
@@ -142,10 +143,12 @@ vars = {
   "source_maps-0.9.4_rev": "38524",
   "source_maps_rev": "87b4fd9027378bbd51b02e9d7df794eee8a82b7a",
   "source_span_tag": "1.7.0",
+  "sse_tag": "e5cf68975e8e87171a3dc297577aa073454a91dc",
   "stack_trace_tag": "56811dbb2530d823b764fe167ec335879a4adb32",
   "stagehand_tag": "v3.3.9",
   "stream_channel_tag": "70433d577be02c48cb16d72d65654f3b4d82c6ed",
   "string_scanner_rev": "a918e7371af6b6e73bfd534ff9da6084741c1f99",
+  "sync_http_rev": "a85d7ec764ea485cbbc49f3f3e7f1b43f87a1c74",
   "test_descriptor_tag": "1.1.1",
   "test_process_tag": "1.0.3",
   "term_glyph_rev": "b3da31e9684a99cfe5f192b89914492018b44da7",
@@ -156,6 +159,7 @@ vars = {
   "usage_tag": "3.4.0",
   "vector_math_rev": "90631fbb609f61d42f28621253c0ec9fc6a326d2",
   "watcher_rev": "fc3c9aae5d31d707b3013b42634dde8d8a1161b4",
+  "webdriver_rev": "5a8d6805d9cf8a3cbb4fcd64849b538b7491e50e",
   "web_components_rev": "8f57dac273412a7172c8ade6f361b407e2e4ed02",
   "web_socket_channel_rev": "490061ef0e22d3c8460ad2802f9948219365ad6b",
   "WebCore_rev": "fb11e887f77919450e497344da570d780e078bc8",
@@ -408,6 +412,8 @@ deps = {
   Var("dart_root") + "/third_party/pkg/source_map_stack_trace":
       Var("dart_git") + "source_map_stack_trace.git" +
       "@" + Var("source_map_stack_trace_tag"),
+  Var("dart_root") + "/third_party/pkg/sse":
+      Var("dart_git") + "sse.git" + "@" + Var("sse_tag"),
   Var("dart_root") + "/third_party/pkg/stack_trace":
       Var("dart_git") + "stack_trace.git" + "@" + Var("stack_trace_tag"),
   Var("dart_root") + "/third_party/pkg/stagehand":
@@ -418,6 +424,8 @@ deps = {
   Var("dart_root") + "/third_party/pkg/string_scanner":
       Var("dart_git") + "string_scanner.git" +
       "@" + Var("string_scanner_rev"),
+  Var("dart_root") + "/third_party/pkg/sync_http":
+      Var("dart_git") + "sync_http.git" + "@" + Var("sync_http_rev"),
   Var("dart_root") + "/third_party/pkg/term_glyph":
       Var("dart_git") + "term_glyph.git" + "@" + Var("term_glyph_rev"),
   Var("dart_root") + "/third_party/pkg/test":
@@ -443,6 +451,10 @@ deps = {
   Var("dart_root") + "/third_party/pkg/web_components":
       Var("dart_git") + "web-components.git" +
       "@" + Var("web_components_rev"),
+  Var("dart_root") + "/third_party/pkg/webdriver":
+      Var("dart_git") + "external/github.com/google/webdriver.dart.git" +
+      "@" + Var("webdriver_rev"),
+
   Var("dart_root") + "/third_party/pkg/web_socket_channel":
       Var("dart_git") + "web_socket_channel.git" +
       "@" + Var("web_socket_channel_rev"),
@@ -460,6 +472,17 @@ deps = {
       "dep_type": "cipd",
   },
 
+  Var("dart_root") + "/third_party/webdriver/chrome": {
+    "packages": [
+      {
+        "package": "dart/third_party/chromedriver/${{platform}}",
+        "version": "version:" + Var("chromedriver_tag"),
+      }
+    ],
+    "condition": "host_cpu == 'x64'",
+    "dep_type": "cipd",
+  },
+
   Var("dart_root") + "/pkg/analysis_server/language_model": {
     "packages": [
       {

pkg/dds/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 1.3.0
+
+- Added support for SSE connections from web-based clients.
+
 # 1.2.4
 
 - Fixed another issue where a `StateError` could be raised within `DartDevelopmentService`

pkg/dds/README.md

@@ -2,7 +2,35 @@ A package used to spawn the Dart Developer Service (DDS), which is used to commu
 
 # Functionality
 
-Existing VM Service clients can issue both HTTP and websocket requests to a running DDS instance as if it were an instance of the VM Service itself. If a request corresponds to an RPC defined in the [VM Service Protocol][service-protocol], DDS will forward the request and return the response from the VM Service. Requests corresponding to an RPC defined in the [DDS Protocol][dds-protocol] will be handled directly by the DDS instance.
+Existing VM Service clients can issue both HTTP, websocket, and SSE requests to a running DDS instance as if it were an instance of the VM Service itself. If a request corresponds to an RPC defined in the [VM Service Protocol][service-protocol], DDS will forward the request and return the response from the VM Service. Requests corresponding to an RPC defined in the [DDS Protocol][dds-protocol] will be handled directly by the DDS instance.
+
+# SSE Support
+
+For certain web clients it may be preferrable or required to communicate with DDS using server-sent events (SSE). DDS has a SSE handler listening for requests on `/$debugHandler`.
+
+## SSE and package:vm_service example
+
+```dart
+import 'package:sse/sse.dart';
+import 'package:vm_service/vm_service.dart';
+
+void main() {
+  // Establish connection with DDS using SSE.
+  final ddsChannel = SseClient('${ddsUri}\$debugHandler');
+
+  // Wait for ddsChannel to be established
+  await ddsChannel.onOpen.first;
+
+  // Initialize VmService using the sink and stream from ddsChannel.
+  final vmService = VmService(
+    ddsChannel.stream,
+    (e) => ddsChannel.sink.add(e),
+  );
+
+  // You're ready to query DDS and the VM service!
+  print(await vmService.getVersion());
+}
+```
 
 [dds-protocol]: dds_protocol.md
 [service-protocol]: https://github.com/dart-lang/sdk/blob/master/runtime/vm/service/service.md

pkg/dds/lib/dds.dart

@@ -21,6 +21,7 @@ import 'package:shelf/shelf.dart';
 import 'package:shelf/shelf_io.dart' as io;
 import 'package:shelf_proxy/shelf_proxy.dart';
 import 'package:shelf_web_socket/shelf_web_socket.dart';
+import 'package:sse/server/sse_handler.dart';
 import 'package:stream_channel/stream_channel.dart';
 import 'package:web_socket_channel/web_socket_channel.dart';
 
@@ -110,6 +111,12 @@ abstract class DartDevelopmentService {
   /// Returns `null` if the service is not running.
   Uri get uri;
 
+  /// The [Uri] VM service clients can use to communicate with this
+  /// [DartDevelopmentService] via server-sent events (SSE).
+  ///
+  /// Returns `null` if the service is not running.
+  Uri get sseUri;
+
   /// The [Uri] VM service clients can use to communicate with this
   /// [DartDevelopmentService] via a [WebSocket].
   ///

pkg/dds/lib/src/client.dart

@@ -7,24 +7,46 @@ part of dds;
 /// Representation of a single DDS client which manages the connection and
 /// DDS request intercepting / forwarding.
 class _DartDevelopmentServiceClient {
-  _DartDevelopmentServiceClient(
+  factory _DartDevelopmentServiceClient.fromWebSocket(
+    DartDevelopmentService dds,
+    WebSocketChannel ws,
+    json_rpc.Peer vmServicePeer,
+  ) =>
+      _DartDevelopmentServiceClient._(
+        dds,
+        ws,
+        vmServicePeer,
+      );
+
+  factory _DartDevelopmentServiceClient.fromSSEConnection(
+    DartDevelopmentService dds,
+    SseConnection sse,
+    json_rpc.Peer vmServicePeer,
+  ) =>
+      _DartDevelopmentServiceClient._(
+        dds,
+        sse,
+        vmServicePeer,
+      );
+
+  _DartDevelopmentServiceClient._(
     this.dds,
-    this.ws,
+    this.connection,
     json_rpc.Peer vmServicePeer,
   ) : _vmServicePeer = vmServicePeer {
     _clientPeer = json_rpc.Peer(
       // Manually create a StreamChannel<String> instead of calling
-      // ws.cast<String>() as cast() results in addStream() being called,
+      // .cast<String>() as cast() results in addStream() being called,
       // binding the underlying sink. This results in a StateError being thrown
       // if we try and add directly to the sink, which we do for binary events
       // in _StreamManager's streamNotify().
       StreamChannel<String>(
-        ws.stream.cast(),
+        connection.stream.cast(),
         StreamController(sync: true)
           ..stream
               .cast()
-              .listen((event) => ws.sink.add(event))
-              .onDone(() => ws.sink.close()),
+              .listen((event) => connection.sink.add(event))
+              .onDone(() => connection.sink.close()),
       ),
       strictProtocolChecks: false,
     );
@@ -231,8 +253,8 @@ class _DartDevelopmentServiceClient {
   String _name;
 
   final _DartDevelopmentService dds;
+  final StreamChannel connection;
   final Map<String, String> services = {};
   final json_rpc.Peer _vmServicePeer;
-  final WebSocketChannel ws;
   json_rpc.Peer _clientPeer;
 }

pkg/dds/lib/src/dds_impl.dart

@@ -139,28 +139,47 @@ class _DartDevelopmentService implements DartDevelopmentService {
   // Attempt to upgrade HTTP requests to a websocket before processing them as
   // standard HTTP requests. The websocket handler will fail quickly if the
   // request doesn't appear to be a websocket upgrade request.
-  Cascade _handlers() => Cascade().add(_webSocketHandler()).add(_httpHandler());
+  Cascade _handlers() {
+    return Cascade()
+        .add(_webSocketHandler())
+        .add(_sseHandler())
+        .add(_httpHandler());
+  }
 
   Handler _webSocketHandler() => webSocketHandler((WebSocketChannel ws) {
-        final client = _DartDevelopmentServiceClient(
+        final client = _DartDevelopmentServiceClient.fromWebSocket(
           this,
           ws,
           _vmServiceClient,
         );
         clientManager.addClient(client);
       });
 
+  Handler _sseHandler() {
+    final handler = authCodesEnabled
+        ? SseHandler(Uri.parse('/$_authCode/$_kSseHandlerPath'))
+        : SseHandler(Uri.parse('/$_kSseHandlerPath'));
+
+    handler.connections.rest.listen((sseConnection) {
+      final client = _DartDevelopmentServiceClient.fromSSEConnection(
+        this,
+        sseConnection,
+        _vmServiceClient,
+      );
+      clientManager.addClient(client);
+    });
+
+    return handler.handler;
+  }
+
   Handler _httpHandler() {
     // DDS doesn't support any HTTP requests itself, so we just forward all of
     // them to the VM service.
     final cascade = Cascade().add(proxyHandler(remoteVmServiceUri));
     return cascade.handler;
   }
 
-  Uri _toWebSocket(Uri uri) {
-    if (uri == null) {
-      return null;
-    }
+  List<String> _cleanupPathSegments(Uri uri) {
     final pathSegments = <String>[];
     if (uri.pathSegments.isNotEmpty) {
       pathSegments.addAll(uri.pathSegments.where(
@@ -170,10 +189,27 @@ class _DartDevelopmentService implements DartDevelopmentService {
         (s) => s.isNotEmpty,
       ));
     }
+    return pathSegments;
+  }
+
+  Uri _toWebSocket(Uri uri) {
+    if (uri == null) {
+      return null;
+    }
+    final pathSegments = _cleanupPathSegments(uri);
     pathSegments.add('ws');
     return uri.replace(scheme: 'ws', pathSegments: pathSegments);
   }
 
+  Uri _toSse(Uri uri) {
+    if (uri == null) {
+      return null;
+    }
+    final pathSegments = _cleanupPathSegments(uri);
+    pathSegments.add(_kSseHandlerPath);
+    return uri.replace(pathSegments: pathSegments);
+  }
+
   String _getNamespace(_DartDevelopmentServiceClient client) =>
       clientManager.clients.keyOf(client);
 
@@ -186,6 +222,7 @@ class _DartDevelopmentService implements DartDevelopmentService {
   Uri _remoteVmServiceUri;
 
   Uri get uri => _uri;
+  Uri get sseUri => _toSse(_uri);
   Uri get wsUri => _toWebSocket(_uri);
   Uri _uri;
 
@@ -210,6 +247,8 @@ class _DartDevelopmentService implements DartDevelopmentService {
   _StreamManager get streamManager => _streamManager;
   _StreamManager _streamManager;
 
+  static const _kSseHandlerPath = '\$debugHandler';
+
   json_rpc.Peer _vmServiceClient;
   WebSocketChannel _vmServiceSocket;
   HttpServer _server;

pkg/dds/lib/src/stream_manager.dart

@@ -28,7 +28,7 @@ class _StreamManager {
           continue;
         }
         if (isBinaryData) {
-          listener.ws.sink.add(data);
+          listener.connection.sink.add(data);
         } else {
           listener.sendNotification('streamNotify', data);
         }