Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Reland "[ Service / package:dds ] Add stream support to package:dds a…
…nd enable DDS for VM service tests" This reverts commit cccddf3. Change-Id: Iabde3542d5be33ffabf50efd9226597aef876ab7 Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/143961 Reviewed-by: Alexander Aprelev <aam@google.com> Commit-Queue: Ben Konyi <bkonyi@google.com>
- Loading branch information
Showing
40 changed files
with
708 additions
and
133 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| // Copyright (c) 2020, the Dart project authors. Please see the AUTHORS file | ||
| // 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. | ||
|
|
||
| part of dds; | ||
|
|
||
| /// Adds support for binary events send from the VM service, which are not part | ||
| /// of the official JSON RPC 2.0 specification. | ||
| /// | ||
| /// A binary event from the VM service has the form: | ||
| /// ``` | ||
| /// type BinaryEvent { | ||
| /// dataOffset : uint32, | ||
| /// metadata : uint8[dataOffset-4], | ||
| /// data : uint8[], | ||
| /// } | ||
| /// ``` | ||
| /// where `metadata` is the JSON body of the event. | ||
| /// | ||
| /// [_BinaryCompatiblePeer] assumes that only stream events can contain a | ||
| /// binary payload (e.g., clients cannot send a `BinaryEvent` to the VM service). | ||
| class _BinaryCompatiblePeer extends json_rpc.Peer { | ||
| _BinaryCompatiblePeer(WebSocketChannel ws, _StreamManager streamManager) | ||
| : super( | ||
| ws.transform<String>( | ||
| StreamChannelTransformer( | ||
| StreamTransformer<dynamic, String>.fromHandlers( | ||
| handleData: (data, EventSink<String> sink) => | ||
| _transformStream(streamManager, data, sink)), | ||
| StreamSinkTransformer<String, dynamic>.fromHandlers( | ||
| handleData: (String data, EventSink<dynamic> sink) { | ||
| sink.add(data); | ||
| }, | ||
| ), | ||
| ), | ||
| ), | ||
| ); | ||
|
|
||
| static void _transformStream( | ||
| _StreamManager streamManager, dynamic data, EventSink<String> sink) { | ||
| if (data is String) { | ||
| // Non-binary messages come in as Strings. Simply forward to the sink. | ||
| sink.add(data); | ||
| } else if (data is Uint8List) { | ||
| // Only binary events will result in `data` being of type Uint8List. We | ||
| // need to manually forward them here. | ||
| final bytesView = | ||
| ByteData.view(data.buffer, data.offsetInBytes, data.lengthInBytes); | ||
| const metadataOffset = 4; | ||
| final dataOffset = bytesView.getUint32(0, Endian.little); | ||
| final metadataLength = dataOffset - metadataOffset; | ||
| final metadata = Utf8Decoder().convert(new Uint8List.view( | ||
| bytesView.buffer, | ||
| bytesView.offsetInBytes + metadataOffset, | ||
| metadataLength)); | ||
| final decodedMetadata = json.decode(metadata); | ||
| streamManager.streamNotify(decodedMetadata['params']['streamId'], data); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| // Copyright (c) 2020, the Dart project authors. Please see the AUTHORS file | ||
| // 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. | ||
|
|
||
| part of dds; | ||
|
|
||
| /// Representation of a single DDS client which manages the connection and | ||
| /// DDS request intercepting / forwarding. | ||
| class _DartDevelopmentServiceClient { | ||
| _DartDevelopmentServiceClient( | ||
| this.dds, | ||
| this.ws, | ||
| json_rpc.Peer vmServicePeer, | ||
| ) : _vmServicePeer = vmServicePeer { | ||
| _clientPeer = json_rpc.Peer(ws.cast<String>()); | ||
| _registerJsonRpcMethods(); | ||
| } | ||
|
|
||
| /// Start receiving JSON RPC requests from the client. | ||
| /// | ||
| /// Returned future completes when the peer is closed. | ||
| Future<void> listen() => _clientPeer.listen().then( | ||
| (_) => dds.streamManager.clientDisconnect(this), | ||
| ); | ||
|
|
||
| /// Close the connection to the client. | ||
| Future<void> close() async { | ||
| // Cleanup the JSON RPC server for this connection if DDS has shutdown. | ||
| await _clientPeer.close(); | ||
| } | ||
|
|
||
| /// Send a JSON RPC notification to the client. | ||
| void sendNotification(String method, [dynamic parameters]) async { | ||
| if (_clientPeer.isClosed) { | ||
| return; | ||
| } | ||
| _clientPeer.sendNotification(method, parameters); | ||
| } | ||
|
|
||
| /// Registers handlers for JSON RPC methods which need to be intercepted by | ||
| /// DDS as well as fallback request forwarder. | ||
| void _registerJsonRpcMethods() { | ||
| _clientPeer.registerMethod('streamListen', (parameters) async { | ||
| final streamId = parameters['streamId'].asString; | ||
| await dds.streamManager.streamListen(this, streamId); | ||
| return _success; | ||
| }); | ||
|
|
||
| _clientPeer.registerMethod('streamCancel', (parameters) async { | ||
| final streamId = parameters['streamId'].asString; | ||
| await dds.streamManager.streamCancel(this, streamId); | ||
| return _success; | ||
| }); | ||
|
|
||
| // Unless otherwise specified, the request is forwarded to the VM service. | ||
| _clientPeer.registerFallback((parameters) async => | ||
| await _vmServicePeer.sendRequest(parameters.method, parameters.asMap)); | ||
| } | ||
|
|
||
| static const _success = <String, dynamic>{ | ||
| 'type': 'Success', | ||
| }; | ||
|
|
||
| final _DartDevelopmentService dds; | ||
| final json_rpc.Peer _vmServicePeer; | ||
| final WebSocketChannel ws; | ||
| json_rpc.Peer _clientPeer; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,113 @@ | ||
| // Copyright (c) 2020, the Dart project authors. Please see the AUTHORS file | ||
| // 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. | ||
|
|
||
| part of dds; | ||
|
|
||
| class _StreamManager { | ||
| _StreamManager(this.dds); | ||
|
|
||
| /// Send `streamNotify` notifications to clients subscribed to `streamId`. | ||
| /// | ||
| /// If `data` is of type `Uint8List`, the notification is assumed to be a | ||
| /// binary event and is forwarded directly over the subscriber's websocket. | ||
| /// Otherwise, the event is sent via the JSON RPC client. | ||
| void streamNotify(String streamId, data) { | ||
| if (streamListeners.containsKey(streamId)) { | ||
| final listeners = streamListeners[streamId]; | ||
| final isBinaryData = data is Uint8List; | ||
| for (final listener in listeners) { | ||
| if (isBinaryData) { | ||
| listener.ws.sink.add(data); | ||
| } else { | ||
| listener.sendNotification('streamNotify', data); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /// Start listening for `streamNotify` events from the VM service and forward | ||
| /// them to the clients which have subscribed to the stream. | ||
| void listen() => dds._vmServiceClient.registerMethod( | ||
| 'streamNotify', | ||
| (parameters) { | ||
| final streamId = parameters['streamId'].asString; | ||
| streamNotify(streamId, parameters.value); | ||
| }, | ||
| ); | ||
|
|
||
| /// Subscribes `client` to a stream. | ||
| /// | ||
| /// If `client` is the first client to listen to `stream`, DDS will send a | ||
| /// `streamListen` request for `stream` to the VM service. | ||
| Future<void> streamListen( | ||
| _DartDevelopmentServiceClient client, | ||
| String stream, | ||
| ) async { | ||
| assert(stream != null && stream.isNotEmpty); | ||
| if (!streamListeners.containsKey(stream)) { | ||
| // This will return an RPC exception if the stream doesn't exist. This | ||
| // will throw and the exception will be forwarded to the client. | ||
| final result = await dds._vmServiceClient.sendRequest('streamListen', { | ||
| 'streamId': stream, | ||
| }); | ||
| assert(result['type'] == 'Success'); | ||
| streamListeners[stream] = <_DartDevelopmentServiceClient>[]; | ||
| } | ||
| if (streamListeners[stream].contains(client)) { | ||
| throw kStreamAlreadySubscribedException; | ||
| } | ||
| streamListeners[stream].add(client); | ||
| } | ||
|
|
||
| /// Unsubscribes `client` from a stream. | ||
| /// | ||
| /// If `client` is the last client to unsubscribe from `stream`, DDS will | ||
| /// send a `streamCancel` request for `stream` to the VM service. | ||
| Future<void> streamCancel( | ||
| _DartDevelopmentServiceClient client, | ||
| String stream, | ||
| ) async { | ||
| assert(stream != null && stream.isNotEmpty); | ||
| final listeners = streamListeners[stream]; | ||
| if (listeners == null || !listeners.contains(client)) { | ||
| throw kStreamNotSubscribedException; | ||
| } | ||
| listeners.remove(client); | ||
| if (listeners.isEmpty) { | ||
| streamListeners.remove(stream); | ||
| final result = await dds._vmServiceClient.sendRequest('streamCancel', { | ||
| 'streamId': stream, | ||
| }); | ||
| assert(result['type'] == 'Success'); | ||
| } else { | ||
| streamListeners[stream] = listeners; | ||
| } | ||
| } | ||
|
|
||
| /// Cleanup stream subscriptions for `client` when it has disconnected. | ||
| void clientDisconnect(_DartDevelopmentServiceClient client) { | ||
| for (final streamId in streamListeners.keys.toList()) { | ||
| streamCancel(client, streamId); | ||
| } | ||
| } | ||
|
|
||
| // These error codes must be kept in sync with those in vm/json_stream.h and | ||
| // vmservice.dart. | ||
| static const kStreamAlreadySubscribed = 103; | ||
| static const kStreamNotSubscribed = 104; | ||
|
|
||
| // Keep these messages in sync with the VM service. | ||
| static final kStreamAlreadySubscribedException = json_rpc.RpcException( | ||
| kStreamAlreadySubscribed, | ||
| 'Stream already subscribed', | ||
| ); | ||
|
|
||
| static final kStreamNotSubscribedException = json_rpc.RpcException( | ||
| kStreamNotSubscribed, | ||
| 'Stream not subscribed', | ||
| ); | ||
|
|
||
| final _DartDevelopmentService dds; | ||
| final streamListeners = <String, List<_DartDevelopmentServiceClient>>{}; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.