Revert "Add the streamListen and streamCancel rpcs to the vm service."

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);
@@ skipping 414 matching lines @@
   @observable Duration get upTime =>
       (new DateTime.now().difference(startTime));
 
   VM() : super._empty(null) {
     name = 'vm';
     vmName = 'vm';
     _cache['vm'] = this;
     update(toObservable({'id':'vm', 'type':'@VM'}));
   }
 
-  Future streamListen(String streamId) {
-    return invokeRpc('streamListen', {'streamId': streamId});
-  }
+  final StreamController<ServiceEvent> events =
+      new StreamController.broadcast();
 
-  Future streamCancel(String streamId) {
-    return invokeRpc('streamCancel', {'streamId': streamId});
-  }
-
-  static const kIsolateEventStreamId = 'Isolate';
-  static const kDebugEventStreamId = 'Debug';
-  static const kGCEventStreamId = 'GC';
-
-  Map<String,StreamController> _streamControllers = {};
-
-  void _streamListenHandleError(String streamId) {
-    streamListen(streamId).catchError((e, st) {
-      _getEventStreamController(streamId).addError(e, st);
-    });
-  }
-
-  void _streamCancelHandleError(String streamId) {
-    // TODO(turnidge): Consider removing the controller from the map here.
-    streamCancel(streamId).catchError((e, st) {
-      _getEventStreamController(streamId).addError(e, st);
-    });
-  }
-
-  StreamController _getEventStreamController(String streamId) {
-    var controller = _streamControllers.putIfAbsent(
-        streamId, () {
-          return new StreamController.broadcast(
-            onListen: () => _streamListenHandleError(streamId),
-            onCancel: () => _streamCancelHandleError(streamId));
-        });
-    return controller;
-  }
-
-  Stream getEventStream(String streamId) {
-    return _getEventStreamController(streamId).stream;
-  }
-
-  Stream get isolateEvents => getEventStream(kIsolateEventStreamId);
-  Stream get debugEvents => getEventStream(kDebugEventStreamId);
-  Stream get gcEvents => getEventStream(kGCEventStreamId);
-
-  void postServiceEvent(String streamId, Map response, ByteData data) {
-    assert(streamId != null);
+  void postServiceEvent(Map response, ByteData data) {
     var map = toObservable(response);
     assert(!map.containsKey('_data'));
     if (data != null) {
       map['_data'] = data;
     }
     if (map['type'] != 'ServiceEvent') {
       Logger.root.severe(
           "Expected 'ServiceEvent' but found '${map['type']}'");
       return;
     }
 
     var eventIsolate = map['isolate'];
-    var event;
     if (eventIsolate == null) {
-      event = new ServiceObject._fromMap(vm, map);
+      var event = new ServiceObject._fromMap(vm, map);
+      events.add(event);
     } else {
       // getFromMap creates the Isolate if it hasn't been seen already.
       var isolate = getFromMap(map['isolate']);
-      event = new ServiceObject._fromMap(isolate, map);
+      var event = new ServiceObject._fromMap(isolate, map);
       if (event.eventType == ServiceEvent.kIsolateExit) {
         _removeIsolate(isolate.id);
       }
-      // Give the isolate a chance to process the event first before
-      // dispatching it to others.
       isolate._onEvent(event);
+      events.add(event);
     }
-    _getEventStreamController(streamId).add(event);
   }
 
   void _removeIsolate(String isolateId) {
     assert(_isolateCache.containsKey(isolateId));
     _isolateCache.remove(isolateId);
     notifyPropertyChange(#isolates, true, false);
   }
 
   void _removeDeadIsolates(List newIsolates) {
     // Build a set of new isolates.
@@ skipping 536 matching lines @@
       latestSnapshot.graph.process(_snapshotFetch).then((graph) {
         _snapshotFetch.add(latestSnapshot);
         _snapshotFetch.close();
       });
     }
   }
 
   Stream fetchHeapSnapshot() {
     if (_snapshotFetch == null || _snapshotFetch.isClosed) {
       _snapshotFetch = new StreamController();
-      isolate.vm.streamListen('_Graph');
+      // isolate.vm.streamListen('_Graph');
       isolate.invokeRpcNoUpgrade('requestHeapSnapshot', {});
     }
     return _snapshotFetch.stream;
   }
 
   void updateHeapsFromMap(ObservableMap map) {
     newSpace.update(map['new']);
     oldSpace.update(map['old']);
   }
 
@@ skipping 2123 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);
     }
   }
 }