Revert "[flutter_tools] Migrate to vm service 3 (reland): electric boogaloo (#53957)" (#54114)

This reverts commit d6b09626f2.

packages/_flutter_web_build_script/pubspec.yaml

@@ -14,9 +14,9 @@ dependencies:
   test_api: 0.2.15
   test_core: 0.3.3
 
-  build_runner: 1.8.1
+  build_runner: 1.8.0
   build_test: 1.0.0
-  build_runner_core: 5.0.0
+  build_runner_core: 4.5.3
   dart_style: 1.3.3
   code_builder: 3.2.1
   build: 1.2.2
@@ -95,4 +95,4 @@ dartdoc:
   # Exclude this package from the hosted API docs.
   nodoc: true
 
-# PUBSPEC CHECKSUM: 7c61
+# PUBSPEC CHECKSUM: e267

packages/flutter_tools/lib/src/devfs.dart

@@ -4,8 +4,8 @@
 
 import 'dart:async';
 
+import 'package:json_rpc_2/json_rpc_2.dart' as rpc;
 import 'package:meta/meta.dart';
-import 'package:vm_service/vm_service.dart' as vmservice;
 
 import 'asset.dart';
 import 'base/context.dart';
@@ -423,7 +423,7 @@ class DevFS {
     globals.printTrace('DevFS: Creating new filesystem on the device ($_baseUri)');
     try {
       _baseUri = await _operations.create(fsName);
-    } on vmservice.RPCError catch (rpcException) {
+    } on rpc.RpcException catch (rpcException) {
       // 1001 is kFileSystemAlreadyExists in //dart/runtime/vm/json_stream.h
       if (rpcException.code != 1001) {
         rethrow;

packages/flutter_tools/lib/src/fuchsia/fuchsia_device.dart

@@ -620,7 +620,7 @@ class FuchsiaDevice extends Device {
         // loopback (::1).
         final Uri uri = Uri.parse('http://[$_ipv6Loopback]:$port');
         final VMService vmService = await VMService.connect(uri);
-        await vmService.getVMOld();
+        await vmService.getVM();
         await vmService.refreshViews();
         for (final FlutterView flutterView in vmService.vm.views) {
           if (flutterView.uiIsolate == null) {
@@ -717,7 +717,7 @@ class FuchsiaIsolateDiscoveryProtocol {
           continue;
         }
       }
-      await service.getVMOld();
+      await service.getVM();
       await service.refreshViews();
       for (final FlutterView flutterView in service.vm.views) {
         if (flutterView.uiIsolate == null) {

packages/flutter_tools/lib/src/ios/devices.dart

@@ -7,11 +7,11 @@ import 'dart:math' as math;
 
 import 'package:meta/meta.dart';
 import 'package:platform/platform.dart';
-import 'package:vm_service/vm_service.dart' as vm_service;
 import 'package:process/process.dart';
 
 import '../application_package.dart';
 import '../artifacts.dart';
+import '../base/common.dart';
 import '../base/file_system.dart';
 import '../base/io.dart';
 import '../base/logger.dart';
@@ -514,7 +514,7 @@ class IOSDeviceLogReader extends DeviceLogReader {
     // and "Flutter". The regex tries to strike a balance between not producing
     // false positives and not producing false negatives.
     _anyLineRegex = RegExp(r'\w+(\([^)]*\))?\[\d+\] <[A-Za-z]+>: ');
-    _loggingSubscriptions = <StreamSubscription<void>>[];
+    _loggingSubscriptions = <StreamSubscription<ServiceEvent>>[];
   }
 
   /// Create a new [IOSDeviceLogReader].
@@ -554,7 +554,7 @@ class IOSDeviceLogReader extends DeviceLogReader {
   RegExp _anyLineRegex;
 
   StreamController<String> _linesController;
-  List<StreamSubscription<void>> _loggingSubscriptions;
+  List<StreamSubscription<ServiceEvent>> _loggingSubscriptions;
 
   @override
   Stream<String> get logLines => _linesController.stream;
@@ -575,20 +575,18 @@ class IOSDeviceLogReader extends DeviceLogReader {
     if (_majorSdkVersion < _minimumUniversalLoggingSdkVersion) {
       return;
     }
-    try {
+    // The VM service will not publish logging events unless the debug stream is being listened to.
-      await connectedVmService.streamListen('Stdout');
+    // onDebugEvent listens to this stream as a side effect.
-    } on vm_service.RPCError {
+    unawaited(connectedVmService.onDebugEvent);
-      // Do nothing, since the tool is already subscribed.
+    _loggingSubscriptions.add((await connectedVmService.onStdoutEvent).listen((ServiceEvent event) {
-    }
+      final String logMessage = event.message;
-    _loggingSubscriptions.add(connectedVmService.onStdoutEvent.listen((vm_service.Event event) {
+      if (logMessage.isNotEmpty) {
-      final String message = utf8.decode(base64.decode(event.bytes));
+        _linesController.add(logMessage);
-      if (message.isNotEmpty) {
-        _linesController.add(message);
       }
     }));
   }
 
-  void _listenToSysLog() {
+  void _listenToSysLog () {
     // syslog is not written on iOS 13+.
     if (_majorSdkVersion >= _minimumUniversalLoggingSdkVersion) {
       return;
@@ -643,7 +641,7 @@ class IOSDeviceLogReader extends DeviceLogReader {
 
   @override
   void dispose() {
-    for (final StreamSubscription<void> loggingSubscription in _loggingSubscriptions) {
+    for (final StreamSubscription<ServiceEvent> loggingSubscription in _loggingSubscriptions) {
       loggingSubscription.cancel();
     }
     _idevicesyslogProcess?.kill();

packages/flutter_tools/lib/src/resident_runner.dart

@@ -233,7 +233,7 @@ class FlutterDevice {
         : vmService.vm.views).toList();
   }
 
-  Future<void> getVMs() => vmService.getVMOld();
+  Future<void> getVMs() => vmService.getVM();
 
   Future<void> exitApps() async {
     if (!device.supportsFlutterExit) {

packages/flutter_tools/lib/src/run_hot.dart

@@ -3,8 +3,10 @@
 // found in the LICENSE file.
 
 import 'dart:async';
-import 'package:vm_service/vm_service.dart' as vm_service;
+
 import 'package:platform/platform.dart';
+import 'package:json_rpc_2/error_code.dart' as rpc_error_code;
+import 'package:json_rpc_2/json_rpc_2.dart' as rpc;
 import 'package:meta/meta.dart';
 import 'package:pool/pool.dart';
 import 'base/async_guard.dart';
@@ -108,10 +110,9 @@ class HotRunner extends ResidentRunner {
     // TODO(cbernaschina): check that isolateId is the id of the UI isolate.
     final OperationResult result = await restart(pause: pause);
     if (!result.isOk) {
-      throw vm_service.RPCError(
+      throw rpc.RpcException(
+        rpc_error_code.INTERNAL_ERROR,
         'Unable to reload sources',
-        RPCErrorCodes.kInternalError,
-        '',
       );
     }
   }
@@ -120,10 +121,9 @@ class HotRunner extends ResidentRunner {
     final OperationResult result =
       await restart(fullRestart: true, pause: pause);
     if (!result.isOk) {
-      throw vm_service.RPCError(
+      throw rpc.RpcException(
+        rpc_error_code.INTERNAL_ERROR,
         'Unable to restart',
-        RPCErrorCodes.kInternalError,
-        '',
       );
     }
   }
@@ -564,15 +564,16 @@ class HotRunner extends ResidentRunner {
     if (benchmarkMode) {
       final List<Future<void>> isolateNotifications = <Future<void>>[];
       for (final FlutterDevice device in flutterDevices) {
-        try {
-          await device.vmService.streamListen('Isolate');
-        } on vm_service.RPCError {
-          // Do nothing, we're already subcribed.
-        }
         for (final FlutterView view in device.views) {
           isolateNotifications.add(
-            view.owner.vm.vmService.onIsolateEvent.firstWhere((vm_service.Event event) {
+            view.owner.vm.vmService.onIsolateEvent
-              return event.kind == vm_service.EventKind.kServiceExtensionAdded;
+              .then((Stream<ServiceEvent> serviceEvents) async {
+              await for (final ServiceEvent serviceEvent in serviceEvents) {
+                if (serviceEvent.owner.name.contains('_spawn')
+                  && serviceEvent.kind == ServiceEvent.kIsolateExit) {
+                  return;
+                }
+              }
             }),
           );
         }
@@ -719,12 +720,9 @@ class HotRunner extends ResidentRunner {
       if (!result.isOk) {
         restartEvent = 'restart-failed';
       }
-    } on vm_service.SentinelException catch (err, st) {
+    } on rpc.RpcException {
       restartEvent = 'exception';
-      return OperationResult(1, 'hot restart failed to complete: $err\n$st', fatal: true);
+      return OperationResult(1, 'hot restart failed to complete', fatal: true);
-    } on vm_service.RPCError  catch (err, st) {
-      restartEvent = 'exception';
-      return OperationResult(1, 'hot restart failed to complete: $err\n$st', fatal: true);
     } finally {
       HotEvent(restartEvent,
         targetPlatform: targetPlatform,
@@ -766,7 +764,7 @@ class HotRunner extends ResidentRunner {
           );
         },
       );
-    } on vm_service.RPCError {
+    } on rpc.RpcException {
       HotEvent('exception',
         targetPlatform: targetPlatform,
         sdkName: sdkName,

packages/flutter_tools/lib/src/test/coverage_collector.dart

@@ -189,7 +189,7 @@ Future<Map<String, dynamic>> collect(Uri serviceUri, bool Function(String) libra
   Future<VMService> Function(Uri) connector = _defaultConnect,
 }) async {
   final VMService vmService = await connector(serviceUri);
-  await vmService.getVMOld();
+  await vmService.getVM();
   final Map<String, dynamic> result = await _getAllCoverage(
       vmService, libraryPredicate);
   await vmService.close();
@@ -197,7 +197,7 @@ Future<Map<String, dynamic>> collect(Uri serviceUri, bool Function(String) libra
 }
 
 Future<Map<String, dynamic>> _getAllCoverage(VMService service, bool Function(String) libraryPredicate) async {
-  await service.getVMOld();
+  await service.getVM();
   final List<Map<String, dynamic>> coverage = <Map<String, dynamic>>[];
   for (final Isolate isolateRef in service.vm.isolates) {
     await isolateRef.load();

packages/flutter_tools/lib/src/tracing.dart

@@ -3,7 +3,6 @@
 // found in the LICENSE file.
 
 import 'dart:async';
-import 'package:vm_service/vm_service.dart' as vm_service;
 
 import 'base/file_system.dart';
 import 'base/logger.dart';
@@ -46,8 +45,7 @@ class Tracing {
       );
       try {
         final Completer<void> whenFirstFrameRendered = Completer<void>();
-        await vmService.streamListen('Extension');
+        (await vmService.onExtensionEvent).listen((ServiceEvent event) {
-        vmService.onExtensionEvent.listen((vm_service.Event event) {
           if (event.extensionKind == 'Flutter.FirstFrame') {
             whenFirstFrameRendered.complete();
           }

packages/flutter_tools/lib/src/vmservice.dart

@@ -5,14 +5,18 @@
 import 'dart:async';
 import 'dart:math' as math;
 
+import 'package:json_rpc_2/error_code.dart' as rpc_error_code;
+import 'package:json_rpc_2/json_rpc_2.dart' as rpc;
 import 'package:meta/meta.dart' show required;
-import 'package:vm_service/vm_service.dart' as vm_service;
+import 'package:stream_channel/stream_channel.dart';
+import 'package:web_socket_channel/io.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
 
 import 'base/common.dart';
 import 'base/context.dart';
 import 'base/io.dart' as io;
 import 'base/utils.dart';
-import 'convert.dart' show base64, json, utf8;
+import 'convert.dart' show base64, utf8;
 import 'device.dart';
 import 'globals.dart' as globals;
 import 'version.dart';
@@ -21,24 +25,10 @@ import 'version.dart';
 /// for [WebSocket]s (used by tests).
 typedef WebSocketConnector = Future<io.WebSocket> Function(String url, {io.CompressionOptions compression});
 
-WebSocketConnector _openChannel = _defaultOpenChannel;
+/// A function that opens a two-way communication channel to the specified [uri].
+typedef _OpenChannel = Future<StreamChannel<String>> Function(Uri uri, {io.CompressionOptions compression});
 
-/// The error codes for the JSON-RPC standard.
+_OpenChannel _openChannel = _defaultOpenChannel;
-///
-/// See also: https://www.jsonrpc.org/specification#error_object
-abstract class RPCErrorCodes {
-  /// The method does not exist or is not available.
-  static const int kMethodNotFound = -32601;
-
-  /// Invalid method parameter(s), such as a mismatched type.
-  static const int kInvalidParams = -32602;
-
-  /// Internal JSON-RPC error.
-  static const int kInternalError = -32603;
-
-  /// Application specific error codes.s
-  static const int kServerError = -32000;
-}
 
 /// A function that reacts to the invocation of the 'reloadSources' service.
 ///
@@ -74,9 +64,7 @@ typedef ReloadMethod = Future<void> Function({
   String libraryId,
 });
 
-Future<io.WebSocket> _defaultOpenChannel(String url, {
+Future<StreamChannel<String>> _defaultOpenChannel(Uri uri, {io.CompressionOptions compression = io.CompressionOptions.compressionDefault}) async {
-  io.CompressionOptions compression = io.CompressionOptions.compressionDefault
-}) async {
   Duration delay = const Duration(milliseconds: 100);
   int attempts = 0;
   io.WebSocket socket;
@@ -102,14 +90,14 @@ Future<io.WebSocket> _defaultOpenChannel(String url, {
   while (socket == null) {
     attempts += 1;
     try {
-      socket = await constructor(url, compression: compression);
+      socket = await constructor(uri.toString(), compression: compression);
     } on io.WebSocketException catch (e) {
       await handleError(e);
     } on io.SocketException catch (e) {
       await handleError(e);
     }
   }
-  return socket;
+  return IOWebSocketChannel(socket).cast<String>();
 }
 
 /// Override `VMServiceConnector` in [context] to return a different VMService
@@ -124,10 +112,10 @@ typedef VMServiceConnector = Future<VMService> Function(Uri httpUri, {
 });
 
 /// A connection to the Dart VM Service.
-///
+// TODO(mklim): Test this, https://github.com/flutter/flutter/issues/23031
-/// This also implements the package:vm_service API to enable a gradual migration.
+class VMService {
-class VMService implements vm_service.VmService {
   VMService(
+    this._peer,
     this.httpAddress,
     this.wsAddress,
     ReloadSources reloadSources,
@@ -135,49 +123,38 @@ class VMService implements vm_service.VmService {
     CompileExpression compileExpression,
     Device device,
     ReloadMethod reloadMethod,
-    this._delegateService,
-    this.streamClosedCompleter,
-    Stream<dynamic> secondary,
   ) {
     _vm = VM._empty(this);
+    _peer.listen().catchError(_connectionError.completeError);
 
-    // TODO(jonahwilliams): this is temporary to support the current vm_service
+    _peer.registerMethod('streamNotify', (rpc.Parameters event) {
-    // semantics of update-in-place.
+      _handleStreamNotify(event.asMap.cast<String, dynamic>());
-    secondary.listen((dynamic rawData) {
-      final String message = rawData as String;
-      final dynamic map = json.decode(message);
-      if (map != null && map['method'] == 'streamNotify') {
-        _handleStreamNotify(map['params'] as Map<String, dynamic>);
-      }
     });
 
     if (reloadSources != null) {
-      _delegateService.registerServiceCallback('reloadSources', (Map<String, dynamic> params) async {
+      _peer.registerMethod('reloadSources', (rpc.Parameters params) async {
         final String isolateId = params['isolateId'].value as String;
-        final bool force = params['force'] as bool ?? false;
+        final bool force = params.asMap['force'] as bool ?? false;
-        final bool pause = params['pause'] as bool ?? false;
+        final bool pause = params.asMap['pause'] as bool ?? false;
 
         if (isolateId.isEmpty) {
-          throw vm_service.RPCError(
+          throw rpc.RpcException.invalidParams("Invalid 'isolateId': $isolateId");
-            "Invalid 'isolateId': $isolateId",
-            RPCErrorCodes.kInvalidParams,
-            '',
-          );
         }
         try {
           await reloadSources(isolateId, force: force, pause: pause);
           return <String, String>{'type': 'Success'};
-        } on vm_service.RPCError {
+        } on rpc.RpcException {
           rethrow;
         } on Exception catch (e, st) {
-          throw vm_service.RPCError(
+          throw rpc.RpcException(rpc_error_code.SERVER_ERROR,
-            'Error during Sources Reload: $e\n$st',
+              'Error during Sources Reload: $e\n$st');
-            RPCErrorCodes.kServerError,
-            '',
-          );
         }
       });
-      _delegateService.registerService('reloadSources', 'Flutter Tools');
+
+      _peer.sendNotification('registerService', <String, String>{
+        'service': 'reloadSources',
+        'alias': 'Flutter Tools',
+      });
 
     }
 
@@ -191,23 +168,15 @@ class VMService implements vm_service.VmService {
       // if the build method of a StatelessWidget is updated, this is the name of class.
       // If the build method of a StatefulWidget is updated, then this is the name
       // of the Widget class that created the State object.
-      _delegateService.registerServiceCallback('reloadMethod', (Map<String, dynamic> params) async {
+      _peer.registerMethod('reloadMethod', (rpc.Parameters params) async {
-        final String libraryId = params['library'] as String;
+        final String libraryId = params['library'].value as String;
-        final String classId = params['class'] as String;
+        final String classId = params['class'].value as String;
 
         if (libraryId.isEmpty) {
-          throw vm_service.RPCError(
+          throw rpc.RpcException.invalidParams("Invalid 'libraryId': $libraryId");
-            "Invalid 'libraryId': $libraryId",
-            RPCErrorCodes.kInvalidParams,
-            '',
-          );
         }
         if (classId.isEmpty) {
-          throw vm_service.RPCError(
+          throw rpc.RpcException.invalidParams("Invalid 'classId': $classId");
-            "Invalid 'classId': $classId",
-            RPCErrorCodes.kInvalidParams,
-            '',
-          );
         }
 
         globals.printTrace('reloadMethod not yet supported, falling back to hot reload');
@@ -218,97 +187,113 @@ class VMService implements vm_service.VmService {
             classId: classId,
           );
           return <String, String>{'type': 'Success'};
-        } on vm_service.RPCError {
+        } on rpc.RpcException {
           rethrow;
         } on Exception catch (e, st) {
-          throw vm_service.RPCError('Error during Sources Reload: $e\n$st', -32000, '');
+          throw rpc.RpcException(rpc_error_code.SERVER_ERROR,
+              'Error during Sources Reload: $e\n$st');
         }
       });
-      _delegateService.registerService('reloadMethod', 'Flutter Tools');
+      _peer.sendNotification('registerService', <String, String>{
+        'service': 'reloadMethod',
+        'alias': 'Flutter Tools',
+      });
     }
 
     if (restart != null) {
-      _delegateService.registerServiceCallback('hotRestart', (Map<String, dynamic> params) async {
+      _peer.registerMethod('hotRestart', (rpc.Parameters params) async {
-        final bool pause = params['pause'] as bool ?? false;
+        final bool pause = params.asMap['pause'] as bool ?? false;
+
+        if (pause is! bool) {
+          throw rpc.RpcException.invalidParams("Invalid 'pause': $pause");
+        }
+
         try {
           await restart(pause: pause);
           return <String, String>{'type': 'Success'};
-        } on vm_service.RPCError {
+        } on rpc.RpcException {
           rethrow;
         } on Exception catch (e, st) {
-          throw vm_service.RPCError(
+          throw rpc.RpcException(rpc_error_code.SERVER_ERROR,
-            'Error during Hot Restart: $e\n$st',
+              'Error during Hot Restart: $e\n$st');
-            RPCErrorCodes.kServerError,
-            '',
-          );
         }
       });
-      _delegateService.registerService('hotRestart', 'Flutter Tools');
+
+      _peer.sendNotification('registerService', <String, String>{
+        'service': 'hotRestart',
+        'alias': 'Flutter Tools',
+      });
     }
 
-    _delegateService.registerServiceCallback('flutterVersion', (Map<String, dynamic> params) async {
+    _peer.registerMethod('flutterVersion', (rpc.Parameters params) async {
       final FlutterVersion version = context.get<FlutterVersion>() ?? FlutterVersion();
       final Map<String, Object> versionJson = version.toJson();
       versionJson['frameworkRevisionShort'] = version.frameworkRevisionShort;
       versionJson['engineRevisionShort'] = version.engineRevisionShort;
       return versionJson;
     });
-    _delegateService.registerService('flutterVersion', 'Flutter Tools');
+
+    _peer.sendNotification('registerService', <String, String>{
+      'service': 'flutterVersion',
+      'alias': 'Flutter Tools',
+    });
 
     if (compileExpression != null) {
-      _delegateService.registerServiceCallback('compileExpression', (Map<String, dynamic> params) async {
+      _peer.registerMethod('compileExpression', (rpc.Parameters params) async {
-        final String isolateId = params['isolateId'] as String;
+        final String isolateId = params['isolateId'].asString;
         if (isolateId is! String || isolateId.isEmpty) {
-          throw throw vm_service.RPCError(
+          throw rpc.RpcException.invalidParams(
-            "Invalid 'isolateId': $isolateId",
+              "Invalid 'isolateId': $isolateId");
-            RPCErrorCodes.kInvalidParams,
-            '',
-          );
         }
-        final String expression = params['expression'] as String;
+        final String expression = params['expression'].asString;
         if (expression is! String || expression.isEmpty) {
-          throw throw vm_service.RPCError(
+          throw rpc.RpcException.invalidParams(
-            "Invalid 'expression': $expression",
+              "Invalid 'expression': $expression");
-            RPCErrorCodes.kInvalidParams,
-            '',
-          );
         }
-        final List<String> definitions = List<String>.from(params['definitions'] as List<dynamic>);
+        final List<String> definitions =
-        final List<String> typeDefinitions = List<String>.from(params['typeDefinitions'] as List<dynamic>);
+            List<String>.from(params['definitions'].asList);
-        final String libraryUri = params['libraryUri'] as String;
+        final List<String> typeDefinitions =
-        final String klass = params['klass'] as String;
+            List<String>.from(params['typeDefinitions'].asList);
-        final bool isStatic = params['isStatic'] as bool ?? false;
+        final String libraryUri = params['libraryUri'].asString;
+        final String klass = params['klass'].exists ? params['klass'].asString : null;
+        final bool isStatic = params['isStatic'].asBoolOr(false);
+
         try {
           final String kernelBytesBase64 = await compileExpression(isolateId,
               expression, definitions, typeDefinitions, libraryUri, klass,
               isStatic);
-          return <String, dynamic>{
+          return <String, dynamic>{'type': 'Success',
-            'type': 'Success',
+            'result': <String, dynamic> {'kernelBytes': kernelBytesBase64}};
-            'result': <String, dynamic>{
+        } on rpc.RpcException {
-              'result': <String, dynamic>{'kernelBytes': kernelBytesBase64},
-            },
-          };
-        } on vm_service.RPCError {
           rethrow;
         } on Exception catch (e, st) {
-          throw vm_service.RPCError(
+          throw rpc.RpcException(rpc_error_code.SERVER_ERROR,
-            'Error during expression compilation: $e\n$st',
+              'Error during expression compilation: $e\n$st');
-            RPCErrorCodes.kServerError,
-            '',
-          );
         }
       });
-      _delegateService.registerService('compileExpression', 'Flutter Tools');
+
+      _peer.sendNotification('registerService', <String, String>{
+        'service': 'compileExpression',
+        'alias': 'Flutter Tools',
+      });
     }
     if (device != null) {
-      _delegateService.registerServiceCallback('flutterMemoryInfo', (Map<String, dynamic> params) async {
+      _peer.registerMethod('flutterMemoryInfo', (rpc.Parameters params) async {
         final MemoryInfo result = await device.queryMemoryInfo();
         return result.toJson();
       });
-      _delegateService.registerService('flutterMemoryInfo', 'Flutter Tools');
+      _peer.sendNotification('registerService', <String, String>{
+        'service': 'flutterMemoryInfo',
+        'alias': 'Flutter Tools',
+      });
     }
   }
 
+  static void _unhandledError(dynamic error, dynamic stack) {
+    globals.logger.printTrace('Error in internal implementation of JSON RPC.\n$error\n$stack');
+    assert(false);
+  }
+
   /// Connect to a Dart VM Service at [httpUri].
   ///
   /// If the [reloadSources] parameter is not null, the 'reloadSources' service
@@ -346,36 +331,11 @@ class VMService implements vm_service.VmService {
     io.CompressionOptions compression = io.CompressionOptions.compressionDefault,
     Device device,
   }) async {
-    // Create an instance of the package:vm_service API in addition to the flutter
-    // tool's to allow gradual migration.
-    final Completer<void> streamClosedCompleter = Completer<void>();
-
     final Uri wsUri = httpUri.replace(scheme: 'ws', path: globals.fs.path.join(httpUri.path, 'ws'));
-    final io.WebSocket channel = await _openChannel(wsUri.toString(), compression: compression);
+    final StreamChannel<String> channel = await _openChannel(wsUri, compression: compression);
-    final StreamController<dynamic> primary = StreamController<dynamic>();
+    final rpc.Peer peer = rpc.Peer.withoutJson(jsonDocument.bind(channel), onUnhandledError: _unhandledError);
-    final StreamController<dynamic> secondary = StreamController<dynamic>();
-    channel.listen((dynamic data) {
-      primary.add(data);
-      secondary.add(data);
-    }, onDone: ()  {
-      primary.close();
-      secondary.close();
-      streamClosedCompleter.complete();
-    }, onError: (dynamic error, StackTrace stackTrace) {
-      primary.addError(error, stackTrace);
-      secondary.addError(error, stackTrace);
-    });
-
-    final vm_service.VmService delegateService = vm_service.VmService(
-      primary.stream,
-      channel.add,
-      log: null,
-      disposeHandler: () async {
-        streamClosedCompleter.complete();
-      },
-    );
-
     final VMService service = VMService(
+      peer,
       httpUri,
       wsUri,
       reloadSources,
@@ -383,21 +343,17 @@ class VMService implements vm_service.VmService {
       compileExpression,
       device,
       reloadMethod,
-      delegateService,
-      streamClosedCompleter,
-      secondary.stream,
     );
-
     // This call is to ensure we are able to establish a connection instead of
     // keeping on trucking and failing farther down the process.
-    await delegateService.getVersion();
+    await service._sendRequest('getVersion', const <String, dynamic>{});
     return service;
   }
 
-  final vm_service.VmService _delegateService;
   final Uri httpAddress;
   final Uri wsAddress;
-  final Completer<void> streamClosedCompleter;
+  final rpc.Peer _peer;
+  final Completer<Map<String, dynamic>> _connectionError = Completer<Map<String, dynamic>>();
 
   VM _vm;
   /// The singleton [VM] object. Owns [Isolate] and [FlutterView] objects.
@@ -406,36 +362,46 @@ class VMService implements vm_service.VmService {
   final Map<String, StreamController<ServiceEvent>> _eventControllers =
       <String, StreamController<ServiceEvent>>{};
 
+  final Set<String> _listeningFor = <String>{};
+
   /// Whether our connection to the VM service has been closed;
-  bool get isClosed => streamClosedCompleter.isCompleted;
+  bool get isClosed => _peer.isClosed;
 
   Future<void> get done async {
-    return streamClosedCompleter.future;
+    await _peer.done;
   }
 
-  @override
+  // Events
-  Stream<vm_service.Event> get onDebugEvent => onEvent('Debug');
+  Future<Stream<ServiceEvent>> get onDebugEvent => onEvent('Debug');
+  Future<Stream<ServiceEvent>> get onExtensionEvent => onEvent('Extension');
+  // IsolateStart, IsolateRunnable, IsolateExit, IsolateUpdate, ServiceExtensionAdded
+  Future<Stream<ServiceEvent>> get onIsolateEvent => onEvent('Isolate');
+  Future<Stream<ServiceEvent>> get onTimelineEvent => onEvent('Timeline');
+  Future<Stream<ServiceEvent>> get onStdoutEvent => onEvent('Stdout'); // WriteEvent
 
-  @override
+  // TODO(johnmccutchan): Add FlutterView events.
-  Stream<vm_service.Event> get onExtensionEvent => onEvent('Extension');
 
-  @override
+  /// Returns a stream of VM service events.
-  Stream<vm_service.Event> get onIsolateEvent => onEvent('Isolate');
+  ///
-
+  /// This purposely returns a `Future<Stream<T>>` rather than a `Stream<T>`
-  @override
+  /// because it first registers with the VM to receive events on the stream,
-  Stream<vm_service.Event> get onTimelineEvent => onEvent('Timeline');
+  /// and only once the VM has acknowledged that the stream has started will
-
+  /// we return the associated stream. Any attempt to streamline this API into
-  @override
+  /// returning `Stream<T>` should take that into account to avoid race
-  Stream<vm_service.Event> get onStdoutEvent => onEvent('Stdout');
+  /// conditions.
-
+  Future<Stream<ServiceEvent>> onEvent(String streamId) async {
-  @override
+    await _streamListen(streamId);
-  Future<vm_service.Success> streamListen(String streamId) {
+    return _getEventController(streamId).stream;
-    return _delegateService.streamListen(streamId);
   }
 
-  @override
+  Future<Map<String, dynamic>> _sendRequest(
-  Stream<vm_service.Event> onEvent(String streamId) {
+    String method,
-    return _delegateService.onEvent(streamId);
+    Map<String, dynamic> params,
+  ) {
+    return Future.any<Map<String, dynamic>>(<Future<Map<String, dynamic>>>[
+      _peer.sendRequest(method, params).then<Map<String, dynamic>>(castStringKeyedMap),
+      _connectionError.future,
+    ]);
   }
 
   StreamController<ServiceEvent> _getEventController(String eventName) {
@@ -475,20 +441,19 @@ class VMService implements vm_service.VmService {
     _getEventController(streamId).add(event);
   }
 
+  Future<void> _streamListen(String streamId) async {
+    if (!_listeningFor.contains(streamId)) {
+      _listeningFor.add(streamId);
+      await _sendRequest('streamListen', <String, dynamic>{'streamId': streamId});
+    }
+  }
+
   /// Reloads the VM.
-  Future<void> getVMOld() async => await vm.reload();
+  Future<void> getVM() async => await vm.reload();
 
   Future<void> refreshViews({ bool waitForViews = false }) => vm.refreshViews(waitForViews: waitForViews);
 
-  Future<void> close() async {
+  Future<void> close() async => await _peer.close();
-    _delegateService?.dispose();
-  }
-
-  // To enable a gradual migration to package:vm_service
-  @override
-  dynamic noSuchMethod(Invocation invocation) {
-    throw UnsupportedError('${invocation.memberName} is not currently supported');
-  }
 }
 
 /// An error that is thrown when constructing/updating a service object.
@@ -971,15 +936,36 @@ class VM extends ServiceObjectOwner {
     return Future<Isolate>.value(_isolateCache[isolateId]);
   }
 
+  static String _truncate(String message, int width, String ellipsis) {
+    assert(ellipsis.length < width);
+    if (message.length <= width) {
+      return message;
+    }
+    return message.substring(0, width - ellipsis.length) + ellipsis;
+  }
+
   /// Invoke the RPC and return the raw response.
   Future<Map<String, dynamic>> invokeRpcRaw(
     String method, {
     Map<String, dynamic> params = const <String, dynamic>{},
     bool truncateLogs = true,
   }) async {
-    final vm_service.Response response = await _vmService
+    globals.printTrace('Sending to VM service: $method($params)');
-      ._delegateService.callServiceExtension(method, args: params);
+    assert(params != null);
-    return response.json;
+    try {
+      final Map<String, dynamic> result = await _vmService._sendRequest(method, params);
+      final String resultString =
+          truncateLogs ? _truncate(result.toString(), 250, '...') : result.toString();
+      globals.printTrace('Result: $resultString');
+      return result;
+    } on WebSocketChannelException catch (error) {
+      throwToolExit('Error connecting to observatory: $error');
+      return null;
+    } on rpc.RpcException catch (error) {
+      globals.printError('Error ${error.code} received from application: ${error.message}');
+      globals.printTrace('${error.data}');
+      rethrow;
+    }
   }
 
   /// Invoke the RPC and return a [ServiceObject] response.
@@ -1278,17 +1264,13 @@ class Isolate extends ServiceObjectOwner {
       }
       final Map<String, dynamic> response = await invokeRpcRaw('_reloadSources', params: arguments);
       return response;
-    } on vm_service.RPCError catch (e) {
+    } on rpc.RpcException catch (e) {
       return Future<Map<String, dynamic>>.value(<String, dynamic>{
         'code': e.code,
         'message': e.message,
         'data': e.data,
       });
-    } on vm_service.SentinelException catch (e) {
-      throwToolExit('Unexpected Sentinel while reloading sources: $e');
     }
-    assert(false);
-    return null;
   }
 
   Future<Map<String, dynamic>> getObject(Map<String, dynamic> objectRef) {
@@ -1311,9 +1293,9 @@ class Isolate extends ServiceObjectOwner {
   }) async {
     try {
       return await invokeRpcRaw(method, params: params);
-    } on vm_service.RPCError catch (err) {
+    } on rpc.RpcException catch (e) {
       // If an application is not using the framework
-      if (err.code == RPCErrorCodes.kMethodNotFound) {
+      if (e.code == rpc_error_code.METHOD_NOT_FOUND) {
         return null;
       }
       rethrow;
@@ -1509,13 +1491,10 @@ class FlutterView extends ServiceObject {
     final String viewId = id;
     // When this completer completes the isolate is running.
     final Completer<void> completer = Completer<void>();
-    try {
+    final StreamSubscription<ServiceEvent> subscription =
-      await owner.vm.vmService.streamListen('Isolate');
+      (await owner.vm.vmService.onIsolateEvent).listen((ServiceEvent event) {
-    } on vm_service.RPCError {
+        // TODO(johnmccutchan): Listen to the debug stream and catch initial
-      // Do nothing, since the tool is already subscribed.
+        // launch errors.
-    }
-    final StreamSubscription<vm_service.Event> subscription =
-      owner.vm.vmService.onIsolateEvent.listen((vm_service.Event event) {
         if (event.kind == ServiceEvent.kIsolateRunnable) {
           globals.printTrace('Isolate is runnable.');
           if (!completer.isCompleted) {

packages/flutter_tools/pubspec.yaml

@@ -19,6 +19,7 @@ dependencies:
   flutter_template_images: 1.0.0
   http: 0.12.0+4
   intl: 0.16.1
+  json_rpc_2: 2.1.0
   meta: 1.1.8
   multicast_dns: 0.2.2
   mustache_template: 1.0.0+1
@@ -27,8 +28,10 @@ dependencies:
   process: 3.0.12
   quiver: 2.1.3
   stack_trace: 1.9.3
+  stream_channel: 2.0.0
   usage: 3.4.1
   webdriver: 2.1.2
+  web_socket_channel: 1.1.0
   webkit_inspection_protocol: 0.5.0+1
   xml: 3.6.1
   yaml: 2.2.0
@@ -85,7 +88,6 @@ dependencies:
   source_maps: 0.10.9 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   source_span: 1.7.0 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   sse: 3.2.1 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
-  stream_channel: 2.0.0 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   stream_transform: 1.2.0 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   string_scanner: 1.0.5 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   sync_http: 0.2.0 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
@@ -93,7 +95,6 @@ dependencies:
   typed_data: 1.1.6 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   uuid: 2.0.4 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   watcher: 0.9.7+14 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
-  web_socket_channel: 1.1.0 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
 
 dev_dependencies:
   collection: 1.14.12
@@ -111,4 +112,4 @@ dartdoc:
   # Exclude this package from the hosted API docs.
   nodoc: true
 
-# PUBSPEC CHECKSUM: a585
+# PUBSPEC CHECKSUM: 53c2

packages/flutter_tools/test/general.shard/devfs_test.dart

@@ -14,8 +14,8 @@ import 'package:flutter_tools/src/base/net.dart';
 import 'package:flutter_tools/src/compile.dart';
 import 'package:flutter_tools/src/devfs.dart';
 import 'package:flutter_tools/src/vmservice.dart';
+import 'package:json_rpc_2/json_rpc_2.dart' as rpc;
 import 'package:mockito/mockito.dart';
-import 'package:vm_service/vm_service.dart' as vm_service;
 
 import '../src/common.dart';
 import '../src/context.dart';
@@ -389,7 +389,7 @@ class MockVM implements VM {
   Future<Map<String, dynamic>> createDevFS(String fsName) async {
     _service.messages.add('create $fsName');
     if (_devFSExists) {
-      throw vm_service.RPCError('File system already exists', kFileSystemAlreadyExists, '');
+      throw rpc.RpcException(kFileSystemAlreadyExists, 'File system already exists');
     }
     _devFSExists = true;
     return <String, dynamic>{'uri': '$_baseUri'};

packages/flutter_tools/test/general.shard/fuchsia/fuchsia_device_test.dart

@@ -596,7 +596,7 @@ void main() {
           .thenAnswer((Invocation invocation) async => <int>[1]);
       when(portForwarder.forward(1))
           .thenAnswer((Invocation invocation) async => 2);
-      when(vmService.getVMOld())
+      when(vmService.getVM())
           .thenAnswer((Invocation invocation) => Future<void>.value(null));
       when(vmService.refreshViews())
           .thenAnswer((Invocation invocation) => Future<void>.value(null));

packages/flutter_tools/test/general.shard/ios/ios_device_logger_test.dart

@@ -2,18 +2,13 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-import 'dart:async';
-
 import 'package:flutter_tools/src/artifacts.dart';
 import 'package:flutter_tools/src/base/logger.dart';
 import 'package:flutter_tools/src/build_info.dart';
-import 'package:flutter_tools/src/convert.dart';
 import 'package:flutter_tools/src/device.dart';
 import 'package:flutter_tools/src/ios/devices.dart';
 import 'package:flutter_tools/src/ios/mac.dart';
-import 'package:flutter_tools/src/vmservice.dart';
 import 'package:mockito/mockito.dart';
-import 'package:vm_service/vm_service.dart';
 
 import '../../src/common.dart';
 import '../../src/context.dart';
@@ -142,39 +137,6 @@ Runner(libsystem_asl.dylib)[297] <Notice>: libMobileGestalt
       '  with a non-Flutter log message following it.',
     ]);
   });
-
-  testWithoutContext('IOSDeviceLogReader can listen to VM Service logs', () async {
-    final MockVmService vmService = MockVmService();
-    final DeviceLogReader logReader = IOSDeviceLogReader.test(
-      useSyslog: false,
-      iMobileDevice: IMobileDevice(
-        artifacts: artifacts,
-        processManager: processManager,
-        cache: fakeCache,
-        logger: logger,
-      ),
-    );
-    final StreamController<Event> controller = StreamController<Event>();
-    final Completer<Success> stdoutCompleter = Completer<Success>();
-    when(vmService.streamListen('Stdout')).thenAnswer((Invocation invocation) {
-      return stdoutCompleter.future;
-    });
-    when(vmService.onStdoutEvent).thenAnswer((Invocation invocation) {
-      return controller.stream;
-    });
-    logReader.connectedVMService = vmService;
-
-    stdoutCompleter.complete(Success());
-    controller.add(Event(
-      kind: 'Stdout',
-      timestamp: 0,
-      bytes: base64.encode(utf8.encode('  This is a message ')),
-    ));
-
-    // Wait for stream listeners to fire.
-    await expectLater(logReader.logLines, emits('  This is a message '));
-  });
 }
 
 class MockArtifacts extends Mock implements Artifacts {}
-class MockVmService extends Mock implements VMService, VmService {}

packages/flutter_tools/test/general.shard/resident_runner_test.dart

@@ -23,8 +23,8 @@ import 'package:flutter_tools/src/resident_runner.dart';
 import 'package:flutter_tools/src/run_cold.dart';
 import 'package:flutter_tools/src/run_hot.dart';
 import 'package:flutter_tools/src/vmservice.dart';
+import 'package:json_rpc_2/json_rpc_2.dart';
 import 'package:mockito/mockito.dart';
-import 'package:vm_service/vm_service.dart' as vm_service;
 
 import '../src/common.dart';
 import '../src/context.dart';
@@ -222,7 +222,7 @@ void main() {
       pathToReload: anyNamed('pathToReload'),
       invalidatedFiles: anyNamed('invalidatedFiles'),
       dillOutputPath: anyNamed('dillOutputPath'),
-    )).thenThrow(vm_service.RPCError('something bad happened', 666, ''));
+    )).thenThrow(RpcException(666, 'something bad happened'));
 
     final OperationResult result = await residentRunner.restart(fullRestart: false);
     expect(result.fatal, true);
@@ -327,7 +327,7 @@ void main() {
       pathToReload: anyNamed('pathToReload'),
       invalidatedFiles: anyNamed('invalidatedFiles'),
       dillOutputPath: anyNamed('dillOutputPath'),
-    )).thenThrow(vm_service.RPCError('something bad happened', 666, ''));
+    )).thenThrow(RpcException(666, 'something bad happened'));
 
     final OperationResult result = await residentRunner.restart(fullRestart: true);
     expect(result.fatal, true);

packages/flutter_tools/test/general.shard/vmservice_test.dart

@@ -3,218 +3,362 @@
 // found in the LICENSE file.
 
 import 'dart:async';
+import 'dart:io';
 
-import 'package:vm_service/vm_service.dart' as vm_service;
+import 'package:flutter_tools/src/base/io.dart';
-import 'package:mockito/mockito.dart';
 import 'package:flutter_tools/src/base/logger.dart';
+import 'package:flutter_tools/src/base/terminal.dart';
 import 'package:flutter_tools/src/device.dart';
+import 'package:flutter_tools/src/globals.dart' as globals;
 import 'package:flutter_tools/src/version.dart';
 import 'package:flutter_tools/src/vmservice.dart';
+import 'package:json_rpc_2/json_rpc_2.dart' as rpc;
+import 'package:mockito/mockito.dart';
+import 'package:platform/platform.dart';
+import 'package:quiver/testing/async.dart';
 
 import '../src/common.dart';
 import '../src/context.dart';
+import '../src/mocks.dart';
 
-final Map<String, Object> vm = <String, dynamic>{
+class MockPeer implements rpc.Peer {
-  'type': 'VM',
-  'name': 'vm',
-  'architectureBits': 64,
-  'targetCPU': 'x64',
-  'hostCPU': '      Intel(R) Xeon(R) CPU    E5-1650 v2 @ 3.50GHz',
-  'version': '2.1.0-dev.7.1.flutter-45f9462398 (Fri Oct 19 19:27:56 2018 +0000) on "linux_x64"',
-  '_profilerMode': 'Dart',
-  '_nativeZoneMemoryUsage': 0,
-  'pid': 103707,
-  'startTime': 1540426121876,
-  '_embedder': 'Flutter',
-  '_maxRSS': 312614912,
-  '_currentRSS': 33091584,
-  'isolates': <dynamic>[
-    <String, dynamic>{
-      'type': '@Isolate',
-      'fixedId': true,
-      'id': 'isolates/242098474',
-      'name': 'main.dart:main()',
-      'number': 242098474,
-    },
-  ],
-};
 
-final vm_service.Isolate isolate = vm_service.Isolate.parse(
+  Function _versionFn = (dynamic _) => null;
-  <String, dynamic>{
+
-    'type': 'Isolate',
+  @override
-    'fixedId': true,
+  rpc.ErrorCallback get onUnhandledError => null;
-    'id': 'isolates/242098474',
+
-    'name': 'main.dart:main()',
+  @override
-    'number': 242098474,
+  Future<dynamic> get done async {
-    '_originNumber': 242098474,
+    throw 'unexpected call to done';
-    'startTime': 1540488745340,
-    '_heaps': <String, dynamic>{
-      'new': <String, dynamic>{
-        'used': 0,
-        'capacity': 0,
-        'external': 0,
-        'collections': 0,
-        'time': 0.0,
-        'avgCollectionPeriodMillis': 0.0,
-      },
-      'old': <String, dynamic>{
-        'used': 0,
-        'capacity': 0,
-        'external': 0,
-        'collections': 0,
-        'time': 0.0,
-        'avgCollectionPeriodMillis': 0.0,
-      },
-    },
   }
-);
 
-final Map<String, Object> listViews = <String, dynamic>{
+  @override
-  'type': 'FlutterViewList',
+  bool get isClosed => _isClosed;
-  'views': <dynamic>[
+
-    <String, dynamic>{
+  @override
-      'type': 'FlutterView',
+  Future<dynamic> close() async {
-      'id': '_flutterView/0x4a4c1f8',
+    _isClosed = true;
-      'isolate': <String, dynamic>{
+  }
-        'type': '@Isolate',
+
+  @override
+  Future<dynamic> listen() async {
+    // this does get called
+  }
+
+  @override
+  void registerFallback(dynamic callback(rpc.Parameters parameters)) {
+    throw 'unexpected call to registerFallback';
+  }
+
+  @override
+  void registerMethod(String name, Function callback) {
+    registeredMethods.add(name);
+    if (name == 'flutterVersion') {
+      _versionFn = callback;
+    }
+  }
+
+  @override
+  void sendNotification(String method, [ dynamic parameters ]) {
+    // this does get called
+    sentNotifications.putIfAbsent(method, () => <dynamic>[]).add(parameters);
+  }
+
+  Map<String, List<dynamic>> sentNotifications = <String, List<dynamic>>{};
+  List<String> registeredMethods = <String>[];
+
+  bool isolatesEnabled = false;
+  bool _isClosed = false;
+
+  Future<void> _getVMLatch;
+  Completer<void> _currentGetVMLatchCompleter;
+
+  void tripGetVMLatch() {
+    final Completer<void> lastCompleter = _currentGetVMLatchCompleter;
+    _currentGetVMLatchCompleter = Completer<void>();
+    _getVMLatch = _currentGetVMLatchCompleter.future;
+    lastCompleter?.complete();
+  }
+
+  int returnedFromSendRequest = 0;
+
+  @override
+  Future<dynamic> sendRequest(String method, [ dynamic parameters ]) async {
+    if (method == 'getVM') {
+      await _getVMLatch;
+    }
+    await Future<void>.delayed(Duration.zero);
+    returnedFromSendRequest += 1;
+    if (method == 'getVM') {
+      return <String, dynamic>{
+        'type': 'VM',
+        'name': 'vm',
+        'architectureBits': 64,
+        'targetCPU': 'x64',
+        'hostCPU': '      Intel(R) Xeon(R) CPU    E5-1650 v2 @ 3.50GHz',
+        'version': '2.1.0-dev.7.1.flutter-45f9462398 (Fri Oct 19 19:27:56 2018 +0000) on "linux_x64"',
+        '_profilerMode': 'Dart',
+        '_nativeZoneMemoryUsage': 0,
+        'pid': 103707,
+        'startTime': 1540426121876,
+        '_embedder': 'Flutter',
+        '_maxRSS': 312614912,
+        '_currentRSS': 33091584,
+        'isolates': isolatesEnabled ? <dynamic>[
+          <String, dynamic>{
+            'type': '@Isolate',
+            'fixedId': true,
+            'id': 'isolates/242098474',
+            'name': 'main.dart:main()',
+            'number': 242098474,
+          },
+        ] : <dynamic>[],
+      };
+    }
+    if (method == 'getIsolate') {
+      return <String, dynamic>{
+        'type': 'Isolate',
         'fixedId': true,
         'id': 'isolates/242098474',
         'name': 'main.dart:main()',
         'number': 242098474,
-      },
+        '_originNumber': 242098474,
-    },
+        'startTime': 1540488745340,
-  ]
+        '_heaps': <String, dynamic>{
-};
+          'new': <String, dynamic>{
+            'used': 0,
+            'capacity': 0,
+            'external': 0,
+            'collections': 0,
+            'time': 0.0,
+            'avgCollectionPeriodMillis': 0.0,
+          },
+          'old': <String, dynamic>{
+            'used': 0,
+            'capacity': 0,
+            'external': 0,
+            'collections': 0,
+            'time': 0.0,
+            'avgCollectionPeriodMillis': 0.0,
+          },
+        },
+      };
+    }
+    if (method == '_flutter.listViews') {
+      return <String, dynamic>{
+        'type': 'FlutterViewList',
+        'views': isolatesEnabled ? <dynamic>[
+          <String, dynamic>{
+            'type': 'FlutterView',
+            'id': '_flutterView/0x4a4c1f8',
+            'isolate': <String, dynamic>{
+              'type': '@Isolate',
+              'fixedId': true,
+              'id': 'isolates/242098474',
+              'name': 'main.dart:main()',
+              'number': 242098474,
+            },
+          },
+        ] : <dynamic>[],
+      };
+    }
+    if (method == 'flutterVersion') {
+      return _versionFn(parameters);
+    }
+    return null;
+  }
 
-typedef ServiceCallback = Future<Map<String, dynamic>> Function(Map<String, Object>);
+  @override
+  dynamic withBatch(dynamic callback()) {
+    throw 'unexpected call to withBatch';
+  }
+}
 
 void main() {
-  testUsingContext('VMService can refreshViews', () async {
+  MockStdio mockStdio;
-    final MockVMService mockVmService = MockVMService();
+  final MockFlutterVersion mockVersion = MockFlutterVersion();
-    final VMService vmService = VMService(
+  group('VMService', () {
-      null,
-      null,
-      null,
-      null,
-      null,
-      null,
-      null,
-      mockVmService,
-      Completer<void>(),
-      const Stream<dynamic>.empty(),
-    );
 
-    verify(mockVmService.registerService('flutterVersion', 'Flutter Tools')).called(1);
+    setUp(() {
-
+      mockStdio = MockStdio();
-    when(mockVmService.callServiceExtension('getVM',
-      args: anyNamed('args'), // Empty
-      isolateId: null
-    )).thenAnswer((Invocation invocation) async {
-      return vm_service.Response.parse(vm);
     });
-    await vmService.getVMOld();
 
-
+    testUsingContext('fails connection eagerly in the connect() method', () async {
-    when(mockVmService.callServiceExtension('_flutter.listViews',
+      FakeAsync().run((FakeAsync time) {
-      args: anyNamed('args'),
+        bool failed = false;
-      isolateId: anyNamed('isolateId')
+        final Future<VMService> future = VMService.connect(Uri.parse('http://host.invalid:9999/'));
-    )).thenAnswer((Invocation invocation) async {
+        future.whenComplete(() {
-      return vm_service.Response.parse(listViews);
+          failed = true;
+        });
+        time.elapse(const Duration(seconds: 5));
+        expect(failed, isFalse);
+        expect(mockStdio.writtenToStdout.join(''), '');
+        expect(mockStdio.writtenToStderr.join(''), '');
+        time.elapse(const Duration(seconds: 5));
+        expect(failed, isFalse);
+        expect(mockStdio.writtenToStdout.join(''), 'This is taking longer than expected...\n');
+        expect(mockStdio.writtenToStderr.join(''), '');
+      });
+    }, overrides: <Type, Generator>{
+      Logger: () => StdoutLogger(
+        outputPreferences: OutputPreferences.test(),
+        stdio: mockStdio,
+        terminal: AnsiTerminal(
+          stdio: mockStdio,
+          platform: const LocalPlatform(),
+        ),
+        timeoutConfiguration: const TimeoutConfiguration(),
+      ),
+      WebSocketConnector: () => (String url, {CompressionOptions compression}) async => throw const SocketException('test'),
     });
-    await vmService.refreshViews(waitForViews: true);
 
-    expect(vmService.vm.name, 'vm');
+    testUsingContext('closing VMService closes Peer', () async {
-    expect(vmService.vm.views.single.id, '_flutterView/0x4a4c1f8');
+      final MockPeer mockPeer = MockPeer();
-  }, overrides: <Type, Generator>{
+      final VMService vmService = VMService(mockPeer, null, null, null, null, null, MockDevice(), null);
-    Logger: () => BufferLogger.test()
+      expect(mockPeer.isClosed, equals(false));
-  });
+      await vmService.close();
+      expect(mockPeer.isClosed, equals(true));
+    });
 
-  testUsingContext('VmService registers reloadSources', () {
+    testUsingContext('refreshViews', () {
-    Future<void> reloadSources(String isolateId, { bool pause, bool force}) async {}
+      FakeAsync().run((FakeAsync time) {
-    final MockVMService mockVMService = MockVMService();
+        bool done = false;
-    VMService(
+        final MockPeer mockPeer = MockPeer();
-      null,
+        expect(mockPeer.returnedFromSendRequest, 0);
-      null,
+        final VMService vmService = VMService(mockPeer, null, null, null, null, null, null, null);
-      reloadSources,
+        expect(mockPeer.sentNotifications, contains('registerService'));
-      null,
+        final List<String> registeredServices =
-      null,
+          mockPeer.sentNotifications['registerService']
-      null,
+            .map((dynamic service) => (service as Map<String, String>)['service'])
-      null,
+            .toList();
-      mockVMService,
+        expect(registeredServices, contains('flutterVersion'));
-      Completer<void>(),
+        vmService.getVM().then((void value) { done = true; });
-      const Stream<dynamic>.empty(),
+        expect(done, isFalse);
-    );
+        expect(mockPeer.returnedFromSendRequest, 0);
+        time.elapse(Duration.zero);
+        expect(done, isTrue);
+        expect(mockPeer.returnedFromSendRequest, 1);
 
-    verify(mockVMService.registerService('reloadSources', 'Flutter Tools')).called(1);
+        done = false;
-  }, overrides: <Type, Generator>{
+        mockPeer.tripGetVMLatch(); // this blocks the upcoming getVM call
-    Logger: () => BufferLogger.test()
+        final Future<void> ready = vmService.refreshViews(waitForViews: true);
-  });
+        ready.then((void value) { done = true; });
+        expect(mockPeer.returnedFromSendRequest, 1);
+        time.elapse(Duration.zero); // this unblocks the listViews call which returns nothing
+        expect(mockPeer.returnedFromSendRequest, 2);
+        time.elapse(const Duration(milliseconds: 50)); // the last listViews had no views, so it waits 50ms, then calls getVM
+        expect(done, isFalse);
+        expect(mockPeer.returnedFromSendRequest, 2);
+        mockPeer.tripGetVMLatch(); // this unblocks the getVM call
+        expect(mockPeer.returnedFromSendRequest, 2);
+        time.elapse(Duration.zero); // here getVM returns with no isolates and listViews returns no views
+        expect(mockPeer.returnedFromSendRequest, 4);
+        time.elapse(const Duration(milliseconds: 50)); // so refreshViews waits another 50ms
+        expect(done, isFalse);
+        expect(mockPeer.returnedFromSendRequest, 4);
+        mockPeer.tripGetVMLatch(); // this unblocks the getVM call
+        expect(mockPeer.returnedFromSendRequest, 4);
+        time.elapse(Duration.zero); // here getVM returns with no isolates and listViews returns no views
+        expect(mockPeer.returnedFromSendRequest, 6);
+        time.elapse(const Duration(milliseconds: 50)); // so refreshViews waits another 50ms
+        expect(done, isFalse);
+        expect(mockPeer.returnedFromSendRequest, 6);
+        mockPeer.tripGetVMLatch(); // this unblocks the getVM call
+        expect(mockPeer.returnedFromSendRequest, 6);
+        time.elapse(Duration.zero); // here getVM returns with no isolates and listViews returns no views
+        expect(mockPeer.returnedFromSendRequest, 8);
+        time.elapse(const Duration(milliseconds: 50)); // so refreshViews waits another 50ms
+        expect(done, isFalse);
+        expect(mockPeer.returnedFromSendRequest, 8);
+        mockPeer.tripGetVMLatch(); // this unblocks the getVM call
+        expect(mockPeer.returnedFromSendRequest, 8);
+        time.elapse(Duration.zero); // here getVM returns with no isolates and listViews returns no views
+        expect(mockPeer.returnedFromSendRequest, 10);
+        const String message = 'Flutter is taking longer than expected to report its views. Still trying...\n';
+        expect(mockStdio.writtenToStdout.join(''), message);
+        expect(mockStdio.writtenToStderr.join(''), '');
+        time.elapse(const Duration(milliseconds: 50)); // so refreshViews waits another 50ms
+        expect(done, isFalse);
+        expect(mockPeer.returnedFromSendRequest, 10);
+        mockPeer.isolatesEnabled = true;
+        mockPeer.tripGetVMLatch(); // this unblocks the getVM call
+        expect(mockPeer.returnedFromSendRequest, 10);
+        time.elapse(Duration.zero); // now it returns an isolate and the listViews call returns views
+        expect(mockPeer.returnedFromSendRequest, 13);
+        expect(done, isTrue);
+        expect(mockStdio.writtenToStdout.join(''), message);
+        expect(mockStdio.writtenToStderr.join(''), '');
+      });
+    }, overrides: <Type, Generator>{
+      Logger: () => StdoutLogger(
+        outputPreferences: globals.outputPreferences,
+        terminal: AnsiTerminal(
+          stdio: mockStdio,
+          platform: const LocalPlatform(),
+        ),
+        stdio: mockStdio,
+        timeoutConfiguration: const TimeoutConfiguration(),
+      ),
+    });
 
-  testUsingContext('VmService registers reloadMethod', () {
+    testUsingContext('registers hot UI method', () {
-    Future<void> reloadMethod({  String classId, String libraryId,}) async {}
+      FakeAsync().run((FakeAsync time) {
-    final MockVMService mockVMService = MockVMService();
+        final MockPeer mockPeer = MockPeer();
-    VMService(
+        Future<void> reloadMethod({ String classId, String libraryId }) async {}
-      null,
+        VMService(mockPeer, null, null, null, null, null, null, reloadMethod);
-      null,
-      null,
-      null,
-      null,
-      null,
-      reloadMethod,
-      mockVMService,
-      Completer<void>(),
-      const Stream<dynamic>.empty(),
-    );
 
-    verify(mockVMService.registerService('reloadMethod', 'Flutter Tools')).called(1);
+        expect(mockPeer.registeredMethods, contains('reloadMethod'));
-  }, overrides: <Type, Generator>{
+      });
-    Logger: () => BufferLogger.test()
+    }, overrides: <Type, Generator>{
-  });
+      Logger: () => StdoutLogger(
+        outputPreferences: globals.outputPreferences,
+        terminal: AnsiTerminal(
+          stdio: mockStdio,
+          platform: const LocalPlatform(),
+        ),
+        stdio: mockStdio,
+        timeoutConfiguration: const TimeoutConfiguration(),
+      ),
+    });
 
-  testUsingContext('VmService registers flutterMemoryInfo service', () {
+    testUsingContext('registers flutterMemoryInfo service', () {
-    final MockDevice mockDevice = MockDevice();
+      FakeAsync().run((FakeAsync time) {
-    final MockVMService mockVMService = MockVMService();
+        final MockDevice mockDevice = MockDevice();
-    VMService(
+        final MockPeer mockPeer = MockPeer();
-      null,
+        Future<void> reloadSources(String isolateId, { bool pause, bool force}) async {}
-      null,
+        VMService(mockPeer, null, null, reloadSources, null, null, mockDevice, null);
-      null,
-      null,
-      null,
-      mockDevice,
-      null,
-      mockVMService,
-      Completer<void>(),
-      const Stream<dynamic>.empty(),
-    );
 
-    verify(mockVMService.registerService('flutterMemoryInfo', 'Flutter Tools')).called(1);
+        expect(mockPeer.registeredMethods, contains('flutterMemoryInfo'));
-  }, overrides: <Type, Generator>{
+      });
-    Logger: () => BufferLogger.test()
+    }, overrides: <Type, Generator>{
-  });
+      Logger: () => StdoutLogger(
+        outputPreferences: globals.outputPreferences,
+        terminal: AnsiTerminal(
+          stdio: mockStdio,
+          platform: const LocalPlatform(),
+        ),
+        stdio: mockStdio,
+        timeoutConfiguration: const TimeoutConfiguration(),
+      ),
+    });
 
-  testUsingContext('VMService returns correct FlutterVersion', () async {
+    testUsingContext('returns correct FlutterVersion', () {
-    final MockVMService mockVMService = MockVMService();
+      FakeAsync().run((FakeAsync time) async {
-    VMService(
+        final MockPeer mockPeer = MockPeer();
-      null,
+        VMService(mockPeer, null, null, null, null, null, MockDevice(), null);
-      null,
-      null,
-      null,
-      null,
-      null,
-      null,
-      mockVMService,
-      Completer<void>(),
-      const Stream<dynamic>.empty(),
-    );
 
-    verify(mockVMService.registerService('flutterVersion', 'Flutter Tools')).called(1);
+        expect(mockPeer.registeredMethods, contains('flutterVersion'));
-  }, overrides: <Type, Generator>{
+        expect(await mockPeer.sendRequest('flutterVersion'), equals(mockVersion.toJson()));
-    FlutterVersion: () => MockFlutterVersion(),
+      });
+    }, overrides: <Type, Generator>{
+      FlutterVersion: () => mockVersion,
+    });
   });
 }
 
 class MockDevice extends Mock implements Device {}
-class MockVMService extends Mock implements vm_service.VmService {}
+
 class MockFlutterVersion extends Mock implements FlutterVersion {
   @override
   Map<String, Object> toJson() => const <String, Object>{'Mock': 'Version'};