Make VM event streams look like real dart streams.

runtime/observatory/lib/src/elements/heap_profile.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.
 
 library heap_profile_element;
 
 import 'dart:async';
 import 'dart:html';
 import 'observatory_element.dart';
 import 'package:observatory/app.dart';
@@ skipping 25 matching lines @@
   var _newPieChart;
 
   // Pie chart of old space usage.
   var _oldPieDataTable;
   var _oldPieChart;
 
   @observable ClassSortedTable classTable;
   var _classTableBody;
 
   @published bool autoRefresh = false;
-  var _subscription;
+  var _subscriptionFuture;
 
   @published Isolate isolate;
   @observable ServiceMap profile;
 
   HeapProfileElement.created() : super.created() {
     // Create pie chart models.
     _newPieDataTable = new DataTable();
     _newPieDataTable.addColumn('string', 'Type');
     _newPieDataTable.addColumn('number', 'Size');
     _oldPieDataTable = new DataTable();
@@ skipping 29 matching lines @@
 
   @override
   void attached() {
     super.attached();
     // Grab the pie chart divs.
     _newPieChart = new Chart('PieChart',
         shadowRoot.querySelector('#newPieChart'));
     _oldPieChart = new Chart('PieChart',
         shadowRoot.querySelector('#oldPieChart'));
     _classTableBody = shadowRoot.querySelector('#classTableBody');
-    _subscription = app.vm.events.stream.where(
-        (event) => event.isolate == isolate).listen(_onEvent);
+    _subscriptionFuture =
+        app.vm.listenEventStream(VM.kGCStream, _onEvent);
   }
 
   @override
   void detached() {
-    _subscription.cancel();
+    cancelFutureSubscription(_subscriptionFuture);
+    _subscriptionFuture = null;
     super.detached();
   }
 
   // Keep at most one outstanding auto-refresh RPC.
   bool refreshAutoPending = false;
   bool refreshAutoQueued = false;
 
   void _onEvent(ServiceEvent event) {
-    if (autoRefresh && event.kind == 'GC') {
+    assert(event.kind == 'GC');
+    if (autoRefresh) {
       if (!refreshAutoPending) {
         refreshAuto();
       } else {
         // Remember to refresh once more, to ensure latest profile.
         refreshAutoQueued = true;
       }
     }
   }
 
   void refreshAuto() {
@@ skipping 241 matching lines @@
   }
 
   @observable String formattedTotalCollectionTime(bool newSpace) {
     if (profile == null) {
       return '';
     }
     var heap = newSpace ? isolate.newSpace : isolate.oldSpace;
     return '${Utils.formatSeconds(heap.totalCollectionTimeInSeconds)} secs';
   }
 }