Make VM event streams look like real dart streams.

runtime/observatory/lib/src/app/application.dart

 // Copyright (c) 2013, 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 app;
 
 class Notification {
   Notification.fromEvent(this.event);
   Notification.fromException(this.exception, this.stacktrace);
 
   ServiceEvent event;
   var exception;
   var stacktrace;
 }
 
 /// The observatory application. Instances of this are created and owned
 /// by the observatory_application custom element.
 class ObservatoryApplication extends Observable {
   static ObservatoryApplication app;
   final _pageRegistry = new List<Page>();
   LocationManager _locationManager;
   LocationManager get locationManager => _locationManager;
   @observable Page currentPage;
   VM _vm;
   VM get vm => _vm;
+
+  Future<StreamSubscription> isolateStreamSubscription;
+  Future<StreamSubscription> debugStreamSubscription;
+
   set vm(VM vm) {
     if (_vm == vm) {
       // Do nothing.
       return;
     }
     if (_vm != null) {
       // Disconnect from current VM.
       notifications.clear();
       _vm.disconnect();
     }
@@ skipping 10 matching lines @@
       vm.onDisconnect.then((String reason) {
         if (this.vm != vm) {
           // This disconnect event occured *after* a new VM was installed.
           return;
         }
         notifications.add(
             new Notification.fromEvent(
                 new ServiceEvent.connectionClosed(reason)));
       });
 
-      vm.events.stream.listen(_onEvent);
+      vm.getIsolateEventStream().then((stream) {
+        stream.listen(_onEvent);
+      });
+      vm.getDebugEventStream().then((stream) {
+        stream.listen(_onEvent);
+      });
     }
     _vm = vm;
   }
   final TargetManager targets;
   @reflectable final ObservatoryApplicationElement rootElement;
 
   TraceViewElement _traceView = null;
 
   @reflectable ServiceObject lastErrorOrException;
   @observable ObservableList<Notification> notifications =
@@ skipping 14 matching lines @@
         return (event != null &&
                 event.isolate == isolate &&
                 event.isPauseEvent);
       });
   }
 
   void _onEvent(ServiceEvent event) {
     switch(event.kind) {
       case ServiceEvent.kIsolateStart:
       case ServiceEvent.kIsolateUpdate:
-      case ServiceEvent.kGraph:
       case ServiceEvent.kBreakpointAdded:
       case ServiceEvent.kBreakpointResolved:
       case ServiceEvent.kBreakpointRemoved:
-      case ServiceEvent.kGC:
       case ServiceEvent.kDebuggerSettingsUpdate:
         // Ignore for now.
         break;
 
       case ServiceEvent.kIsolateExit:
       case ServiceEvent.kResume:
         removePauseEvents(event.isolate);
         break;
 
       case ServiceEvent.kPauseStart:
@@ skipping 113 matching lines @@
     this.vm = new FakeVM(crashDump['result']);
     app.locationManager.go('#/vm');
   }
 
   void handleException(e, st) {
     // TODO(turnidge): Report this failure via analytics.
     Logger.root.warning('Caught exception: ${e}\n${st}');
     notifications.add(new Notification.fromException(e, st));
   }
 }