Make VM event streams look like real dart streams.

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

Index: runtime/observatory/lib/src/service/object.dart
diff --git a/runtime/observatory/lib/src/service/object.dart b/runtime/observatory/lib/src/service/object.dart
index f3ebd195b37c6fc738faf9252981ee3ab5a8c13e..66c5fdfcce7d8a924431166a0fd8714f5d212a23 100644
--- a/runtime/observatory/lib/src/service/object.dart
+++ b/runtime/observatory/lib/src/service/object.dart
@@ -426,6 +426,62 @@ class SourceLocation extends ServiceObject {
   }
 }
 
+class _EventStreamState {
+  VM _vm;
+  String streamId;
+
+  Function _onDone;
+
+  // A list of all subscribed controllers for this stream.
+  List _controllers = [];
+
+  // Completes when the listen rpc is finished.
+  Future _listenFuture;
+
+  // Completes when then cancel rpc is finished.
+  Future _cancelFuture;
+
+  _EventStreamState(this._vm, this.streamId, this._onDone);
+
+  Future _cancelController(StreamController controller) {
+    _controllers.remove(controller);
+    if (_controllers.isEmpty) {
+      assert(_listenFuture != null);
+      _listenFuture = null;
+      _cancelFuture = _vm._streamCancel(streamId);
+      _cancelFuture.then((_) {
+        if (_controllers.isEmpty) {
+          // No new listeners showed up during cancelation.
+          _onDone();
+        }
+      });
+    }
+    // No need to wait for _cancelFuture here.
+    return new Future.value(null);
+  }
+
+  Future<Stream> addStream() async {
+    var controller;
+    controller = new StreamController(
+        onCancel:() => _cancelController(controller));
+    _controllers.add(controller);
+    if (_cancelFuture != null) {
+      await _cancelFuture;
+    }
+    if (_listenFuture == null) {
+      _listenFuture = _vm._streamListen(streamId);
+    }
+    await _listenFuture;
+    return controller.stream;
+  }
+
+  void addEvent(ServiceEvent event) {
+    for (var controller in _controllers) {
+      controller.add(event);
+    }
+  }
+}
+
 /// State for a VM being inspected.
 abstract class VM extends ServiceObjectOwner {
   @reflectable VM get vm => this;
@@ -459,10 +515,7 @@ abstract class VM extends ServiceObjectOwner {
     update(toObservable({'id':'vm', 'type':'@VM'}));
   }
 
-  final StreamController<ServiceEvent> events =
-      new StreamController.broadcast();
-
-  void postServiceEvent(Map response, ByteData data) {
+  void postServiceEvent(String streamId, Map response, ByteData data) {
     var map = toObservable(response);
     assert(!map.containsKey('_data'));
     if (data != null) {
@@ -475,18 +528,22 @@ abstract class VM extends ServiceObjectOwner {
     }
 
     var eventIsolate = map['isolate'];
+    var event;
     if (eventIsolate == null) {
-      var event = new ServiceObject._fromMap(vm, map);
-      events.add(event);
+      event = new ServiceObject._fromMap(vm, map);
     } else {
       // getFromMap creates the Isolate if it hasn't been seen already.
       var isolate = getFromMap(map['isolate']);
-      var event = new ServiceObject._fromMap(isolate, map);
+      event = new ServiceObject._fromMap(isolate, map);
       if (event.kind == ServiceEvent.kIsolateExit) {
         _removeIsolate(isolate.id);
       }
-      isolate._onEvent(event);
-      events.add(event);
+    }
+    var eventStream = _eventStreams[streamId];
+    if (eventStream != null) {
+      eventStream.addEvent(event);
+    } else {
+      Logger.root.warning("Ignoring unexpected event on stream '${streamId}'");
     }
   }
 
@@ -586,16 +643,20 @@ abstract class VM extends ServiceObjectOwner {
     });
   }
 
+  void _dispatchEventToIsolate(ServiceEvent event) {
+    var isolate = event.isolate;
+    if (isolate != null) {
+      isolate._onEvent(event);
+    }
+  }
+
   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');
+      // The vm service relies on these events to keep the VM and
+      // Isolate types up to date.
+      (await getIsolateEventStream()).listen(_dispatchEventToIsolate);
+      (await getDebugEventStream()).listen(_dispatchEventToIsolate);
+      (await getEventStream('_Graph')).listen(_dispatchEventToIsolate);
     }
     return await invokeRpcNoUpgrade('getVM', {});
   }
@@ -611,6 +672,36 @@ abstract class VM extends ServiceObjectOwner {
     return invokeRpc('streamListen', params);
   }
 
+  Future<ServiceObject> _streamCancel(String streamId) {
+    Map params = {
+      'streamId': streamId,
+    };
+    return invokeRpc('streamCancel', params);
+  }
+
+  // A map from stream id to event stream state.
+  Map<String,_EventStreamState> _eventStreams = {};
+
+  /// Returns a single-subscription Stream object for a VM event stream.
+  Future<Stream> getEventStream(String streamId) async {
+    var eventStream = _eventStreams.putIfAbsent(
+        streamId, () => new _EventStreamState(
+            this, streamId, () => _eventStreams.remove(streamId)));
+    return eventStream.addStream();
+  }
+
+  Future<Stream> getIsolateEventStream() {
+    return getEventStream('Isolate');
+  }
+
+  Future<Stream> getDebugEventStream() {
+    return getEventStream('Debug');
+  }
+
+  Future<Stream> getGCEventStream() {
+    return getEventStream('GC');
+  }
+
   /// Force the VM to disconnect.
   void disconnect();
   /// Completes when the VM first connects.