diff --git a/pkg/dart_runtime_service/lib/src/clients.dart b/pkg/dart_runtime_service/lib/src/clients.dart index a055a880446..438907cc6ff 100644 --- a/pkg/dart_runtime_service/lib/src/clients.dart +++ b/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 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 instead of calling + // .cast() 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(sync: true) + ..stream + .cast() + .listen((event) => connection.sink.add(event)) + .onDone(() => connection.sink.close()); + final manualConnectionStreamCast = connection.stream.cast(); + _clientPeer = json_rpc.Peer( + StreamChannel( + manualConnectionStreamCast, + manualConnectionSinkCast, + ), + strictProtocolChecks: false, + ); _internalRpcs = DartRuntimeServiceRpcs( clients: clients, eventStreamMethods: eventStreamMethods, @@ -40,6 +60,7 @@ base class Client { late final String namespace; + final StreamChannel 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 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 connection, + required StreamChannel connection, String? name, bool artificial = false, }) { diff --git a/pkg/dart_runtime_service/lib/src/dart_runtime_service.dart b/pkg/dart_runtime_service/lib/src/dart_runtime_service.dart index 1ba64ddac83..e09f01a04be 100644 --- a/pkg/dart_runtime_service/lib/src/dart_runtime_service.dart +++ b/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); } diff --git a/pkg/dart_runtime_service/lib/src/event_streams.dart b/pkg/dart_runtime_service/lib/src/event_streams.dart index e15bfe609df..4cd565740d2 100644 --- a/pkg/dart_runtime_service/lib/src/event_streams.dart +++ b/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 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, diff --git a/pkg/dart_runtime_service/lib/src/handlers.dart b/pkg/dart_runtime_service/lib/src/handlers.dart index eb0b32f00bd..5dbaa99d2ef 100644 --- a/pkg/dart_runtime_service/lib/src/handlers.dart +++ b/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()); + clientManager.addClient(connection: ws.cast()); }); return (request) { diff --git a/pkg/dart_runtime_service_vm/lib/dart_runtime_service_vm.dart b/pkg/dart_runtime_service_vm/lib/dart_runtime_service_vm.dart index f066bc27a62..2df2d05bf0c 100644 --- a/pkg/dart_runtime_service_vm/lib/dart_runtime_service_vm.dart +++ b/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 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, - ); + _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 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, + ), + [final Uint8List utf8String] => ForwardingStreamEvent( + streamId: streamId, + event: json.decode(utf8.decode(utf8String)) as Map, + ), + final Uint8List binaryData => BinaryStreamEvent( + streamId: streamId, + data: binaryData, + ), + _ => throw UnimplementedError( + 'Unexpected event type: ${event.runtimeType}.', + ), + }, ); } @@ -316,23 +330,16 @@ class DartRuntimeServiceVMBackend List 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 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;