Make VM event streams look like real dart streams.

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;
 
+/// Helper function for canceling a Future<StreamSubscription>.
+Future cancelFutureSubscription(
+    Future<StreamSubscription> subscriptionFuture) async {
+  if (subscriptionFuture != null) {
+    var subscription = await subscriptionFuture;
+    return subscription.cancel();
+  } else {
+    return null;
+  }
+}
+
 /// 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 {
@@ skipping 402 matching lines @@
 
   String toString() {
     if (endTokenPos == null) {
       return '${script.name}:token(${tokenPos})';
     } else {
       return '${script.name}:tokens(${tokenPos}-${endTokenPos})';
     }
   }
 }
 
+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;
   @reflectable Isolate get isolate => null;
 
   // TODO(turnidge): The connection should not be stored in the VM object.
   bool get isDisconnected;
 
   // TODO(johnmccutchan): Ensure that isolates do not end up in _cache.
   Map<String,ServiceObject> _cache = new Map<String,ServiceObject>();
@@ skipping 13 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'}));
   }
 
-  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) {
       map['_data'] = data;
     }
     if (map['type'] != 'Event') {
       Logger.root.severe(
           "Expected 'Event' but found '${map['type']}'");
       return;
     }
 
     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}'");
     }
   }
 
   void _removeIsolate(String isolateId) {
     assert(_isolateCache.containsKey(isolateId));
     _isolateCache.remove(isolateId);
     notifyPropertyChange(#isolates, true, false);
   }
 
   void _removeDeadIsolates(List newIsolates) {
@@ skipping 79 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;
     });
   }
 
+  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 listenEventStream(kIsolateStream, _dispatchEventToIsolate);
+      await listenEventStream(kDebugStream, _dispatchEventToIsolate);
+      await listenEventStream(_kGraphStream, _dispatchEventToIsolate);
     }
     return await invokeRpcNoUpgrade('getVM', {});
   }
 
   Future<ServiceObject> getFlagList() {
     return invokeRpc('getFlagList', {});
   }
 
   Future<ServiceObject> _streamListen(String streamId) {
     Map params = {
       'streamId': streamId,
     };
     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 = {};
+
+  // Well-known stream ids.
+  static const kIsolateStream = 'Isolate';
+  static const kDebugStream = 'Debug';
+  static const kGCStream = 'GC';
+  static const _kGraphStream = '_Graph';
+
+  /// 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();
+  }
+
+  /// Helper function for listening to an event stream.
+  Future<StreamSubscription> listenEventStream(String streamId,
+                                               Function function) async {
+    var stream = await getEventStream(streamId);
+    return stream.listen(function);
+  }
+
   /// 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 2829 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);
     }
   }
 }