2nd attempt at adding streamListen/streamCancel to the service protocol.

runtime/observatory/lib/src/service/object.dart

 // Copyright (c) 2014, 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 service;
 
 /// An RpcException represents an exceptional event that happened
 /// while invoking an rpc.
 abstract class RpcException implements Exception {
   RpcException(this.message);
 
   String message;
 }
 
 /// A ServerRpcException represents an error returned by the VM.
 class ServerRpcException extends RpcException {
   /// A list of well-known server error codes.
   static const kParseError     = -32700;
   static const kInvalidRequest = -32600;
   static const kMethodNotFound = -32601;
   static const kInvalidParams  = -32602;
   static const kInternalError  = -32603;
-  static const kFeatureDisabled   = 100;
-  static const kVMMustBePaused    = 101;
-  static const kCannotAddBreakpoint = 102;
+  static const kFeatureDisabled         = 100;
+  static const kVMMustBePaused          = 101;
+  static const kCannotAddBreakpoint     = 102;
+  static const kStreamAlreadySubscribed = 103;
+  static const kStreamNotSubscribed     = 104;
 
   int code;
   Map data;
 
   static _getMessage(Map errorMap) {
     Map data = errorMap['data'];
     if (data != null && data['details'] != null) {
       return data['details'];
     } else {
       return errorMap['message'];
@@ skipping 517 matching lines @@
     return invokeRpcNoUpgrade(method, params).then((ObservableMap response) {
       var obj = new ServiceObject._fromMap(this, response);
       if ((obj != null) && obj.canCache) {
         String objId = obj.id;
         _cache.putIfAbsent(objId, () => obj);
       }
       return obj;
     });
   }
 
-  Future<ObservableMap> _fetchDirect() {
-    return invokeRpcNoUpgrade('getVM', {});
+  Future<ObservableMap> _fetchDirect() async {
+    if (!loaded) {
+      // TODO(turnidge): Instead of always listening to all streams,
+      // implement a stream abstraction in the service library so
+      // that we only subscribe to the streams we want.
+      await _streamListen('Isolate');
+      await _streamListen('Debug');
+      await _streamListen('GC');
+      await _streamListen('_Echo');
+      await _streamListen('_Graph');
+    }
+    return await invokeRpcNoUpgrade('getVM', {});
   }
 
   Future<ServiceObject> getFlagList() {
     return invokeRpc('getFlagList', {});
   }
 
+  Future<ServiceObject> _streamListen(String streamId) {
+    Map params = {
+      'streamId': streamId,
+    };
+    return invokeRpc('streamListen', params);
+  }
+
   /// Force the VM to disconnect.
   void disconnect();
   /// Completes when the VM first connects.
   Future get onConnect;
   /// Completes when the VM disconnects or there was an error connecting.
   Future get onDisconnect;
 
   void _update(ObservableMap map, bool mapIsRef) {
     if (mapIsRef) {
       return;
@@ skipping 2667 matching lines @@
     var v = list[i];
     if ((v is ObservableMap) && _isServiceMap(v)) {
       list[i] = owner.getFromMap(v);
     } else if (v is ObservableList) {
       _upgradeObservableList(v, owner);
     } else if (v is ObservableMap) {
       _upgradeObservableMap(v, owner);
     }
   }
 }