add sendRequest to VMIsolateRef

lib/src/isolate.dart

 // Copyright (c) 2015, 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.
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'package:async/async.dart';
 import 'package:crypto/crypto.dart';
 import 'package:json_rpc_2/json_rpc_2.dart' as rpc;
@@ skipping 50 matching lines @@
   /// A broadcast stream that emits a [VMPauseEvent] whenever this isolate is
   /// paused or resumed.
   Stream<VMPauseEvent> get onPauseOrResume => _onPauseOrResume;
   Stream<VMPauseEvent> _onPauseOrResume;
 
   /// A broadcast stream that emits a [VMBreakpoint] whenever a breakpoint is
   /// added.
   Stream<VMBreakpoint> get onBreakpointAdded => _onBreakpointAdded;
   Stream<VMBreakpoint> _onBreakpointAdded;
 
+  /// A broadcast stream that emits custom events posted via `postEvent` from
+  /// the `dart:developer` package.
+  Stream<VMExtensionEvent> get onExtensionEvent => _onExtensionEvent;
+  Stream<VMExtensionEvent> _onExtensionEvent;
+
+  /// A broadcast stream that emits custom events posted via `postEvent` from
+  /// the `dart:developer` package, limited to specific [kind].
+  ///
+  /// If [prefix] is `true`, then [kind] is matched to the event kind as a
+  /// prefix.
+  Stream<VMExtensionEvent> selectExtensionEvents(String kind,
+          {bool prefix: false}) =>
+      transform(_onExtensionEvent, new _VMExtensionEventSelector(kind, prefix));

[nweiz, 2016/02/17 21:35:41]

I'd rather this just be an anonymous function.

[yjbanov, 2016/02/17 23:51:43]

Done.

+
   /// A broadcast stream that emits this isolate's standard output.
   ///
   /// This is only usable for embedders that provide access to `dart:io`.
   ///
   /// Note that as of the VM service version 3.0, this stream doesn't emit
   /// strings passed to the `print()` function unless the host process's
   /// standard output is actively being drained (see [sdk#24351][]).
   ///
   /// [sdk#24351]: https://github.com/dart-lang/sdk/issues/24351
   Stream<List<int>> get stdout => _stdout;
   Stream<List<int>> _stdout;
 
   /// A broadcast stream that emits this isolate's standard error.
   ///
   /// This is only usable for embedders that provide access to `dart:io`.
   Stream<List<int>> get stderr => _stderr;
   Stream<List<int>> _stderr;
 
+  /// A broadcast stream that emits the name of VM service extensions
+  Stream<String> get onServiceExtensionAdded =>
+      _onServiceExtensionAdded;
+  Stream<String> _onServiceExtensionAdded;
+
   /// A future that fires when the isolate exits.
   ///
   /// If the isolate has already exited, this will complete immediately.
   Future get onExit => _onExitMemo.runOnce(() async {
     try {
       await _scope.getInState(_scope.streams.isolate, () async {
         try {
           await load();
           return null;
         } on VMSentinelException catch (_) {
@@ skipping 20 matching lines @@
         name = json["name"] {
     _onGC = _transform(_scope.streams.gc, (json, sink) {
       if (json["kind"] == "GC") sink.add(null);
     });
 
     _onUpdate = _transform(_scope.streams.isolate, (json, sink) {
       if (json["kind"] != "IsolateUpdate") return;
       sink.add(new VMIsolateRef._(_scope, json["isolate"]));
     });
 
+    _onServiceExtensionAdded = _transform(_scope.streams.isolate, (json, sink) {
+      if (json["kind"] != "ServiceExtensionAdded") return;
+      sink.add(json["extensionRPC"]);
+    });
+
     _onPauseOrResume = _transform(_scope.streams.debug, (json, sink) {
       var event = newVMPauseEvent(_scope, json);
       if (event != null) sink.add(event);
     });
 
     _onBreakpointAdded = _transform(_scope.streams.debug, (json, sink) {
       if (json["kind"] != "BreakpointAdded") return;
       sink.add(newVMBreakpoint(_scope, json["breakpoint"]));
     });
 
     _stdout = _transform(_scope.streams.stdout, (json, sink) {
       if (json["kind"] != "WriteEvent") return;
       var bytes = CryptoUtils.base64StringToBytes(json["bytes"]);
       sink.add(bytes);
     });
 
     _stderr = _transform(_scope.streams.stderr, (json, sink) {
       if (json["kind"] != "WriteEvent") return;
       sink.add(CryptoUtils.base64StringToBytes(json["bytes"]));
     });
+
+    _onExtensionEvent = _transform(_scope.streams.extension, (json, sink) {
+      sink.add(
+          new VMExtensionEvent._(json['extensionKind'], json['extensionData']));
+    });
   }
 
   /// Like [transform], but only calls [handleData] for events related to this
   /// isolate.
   Stream _transform(Stream<Map> stream, handleData(Map json, StreamSink sink)) {
     return transform(stream, (json, sink) {
       if (json["isolate"]["id"] != _scope.isolateId) return;
       handleData(json, sink);
     });
   }
@@ skipping 98 matching lines @@
       var response = await _scope.sendRequest(
           "addBreakpointWithScriptUri", params);
       return newVMBreakpoint(_scope, response);
     } on rpc.RpcException catch (error) {
       // Error 102 indicates that the breakpoint couldn't be created.
       if (error.code == 102) return null;
       rethrow;
     }
   }
 
+  /// Returns a future that completes once extension with the given [name] is
+  /// available.
+  ///
+  /// This works whether the extension is already registered or has yet to be
+  /// registered.
+  Future waitForExtension(String name) async {
+    return _scope.getInState(_scope.streams.isolate, () async {
+      var extensions = (await load()).extensionRPCs;
+      return extensions.contains(name);
+    }, (Map json) {
+      return json["kind"] == "ServiceExtensionAdded" &&
+          json["extensionRPC"] == name;
+    }).then((_) => null);
+  }
+
+  /// Makes a raw RPC to a VM service extension registered in this isolate
+  /// corresponding to the ID [number].
+  ///
+  /// [method] must correspond to a VM service extension installed on the VM
+  /// isolate and it must begin with prefix "ext.".
+  ///
+  /// [params] are passed to the extension handler and must be serializable to
+  /// a JSON string.
+  Future<Object> invokeExtension(String method, [Map<String, String> params]) {
+    if (!method.startsWith('ext.')) {
+      throw new ArgumentError.value(method, 'method',
+          'must begin with "ext." prefix');
+    }
+    return _scope.sendRequest(method, params);
+  }
+
   bool operator ==(other) => other is VMIsolateRef &&
       other._scope.isolateId == _scope.isolateId;
 
   int get hashCode => _scope.isolateId.hashCode;
 
   String toString() => name;
 }
 
 /// A full isolate on the remote VM.
 class VMIsolate extends VMIsolateRef {
@@ skipping 13 matching lines @@
 
   /// Whether this isolate is paused.
   bool get isPaused => pauseEvent is! VMResumeEvent;
 
   /// The error that's causing the isolate to exit or `null`.
   final VMError error;
 
   /// All breakpoints currently registered for this isolate.
   final List<VMBreakpoint> breakpoints;
 
+  /// The list of service extension RPCs that are registered for this isolate.
+  final List<String> extensionRPCs;

[nweiz, 2016/02/17 21:35:41]

Nit: Following the Dart style guide, this should be "extensionRpcs" since the
acronym is longer than two letters.

[yjbanov, 2016/02/17 23:51:43]

Done.

+
   VMIsolate._(Scope scope, Map json)
       : startTime = new DateTime.fromMillisecondsSinceEpoch(
             // Prior to v3.0, this was emitted as a double rather than an int.
             json["startTime"].round()),
         livePorts = json["livePorts"],
         pauseOnExit = json["pauseOnExit"],
         pauseEvent = newVMPauseEvent(scope, json["pauseEvent"]),
         error = newVMError(scope, json["error"]),
         breakpoints = new UnmodifiableListView(json["breakpoints"]
             .map((breakpoint) => newVMBreakpoint(scope, breakpoint))
             .toList()),
+        extensionRPCs = json["extensionRPCs"] != null
+          ? new UnmodifiableListView(json["extensionRPCs"])
+          : const <String>[],  // it's null when no extensions are registered

[nweiz, 2016/02/17 21:35:41]

Hmm, this is not what's documented. I've filed
https://github.com/dart-lang/sdk/issues/25801 about it.

Also, I think it's fine to just write `new
UnmodifiableListView(json["extensionRPCs"] ?? [])` here.

[yjbanov, 2016/02/17 23:51:43]

Done.

         super._(scope, json);
 }
 
 /// A full isolate on the remote VM that's ready to run code.
 ///
 /// The VM service exposes isolates very early, before their contents are
 /// fully-loaded. These in-progress isolates, represented by plain [VMIsolate]
 /// instances, have limited amounts of metadata available. Only once they're
 /// runnable is the full suite of metadata available.
 ///
@@ skipping 37 matching lines @@
 
   /// The isolate continues until it exits the current function.
   static const out = const VMStep._("Out");
 
   /// The string name of the step type.
   final String _value;
 
   const VMStep._(this._value);
 
   String toString() => _value;
-}
+}
+
+/// An event posted via `postEvent` from the `dart:developer` package.
+class VMExtensionEvent {
+  /// Event kind.
+  final String kind;
+
+  /// Event data.
+  final Map data;
+
+  VMExtensionEvent._(this.kind, this.data);
+}
+
+/// Stream data handler that picks extension events based on event kind.
+class _VMExtensionEventSelector {
+  /// The kind of events the subscriber is interested in.
+  final String kind;
+
+  /// Whether [kind] must match as prefix.
+  final bool prefix;
+
+  _VMExtensionEventSelector(this.kind, this.prefix);
+
+  call(VMExtensionEvent event, EventSink<VMExtensionEvent> sink) {
+    if (event.kind == kind || (prefix && event.kind.startsWith(kind))) {
+      sink.add(event);
+    }
+  }
+}