Improve VM connection flow in Observatory

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;
 
 // Some value smaller than the object ring, so requesting a large array
 // doesn't result in an expired ref because the elements lapped it in the
 // object ring.
 const int kDefaultFieldLimit = 100;
@@ skipping 571 matching lines @@
     _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();
         }
+      }).catchError((e) {
+        /* ignore */
       });
     }
     // 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;
+      try {
+        await _cancelFuture;
+      } on NetworkRpcException catch (_) { /* ignore */ }
     }
     if (_listenFuture == null) {
       _listenFuture = _vm._streamListen(streamId);
     }
-    await _listenFuture;
+    try {
+      await _listenFuture;
+    } on NetworkRpcException catch (_) { /* ignore */ }
     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 implements M.VM {
   VM get vm => this;
   Isolate get isolate => null;
 
   // TODO(turnidge): The connection should not be stored in the VM object.
   bool get isDisconnected;
+  bool get isConnected;
 
   // Used for verbose logging.
   bool verbose = false;
 
   // TODO(johnmccutchan): Ensure that isolates do not end up in _cache.
   Map<String,ServiceObject> _cache = new Map<String,ServiceObject>();
   final Map<String,Isolate> _isolateCache = <String,Isolate>{};
 
   // The list of live isolates, ordered by isolate start time.
   final List<Isolate> isolates = <Isolate>[];
@@ skipping 169 matching lines @@
   }
 
   Future<ServiceObject> invokeRpc(String method, Map params) {
     return invokeRpcNoUpgrade(method, params).then((Map response) {
       var obj = new ServiceObject._fromMap(this, response);
       if ((obj != null) && obj.canCache) {
         String objId = obj.id;
         _cache.putIfAbsent(objId, () => obj);
       }
       return obj;
+    }).catchError((e) {
+      return new Future.error(e);
     });
   }
 
   void _dispatchEventToIsolate(ServiceEvent event) {
     var isolate = event.isolate;
     if (isolate != null) {
       isolate._onEvent(event);
     }
   }
 
   Future restart() {
     return invokeRpc('_restartVM', {});
   }
 
   Future<Map> _fetchDirect({int count: kDefaultFieldLimit}) async {
     if (!loaded) {
       // The vm service relies on these events to keep the VM and
       // Isolate types up to date.
       try {
         await listenEventStream(kVMStream, _dispatchEventToIsolate);
         await listenEventStream(kIsolateStream, _dispatchEventToIsolate);
         await listenEventStream(kDebugStream, _dispatchEventToIsolate);
         await listenEventStream(_kGraphStream, _dispatchEventToIsolate);
       } on FakeVMRpcException catch (_) {
         // ignore FakeVMRpcExceptions here.
+      } on NetworkRpcException catch (_) {
+        // ignore network errors here.
       }
     }
     return await invokeRpcNoUpgrade('getVM', {});
   }
 
   Future setName(String newName) {
     return invokeRpc('setVMName', { 'name': newName });
   }
 
   Future<ServiceObject> getFlagList() {
     return invokeRpc('getFlagList', {});
   }
 
   Future<ServiceObject> _streamListen(String streamId) {
     Map params = {
       'streamId': streamId,
     };
-    return invokeRpc('streamListen', params);
+    return invokeRpc('streamListen', params).catchError((e) {
+      // Ignore network errors on stream listen.
+      if (e is NetworkRpcException) {
+        return null;
+      }
+    });
   }
 
   Future<ServiceObject> _streamCancel(String streamId) {
     Map params = {
       'streamId': streamId,
     };
-    return invokeRpc('streamCancel', params);
+    return invokeRpc('streamCancel', params).catchError((e) {
+      // Ignore network errors on stream cancel.
+      if (e is NetworkRpcException) {
+        return null;
+      }
+    });
   }
 
   // A map from stream id to event stream state.
   Map<String,_EventStreamState> _eventStreams = {};
 
   // Well-known stream ids.
   static const kVMStream = 'VM';
   static const kIsolateStream = 'Isolate';
   static const kTimelineStream = 'Timeline';
   static const kDebugStream = 'Debug';
   static const kGCStream = 'GC';
   static const kStdoutStream = 'Stdout';
   static const kStderrStream = 'Stderr';
   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();
+    Stream stream = await eventStream.addStream();
+    return stream;
   }
 
   /// 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.
@@ skipping 83 matching lines @@
 
   // Always connected.
   Future _onConnect;
   Future get onConnect {
     if (_onConnect != null) {
       return _onConnect;
     }
     _onConnect = new Future.value(this);
     return _onConnect;
   }
+  bool get isConnected => !isDisconnected;
   // Only complete when requested.
   Completer _onDisconnect = new Completer();
   Future get onDisconnect => _onDisconnect.future;
   bool get isDisconnected => _onDisconnect.isCompleted;
 
   Future<Map> invokeRpcRaw(String method, Map params) {
     if (params.isEmpty) {
       params = null;
     }
     var key = _canonicalizeUri(new Uri(path: method, queryParameters: params));
@@ skipping 3361 matching lines @@
     var v = list[i];
     if ((v is Map) && _isServiceMap(v)) {
       list[i] = owner.getFromMap(v);
     } else if (v is List) {
       _upgradeList(v, owner);
     } else if (v is Map) {
       _upgradeMap(v, owner);
     }
   }
 }