[devicelab] remove vm_service_client from devicelab (#78559)

dev/devicelab/bin/tasks/service_extensions_test.dart

@@ -11,7 +11,8 @@ import 'package:flutter_devicelab/framework/framework.dart';
 import 'package:flutter_devicelab/framework/task_result.dart';
 import 'package:flutter_devicelab/framework/utils.dart';
 import 'package:path/path.dart' as path;
-import 'package:vm_service_client/vm_service_client.dart';
+import 'package:vm_service/vm_service.dart';
+import 'package:vm_service/vm_service_io.dart';
 
 void main() {
   task(() async {
@@ -54,52 +55,58 @@ void main() {
       if (!ok)
         throw 'Failed to run test app.';
 
-      final VMServiceClient client = VMServiceClient.connect('ws://localhost:$vmServicePort/ws');
+      final VmService client = await vmServiceConnectUri('ws://localhost:$vmServicePort/ws');
       final VM vm = await client.getVM();
-      final VMIsolateRef isolate = vm.isolates.first;
+      final IsolateRef isolate = vm.isolates.first;
 
-      final StreamController<VMExtensionEvent> frameEventsController = StreamController<VMExtensionEvent>();
-      final StreamController<VMExtensionEvent> navigationEventsController = StreamController<VMExtensionEvent>();
-      isolate.onExtensionEvent.listen((VMExtensionEvent event) {
-        if (event.kind == 'Flutter.Frame') {
+      final StreamController<Event> frameEventsController = StreamController<Event>();
+      final StreamController<Event> navigationEventsController = StreamController<Event>();
+      try {
+        await client.streamListen(EventKind.kExtension);
+      } catch (err) {
+        // Do nothing on errors.
+      }
+      client.onExtensionEvent.listen((Event event) {
+        if (event.extensionKind == 'Flutter.Frame') {
           frameEventsController.add(event);
-        } else if (event.kind == 'Flutter.Navigation') {
+        } else if (event.extensionKind == 'Flutter.Navigation') {
           navigationEventsController.add(event);
         }
       });
 
-      final Stream<VMExtensionEvent> frameEvents = frameEventsController.stream;
-      final Stream<VMExtensionEvent> navigationEvents = navigationEventsController.stream;
+      final Stream<Event> frameEvents = frameEventsController.stream;
+      final Stream<Event> navigationEvents = navigationEventsController.stream;
 
       print('reassembling app...');
-      final Future<VMExtensionEvent> frameFuture = frameEvents.first;
-      await isolate.invokeExtension('ext.flutter.reassemble');
+      final Future<Event> frameFuture = frameEvents.first;
+      await client.callServiceExtension('ext.flutter.reassemble', isolateId: isolate.id);
 
       // ensure we get an event
-      final VMExtensionEvent event = await frameFuture;
+      final Event event = await frameFuture;
       print('${event.kind}: ${event.data}');
 
       // validate the fields
       // {number: 8, startTime: 0, elapsed: 1437, build: 600, raster: 800}
-      expect(event.data['number'] is int);
-      expect((event.data['number'] as int) >= 0);
-      expect(event.data['startTime'] is int);
-      expect((event.data['startTime'] as int) >= 0);
-      expect(event.data['elapsed'] is int);
-      expect((event.data['elapsed'] as int) >= 0);
-      expect(event.data['build'] is int);
-      expect((event.data['build'] as int) >= 0);
-      expect(event.data['raster'] is int);
-      expect((event.data['raster'] as int) >= 0);
+      print(event.extensionData.data);
+      expect(event.extensionData.data['number'] is int);
+      expect((event.extensionData.data['number'] as int) >= 0);
+      expect(event.extensionData.data['startTime'] is int);
+      expect((event.extensionData.data['startTime'] as int) >= 0);
+      expect(event.extensionData.data['elapsed'] is int);
+      expect((event.extensionData.data['elapsed'] as int) >= 0);
+      expect(event.extensionData.data['build'] is int);
+      expect((event.extensionData.data['build'] as int) >= 0);
+      expect(event.extensionData.data['raster'] is int);
+      expect((event.extensionData.data['raster'] as int) >= 0);
 
-      final Future<VMExtensionEvent> navigationFuture = navigationEvents.first;
+      final Future<Event> navigationFuture = navigationEvents.first;
       // This tap triggers a navigation event.
       device.tap(100, 200);
 
-      final VMExtensionEvent navigationEvent = await navigationFuture;
+      final Event navigationEvent = await navigationFuture;
       // validate the fields
-      expect(navigationEvent.data['route'] is Map<dynamic, dynamic>);
-      final Map<dynamic, dynamic> route = navigationEvent.data['route'] as Map<dynamic, dynamic>;
+      expect(navigationEvent.extensionData.data['route'] is Map<dynamic, dynamic>);
+      final Map<dynamic, dynamic> route = navigationEvent.extensionData.data['route'] as Map<dynamic, dynamic>;
       expect(route['description'] is String);
       expect(route['settings'] is Map<dynamic, dynamic>);
       final Map<dynamic, dynamic> settings = route['settings'] as Map<dynamic, dynamic>;

dev/devicelab/lib/framework/runner.dart

@@ -7,7 +7,7 @@ import 'dart:convert';
 import 'dart:io';
 
 import 'package:meta/meta.dart';
-import 'package:vm_service_client/vm_service_client.dart';
+import 'package:vm_service/vm_service.dart';
 
 import 'adb.dart';
 import 'cocoon.dart';
@@ -131,8 +131,12 @@ Future<TaskResult> runTask(
   });
 
   try {
-    final VMIsolateRef isolate = await _connectToRunnerIsolate(await uri.future);
-    final Map<String, dynamic> taskResultJson = await isolate.invokeExtension('ext.cocoonRunTask', isolateParams) as Map<String, dynamic>;
+    final ConnectionResult result = await _connectToRunnerIsolate(await uri.future);
+    final Map<String, dynamic> taskResultJson = (await result.vmService.callServiceExtension(
+      'ext.cocoonRunTask',
+      args: isolateParams,
+      isolateId: result.isolate.id,
+    )).json;
     final TaskResult taskResult = TaskResult.fromJson(taskResultJson);
     await runner.exitCode;
     return taskResult;
@@ -144,14 +148,13 @@ Future<TaskResult> runTask(
   }
 }
 
-Future<VMIsolateRef> _connectToRunnerIsolate(Uri vmServiceUri) async {
+Future<ConnectionResult> _connectToRunnerIsolate(Uri vmServiceUri) async {
   final List<String> pathSegments = <String>[
     // Add authentication code.
     if (vmServiceUri.pathSegments.isNotEmpty) vmServiceUri.pathSegments[0],
     'ws',
   ];
-  final String url = vmServiceUri.replace(scheme: 'ws', pathSegments:
-      pathSegments).toString();
+  final String url = vmServiceUri.replace(scheme: 'ws', pathSegments: pathSegments).toString();
   final Stopwatch stopwatch = Stopwatch()..start();
 
   while (true) {
@@ -160,13 +163,17 @@ Future<VMIsolateRef> _connectToRunnerIsolate(Uri vmServiceUri) async {
       await (await WebSocket.connect(url)).close();
 
       // Look up the isolate.
-      final VMServiceClient client = VMServiceClient.connect(url);
-      final VM vm = await client.getVM();
-      final VMIsolateRef isolate = vm.isolates.single;
-      final String response = await isolate.invokeExtension('ext.cocoonRunnerReady') as String;
-      if (response != 'ready')
+      final VmService client = await vmServiceConnectUri(url);
+      VM vm = await client.getVM();
+      while (vm.isolates.isEmpty) {
+        await Future<void>.delayed(const Duration(seconds: 1));
+        vm = await client.getVM();
+      }
+      final IsolateRef isolate = vm.isolates.first;
+      final Response response = await client.callServiceExtension('ext.cocoonRunnerReady', isolateId: isolate.id);
+      if (response.json['response'] != 'ready')
         throw 'not ready yet';
-      return isolate;
+      return ConnectionResult(client, isolate);
     } catch (error) {
       if (stopwatch.elapsed > const Duration(seconds: 10))
         print('VM service still not ready after ${stopwatch.elapsed}: $error\nContinuing to retry...');
@@ -174,3 +181,38 @@ Future<VMIsolateRef> _connectToRunnerIsolate(Uri vmServiceUri) async {
     }
   }
 }
+
+class ConnectionResult {
+  ConnectionResult(this.vmService, this.isolate);
+
+  final VmService vmService;
+  final IsolateRef isolate;
+}
+
+/// The cocoon client sends an invalid VM service response, we need to intercept it.
+Future<VmService> vmServiceConnectUri(String wsUri, {Log log}) async {
+  final WebSocket socket = await WebSocket.connect(wsUri);
+  final StreamController<dynamic> controller = StreamController<dynamic>();
+  final Completer<dynamic> streamClosedCompleter = Completer<dynamic>();
+  socket.listen(
+    (dynamic data) {
+      final Map<String, dynamic> rawData = json.decode(data as String) as Map<String, dynamic> ;
+      if (rawData['result'] == 'ready') {
+        rawData['result'] = <String, dynamic>{'response': 'ready'};
+        controller.add(json.encode(rawData));
+      } else {
+        controller.add(data);
+      }
+    },
+    onError: (dynamic err, StackTrace stackTrace) => controller.addError(err, stackTrace),
+    onDone: () => streamClosedCompleter.complete(),
+  );
+
+  return VmService(
+    controller.stream,
+    (String message) => socket.add(message),
+    log: log,
+    disposeHandler: () => socket.close(),
+    streamClosed: streamClosedCompleter.future,
+  );
+}