[ Service ] Add support for binary events in package:dart_runtime_service_vm

package:vm_service test suite is ~98% passing with this change.
Change-Id: I7455eb6414ac6a9f28408613543d18be7940bbb4
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/490600
Reviewed-by: Jessy Yameogo <yjessy@google.com>

pkg/dart_runtime_service/lib/src/clients.dart

@@ -2,6 +2,9 @@
 // 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:typed_data';
+
 import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
 import 'package:logging/logging.dart';
 import 'package:meta/meta.dart';
@@ -21,7 +24,7 @@ typedef ServiceNameAliasPair = ({ServiceName service, ServiceAlias alias});
 /// Represents a client that is connected to a service.
 base class Client {
   Client({
-    required StreamChannel<String> connection,
+    required this.connection,
     required UnmodifiableClientNamedLookup clients,
     required EventStreamMethods eventStreamMethods,
     required this.backend,
@@ -29,7 +32,24 @@ base class Client {
     String? name,
   }) {
     _name = name ?? defaultClientName;
-    _clientPeer = json_rpc.Peer(connection, strictProtocolChecks: false);
+    // Manually create a StreamChannel<String> instead of calling
+    // .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 [EventStreamMethod]'s streamNotify().
+    final manualConnectionSinkCast = StreamController<String>(sync: true)
+      ..stream
+          .cast<String>()
+          .listen((event) => connection.sink.add(event))
+          .onDone(() => connection.sink.close());
+    final manualConnectionStreamCast = connection.stream.cast<String>();
+    _clientPeer = json_rpc.Peer(
+      StreamChannel<String>(
+        manualConnectionStreamCast,
+        manualConnectionSinkCast,
+      ),
+      strictProtocolChecks: false,
+    );
     _internalRpcs = DartRuntimeServiceRpcs(
       clients: clients,
       eventStreamMethods: eventStreamMethods,
@@ -40,6 +60,7 @@ base class Client {
 
   late final String namespace;
 
+  final StreamChannel<Object?> connection;
   late json_rpc.Peer _clientPeer;
   late final DartRuntimeServiceRpcs _internalRpcs;
   final DartRuntimeServiceBackend backend;
@@ -160,6 +181,23 @@ base class Client {
     }
   }
 
+  /// Sends raw binary [data] to the client.
+  ///
+  /// This technically isn't compliant with the JSON-RPC specification and
+  /// should only be used to send binary events to streams.
+  void sendBinaryData({required Uint8List data}) {
+    if (_clientPeer.isClosed) {
+      RpcException.serviceDisappeared.throwException();
+    }
+
+    try {
+      connection.sink.add(data);
+      // ignore: avoid_catching_errors
+    } on StateError {
+      RpcException.serviceDisappeared.throwException();
+    }
+  }
+
   /// The set of services registered by this [Client].
   Iterable<ServiceNameAliasPair> get services =>
       _services.entries.map((e) => (service: e.key, alias: e.value));
@@ -257,7 +295,7 @@ base class ClientManager implements ClientConnectionController {
   /// This should be called when a client connects to the service.
   @mustCallSuper
   Client addClient({
-    required StreamChannel<String> connection,
+    required StreamChannel<Object?> connection,
     String? name,
     bool artificial = false,
   }) {

pkg/dart_runtime_service/lib/src/dart_runtime_service.dart

@@ -262,7 +262,7 @@ class DartRuntimeService {
   }
 
   /// Send a [StreamEvent] to subscribed clients.
-  void sendEvent({required StreamEvent event}) {
+  void sendEvent({required StreamEventBase event}) {
     event.send(eventStreamMethods: eventStreamManager);
   }
 

pkg/dart_runtime_service/lib/src/event_streams.dart

@@ -14,16 +14,11 @@ import 'dart_runtime_service_backend.dart';
 import 'rpc_exceptions.dart';
 import 'utils.dart';
 
-/// A base class for events to be sent on [streamId] with a given [kind].
-abstract base class StreamEvent {
-  StreamEvent({required this.streamId, required this.kind});
-
-  static const kStreamId = 'streamId';
-  static const kEvent = 'event';
+/// A base class for events to be sent on [streamId].
+abstract base class StreamEventBase {
+  const StreamEventBase({required this.streamId});
 
   final String streamId;
-  final String kind;
-  final int timestamp = DateTime.now().millisecondsSinceEpoch;
 
   void send({
     required EventStreamMethods eventStreamMethods,
@@ -35,11 +30,30 @@ abstract base class StreamEvent {
       excludedClient: excludedClient,
     );
   }
+}
+
+/// A base class for JSON-RPC compliant events to be sent on [streamId] with a
+/// given [kind].
+abstract base class StreamEvent extends StreamEventBase {
+  StreamEvent({required super.streamId, required this.kind});
+
+  static const kStreamId = 'streamId';
+  static const kEvent = 'event';
+
+  final String kind;
+  final int timestamp = DateTime.now().millisecondsSinceEpoch;
 
   @mustCallSuper
   Map<String, Object?> toJson();
 }
 
+/// A class for sending non-JSON-RPC compliant binary events on [streamId].
+final class BinaryStreamEvent extends StreamEventBase {
+  const BinaryStreamEvent({required super.streamId, required this.data});
+
+  final Uint8List data;
+}
+
 /// Base class for service registration events which are sent on the Service
 /// stream.
 abstract base class ServiceRegistrationEvent extends StreamEvent {
@@ -187,11 +201,8 @@ class EventStreamManager implements EventStreamMethods {
           continue;
         }
         switch (data) {
-          case Uint8List():
-            // TODO(bkonyi): support sending binary events (e.g., for heap
-            // snapshots).
-            // listener.connection.sink.add(data);
-            throw StateError('Cannot send binary data');
+          case BinaryStreamEvent(data: final binaryData):
+            listener.sendBinaryData(data: binaryData);
           case StreamEvent():
             listener.sendNotification(
               method: kStreamNotify,

pkg/dart_runtime_service/lib/src/handlers.dart

@@ -171,7 +171,7 @@ Handler webSocketClientHandler({required ClientManager clientManager}) {
   // package:shelf_web_socket v2.
   final handler = webSocketHandler((WebSocketChannel ws, _) {
     logger.info('New web socket connection. Creating $Client.');
-    clientManager.addClient(connection: ws.cast<String>());
+    clientManager.addClient(connection: ws.cast<Object?>());
   });
 
   return (request) {

pkg/dart_runtime_service_vm/lib/dart_runtime_service_vm.dart

@@ -7,6 +7,7 @@ import 'dart:collection';
 import 'dart:convert';
 import 'dart:io';
 import 'dart:isolate';
+import 'dart:typed_data';
 
 import 'package:dart_runtime_service/dart_runtime_service.dart';
 
@@ -213,12 +214,9 @@ class DartRuntimeServiceVMBackend
   void _vmMessageHandler(List<Object?> message) {
     _logger.fine('VM message: $message');
     switch (message) {
-      case [final String streamId, final String eventJsonString]:
+      case [final String streamId, final Object event]:
         // This is an event.
-        _eventMessageHandler(
-          streamId,
-          json.decode(eventJsonString) as Map<String, Object?>,
-        );
+        _eventMessageHandler(streamId, event);
       case [final int opcode]:
         // This is a control message directing the vm service to exit.
         assert(opcode == _kServiceExitMessageId);
@@ -257,9 +255,25 @@ class DartRuntimeServiceVMBackend
   }
 
   /// Forward VM service events sent from the VM.
-  void _eventMessageHandler(String streamId, Map<String, Object?> event) {
+  void _eventMessageHandler(String streamId, Object event) {
     frontend.sendEvent(
-      event: ForwardingStreamEvent(streamId: streamId, event: event),
+      event: switch (event) {
+        final String jsonString => ForwardingStreamEvent(
+          streamId: streamId,
+          event: json.decode(jsonString) as Map<String, Object?>,
+        ),
+        [final Uint8List utf8String] => ForwardingStreamEvent(
+          streamId: streamId,
+          event: json.decode(utf8.decode(utf8String)) as Map<String, Object?>,
+        ),
+        final Uint8List binaryData => BinaryStreamEvent(
+          streamId: streamId,
+          data: binaryData,
+        ),
+        _ => throw UnimplementedError(
+          'Unexpected event type: ${event.runtimeType}.',
+        ),
+      },
     );
   }
 
@@ -316,23 +330,16 @@ class DartRuntimeServiceVMBackend
     List<int> message,
     SendPort replyPort,
   ) async {
+    // The original VM service implementation could, in theory, handle binary
+    // and "UTF8String" responses. However, no binary or "UTF8String" responses
+    // are sent in response to service RPCs so we should be able to assume that
+    // `message` can be converted to a `String`.
+    //
+    // If this decode throws an exception, we'll need to revisit this.
     final messageStr = utf8.decode(message);
     _logger.info('Native RPC request: $messageStr');
     _nativeRpcClientStreamChannelController.local.sink.add(messageStr);
 
-    // TODO(bkonyi): handle non-string results
-    /*
-    late List<int> bytes;
-    switch (response.kind) {
-      case ResponsePayloadKind.String:
-        bytes = utf8.encode(response.payload as String);
-        bytes = bytes is Uint8List ? bytes : Uint8List.fromList(bytes);
-      case ResponsePayloadKind.Binary:
-      case ResponsePayloadKind.Utf8String:
-        bytes = response.payload as Uint8List;
-    }
-    */
-
     if (!await _nativeRpcClientResponseStream.moveNext()) {
       _logger.warning('Native RPC client stream has closed.');
       return;