Fix shell launcher test (dart2 compliant and stop polling) (#5009)

* Make test Dart2 compliant

* Use service and debug events instead of polling to wait for isolate to start, run and resume.

* Refactor into _onServiceEvent. Wait for 'paused' event instead of 'isolate runnable'.

shell/testing/observatory/empty_main.dart

@@ -2,5 +2,5 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-main() {
+void main() {
 }

shell/testing/observatory/launcher.dart

@@ -9,24 +9,23 @@ import 'dart:convert';
 import 'dart:io';
 
 class ShellProcess {
-  final Completer _observatoryUriCompleter = new Completer();
+  final Completer<Uri> _observatoryUriCompleter = new Completer<Uri>();
   final Process _process;
 
-  ShellProcess(this._process) {
-    assert(_process != null);
+  ShellProcess(this._process) : assert(_process != null) {
     // Scan stdout and scrape the Observatory Uri.
-    _process.stdout.transform(UTF8.decoder)
-                   .transform(new LineSplitter()).listen((line) {
+    _process.stdout.transform(utf8.decoder)
+                   .transform(const LineSplitter()).listen((String line) {
       const String observatoryUriPrefix = 'Observatory listening on ';
       if (line.startsWith(observatoryUriPrefix)) {
         print(line);
-        Uri uri = Uri.parse(line.substring(observatoryUriPrefix.length));
+        final Uri uri = Uri.parse(line.substring(observatoryUriPrefix.length));
         _observatoryUriCompleter.complete(uri);
       }
     });
   }
 
-  Future kill() async {
+  Future<bool> kill() async {
     if (_process == null) {
       return false;
     }
@@ -39,7 +38,7 @@ class ShellProcess {
 }
 
 class ShellLauncher {
-  final List<String> args = [
+  final List<String> args = <String>[
     '--observatory-port=0',
     '--non-interactive',
     '--run-forever',
@@ -60,13 +59,13 @@ class ShellLauncher {
 
   Future<ShellProcess> launch() async {
     try {
-      List<String> shellArguments = [];
+      final List<String> shellArguments = <String>[];
       if (startPaused) {
         shellArguments.add('--start-paused');
       }
       shellArguments.addAll(args);
       print('Launching $shellExecutablePath $shellArguments');
-      var process = await Process.start(shellExecutablePath, shellArguments);
+      final Process process = await Process.start(shellExecutablePath, shellArguments);
       return new ShellProcess(process);
     } catch (e) {
       print('Error launching shell: $e');

shell/testing/observatory/service_client.dart

@@ -4,60 +4,100 @@
 
 library observatory_sky_shell_service_client;
 
-
 import 'dart:async';
 import 'dart:convert';
+import 'dart:io';
+
 
 class ServiceClient {
-  ServiceClient(this.client) {
+  Completer<dynamic> isolateStartedId;
+  Completer<dynamic> isolatePausedId;
+  Completer<dynamic> isolateResumeId;
+
+  ServiceClient(this.client, {this.isolateStartedId, this.isolatePausedId,
+      this.isolateResumeId}) {
     client.listen(_onData,
                   onError: _onError,
                   cancelOnError: true);
   }
 
-  Future<Map> invokeRPC(String method, [Map params]) async {
-    var key = _createKey();
-    var request = JSON.encode({
+  Future<Map<String, dynamic>> invokeRPC(String method, [Map<String, dynamic> params]) async {
+    final String key = _createKey();
+    final String request = json.encode(<String, dynamic>{
       'jsonrpc': '2.0',
       'method': method,
-      'params': params == null ? {} : params,
+      'params': params == null ? <String, dynamic>{} : params,
       'id': key,
     });
     client.add(request);
-    var completer = new Completer();
-    _outstanding_requests[key] = completer;
+    final Completer<Map<String, dynamic>> completer = new Completer<Map<String, dynamic>>();
+    _outstandingRequests[key] = completer;
     print('-> $key ($method)');
     return completer.future;
   }
 
   String _createKey() {
-    var key = '$_id';
+    final String key = '$_id';
     _id++;
     return key;
   }
 
-  void _onData(String message) {
-    var response = JSON.decode(message);
-    var key = response['id'];
-    print('<- $key');
-    var completer = _outstanding_requests.remove(key);
-    assert(completer != null);
-    var result = response['result'];
-    var error = response['error'];
-    if (error != null) {
-      assert(result == null);
-      completer.completeError(error);
+  void _onData(dynamic message) {
+    final Map<String, dynamic> response = json.decode(message);
+    final dynamic key = response['id'];
+    if (key != null) {
+      print('<- $key');
+      final dynamic completer = _outstandingRequests.remove(key);
+      assert(completer != null);
+      final dynamic result = response['result'];
+      final dynamic error = response['error'];
+      if (error != null) {
+        assert(result == null);
+        completer.completeError(error);
+      } else {
+        assert(result != null);
+        completer.complete(result);
+      }
     } else {
-      assert(result != null);
-      completer.complete(result);
+      if (response['method'] == 'streamNotify') {
+        _onServiceEvent(response['params']);
+      }
+    }
+  }
+
+  void _onServiceEvent(Map<String, dynamic> params) {
+    if (params == null) {
+      return;
+    }
+    final Map<String, dynamic> event = params['event'];
+    if (event == null || event['type'] != 'Event') {
+      return;
+    }
+    final dynamic isolateId = event['isolate']['id'];
+    switch (params['streamId']) {
+      case 'Isolate':
+        if (event['kind'] == 'IsolateStarted') {
+          isolateStartedId?.complete(isolateId);
+        }
+        break;
+      case 'Debug':
+        switch (event['kind']) {
+          case 'Resume':
+            isolateResumeId?.complete(isolateId);
+            break;
+          case 'PauseStart':
+            isolatePausedId?.complete(isolateId);
+            break;
+        }
+        break;
     }
   }
 
-  void _onError(error) {
+  void _onError(dynamic error) {
     print('WebSocket error: $error');
   }
 
   final WebSocket client;
-  final Map<String, Completer> _outstanding_requests = <String, Completer>{};
-  var _id = 1;
+  final Map<String, Completer<dynamic>> _outstandingRequests = <String, Completer<dynamic>>{};
+  int _id = 1;
 }

shell/testing/observatory/test.dart

@@ -12,35 +12,35 @@ import 'launcher.dart';
 import 'service_client.dart';
 
 class Expect {
-  static equals(dynamic actual, dynamic expected) {
+  static void equals(dynamic actual, dynamic expected) {
     if (actual != expected) {
       throw 'Expected $actual == $expected';
     }
   }
 
-  static contains(String needle, String haystack) {
+  static void contains(String needle, String haystack) {
     if (!haystack.contains(needle)) {
       throw 'Expected $haystack to contain $needle';
     }
   }
 
-  static isTrue(bool tf) {
+  static void isTrue(bool tf) {
     if (tf != true) {
-      throw 'Expected $a to be true';
+      throw 'Expected $tf to be true';
     }
   }
 
-  static isFalse(bool tf) {
+  static void isFalse(bool tf) {
     if (tf != false) {
-      throw 'Expected $a to be false';
+      throw 'Expected $tf to be false';
     }
   }
 
-  static notExecuted() {
+  static void notExecuted() {
     throw 'Should not have hit';
   }
 
-  static isNotNull(dynamic a) {
+  static void isNotNull(dynamic a) {
     if (a == null) {
       throw 'Expected $a to not be null';
     }
@@ -48,32 +48,32 @@ class Expect {
 }
 
 Future<String> readResponse(HttpClientResponse response) {
-  var completer = new Completer();
-  var contents = new StringBuffer();
-  response.transform(UTF8.decoder).listen((String data) {
+  final Completer<String> completer = new Completer<String>();
+  final StringBuffer contents = new StringBuffer();
+  response.transform(utf8.decoder).listen((String data) {
     contents.write(data);
   }, onDone: () => completer.complete(contents.toString()));
   return completer.future;
 }
 
 // Test accessing the service protocol over http.
-Future testHttpProtocolRequest(Uri uri) async {
+Future<Null> testHttpProtocolRequest(Uri uri) async {
   uri = uri.replace(path: 'getVM');
-  HttpClient client = new HttpClient();
-  HttpClientRequest request = await client.getUrl(uri);
-  HttpClientResponse response = await request.close();
+  final HttpClient client = new HttpClient();
+  final HttpClientRequest request = await client.getUrl(uri);
+  final HttpClientResponse response = await request.close();
   Expect.equals(response.statusCode, 200);
-  Map responseAsMap = JSON.decode(await readResponse(response));
-  Expect.equals(responseAsMap['jsonrpc'], "2.0");
+  final Map<String, dynamic> responseAsMap = json.decode(await readResponse(response));
+  Expect.equals(responseAsMap['jsonrpc'], '2.0');
   client.close();
 }
 
 // Test accessing the service protocol over ws.
-Future testWebSocketProtocolRequest(Uri uri) async {
+Future<Null> testWebSocketProtocolRequest(Uri uri) async {
   uri = uri.replace(scheme: 'ws', path: 'ws');
-  WebSocket webSocketClient = await WebSocket.connect(uri.toString());
-  ServiceClient serviceClient = new ServiceClient(webSocketClient);
-  Map response = await serviceClient.invokeRPC('getVM');
+  final WebSocket webSocketClient = await WebSocket.connect(uri.toString());
+  final ServiceClient serviceClient = new ServiceClient(webSocketClient);
+  final Map<String, dynamic> response = await serviceClient.invokeRPC('getVM');
   Expect.equals(response['type'], 'VM');
   try {
     await serviceClient.invokeRPC('BART_SIMPSON');
@@ -85,83 +85,85 @@ Future testWebSocketProtocolRequest(Uri uri) async {
 }
 
 // Test accessing an Observatory UI asset.
-Future testHttpAssetRequest(Uri uri) async {
+Future<Null> testHttpAssetRequest(Uri uri) async {
   uri = uri.replace(path: 'third_party/trace_viewer_full.html');
-  HttpClient client = new HttpClient();
-  HttpClientRequest request = await client.getUrl(uri);
-  HttpClientResponse response = await request.close();
+  final HttpClient client = new HttpClient();
+  final HttpClientRequest request = await client.getUrl(uri);
+  final HttpClientResponse response = await request.close();
   Expect.equals(response.statusCode, 200);
-  await response.drain();
+  await response.drain<List<int>>();
   client.close();
 }
 
-Future testStartPaused(Uri uri) async {
+Future<Null> testStartPaused(Uri uri) async {
   uri = uri.replace(scheme: 'ws', path: 'ws');
-  WebSocket webSocketClient = await WebSocket.connect(uri.toString());
-  ServiceClient serviceClient = new ServiceClient(webSocketClient);
-
-  // Wait until we have the isolateId.
+  final WebSocket webSocketClient = await WebSocket.connect(uri.toString());
+  final Completer<dynamic> isolateStartedId = new Completer<dynamic>();
+  final Completer<dynamic> isolatePausedId = new Completer<dynamic>();
+  final Completer<dynamic> isolateResumeId = new Completer<dynamic>();
+  final ServiceClient serviceClient = new ServiceClient(webSocketClient,
+      isolateStartedId: isolateStartedId,
+      isolatePausedId: isolatePausedId,
+      isolateResumeId: isolateResumeId);
+  await serviceClient.invokeRPC('streamListen', <String, String>{ 'streamId': 'Isolate'});
+  await serviceClient.invokeRPC('streamListen', <String, String>{ 'streamId': 'Debug'});
+
+  final Map<String, dynamic> response = await serviceClient.invokeRPC('getVM');
+  Expect.equals(response['type'], 'VM');
   String isolateId;
-  while (isolateId == null) {
-    Map response = await serviceClient.invokeRPC('getVM');
-    Expect.equals(response['type'], 'VM');
-    if (response['isolates'].length > 0) {
-      isolateId = response['isolates'][0]['id'];
-    }
+  if (response['isolates'].length > 0) {
+    isolateId = response['isolates'][0]['id'];
+  } else {
+    // Wait until isolate starts.
+    isolateId = await isolateStartedId.future;
   }
 
   // Grab the isolate.
-  Map isolate;
-  while(true) {
-    isolate = await serviceClient.invokeRPC('getIsolate', {
+  Map<String, dynamic> isolate = await serviceClient.invokeRPC('getIsolate', <String, String>{
+      'isolateId': isolateId,
+  });
+  Expect.equals(isolate['type'], 'Isolate');
+  Expect.isNotNull(isolate['pauseEvent']);
+  // If it is not runnable, wait until it becomes runnable.
+  if (isolate['pauseEvent']['kind'] == 'None') {
+    await isolatePausedId.future;
+    isolate = await serviceClient.invokeRPC('getIsolate', <String, String>{
       'isolateId': isolateId,
     });
-    Expect.equals(isolate['type'], 'Isolate');
-    // Verify that it is paused at start.
-    Expect.isNotNull(isolate['pauseEvent']);
-    if (isolate['pauseEvent']['kind'] != 'None') {
-      break;
-    }
-  }
-
-  Expect.isNotNull(isolate);
+   }
+  // Verify that it is paused at start.
   Expect.equals(isolate['pauseEvent']['kind'], 'PauseStart');
 
   // Resume the isolate.
-  await serviceClient.invokeRPC('resume', {
+  await serviceClient.invokeRPC('resume', <String, String>{
     'isolateId': isolateId,
   });
-
   // Wait until the isolate has resumed.
-  while (true) {
-    Map response = await serviceClient.invokeRPC('getIsolate', {
-      'isolateId': isolateId,
-    });
-    Expect.equals(response['type'], 'Isolate');
-    Expect.isNotNull(response['pauseEvent']);
-    if (response['pauseEvent']['kind'] == 'Resume') {
-      break;
-    }
-  }
+  await isolateResumeId.future;
+  final Map<String, dynamic> resumedResponse = await serviceClient.invokeRPC(
+      'getIsolate', <String, String>{'isolateId': isolateId});
+  Expect.equals(resumedResponse['type'], 'Isolate');
+  Expect.isNotNull(resumedResponse['pauseEvent']);
+  Expect.equals(resumedResponse['pauseEvent']['kind'], 'Resume');
 }
 
-typedef Future TestFunction(Uri uri);
+typedef Future<Null> TestFunction(Uri uri);
 
-final List<TestFunction> basicTests = [
+final List<TestFunction> basicTests = <TestFunction>[
   testHttpProtocolRequest,
   testWebSocketProtocolRequest,
   testHttpAssetRequest
 ];
 
-final List<TestFunction> startPausedTests = [
+final List<TestFunction> startPausedTests = <TestFunction>[
   testStartPaused,
 ];
 
-bool runTests(ShellLauncher launcher, List<TestFunction> tests) async {
-  ShellProcess process = await launcher.launch();
-  Uri uri = await process.waitForObservatory();
+Future<bool> runTests(ShellLauncher launcher, List<TestFunction> tests) async {
+  final ShellProcess process = await launcher.launch();
+  final Uri uri = await process.waitForObservatory();
   try {
-    for (var i = 0; i < tests.length; i++) {
+    for (int i = 0; i < tests.length; i++) {
       print('Executing test ${i+1}/${tests.length}');
       await tests[i](uri);
     }
@@ -173,7 +175,7 @@ bool runTests(ShellLauncher launcher, List<TestFunction> tests) async {
   return exitCode == 0;
 }
 
-main(List<String> args) async {
+Future<Null> main(List<String> args) async {
   if (args.length < 2) {
     print('Usage: dart ${Platform.script} '
           '<sky_shell_executable> <main_dart> ...');
@@ -181,7 +183,7 @@ main(List<String> args) async {
   }
   final String shellExecutablePath = args[0];
   final String mainDartPath = args[1];
-  final List<String> extraArgs = args.length <= 2 ? [] : args.sublist(2);
+  final List<String> extraArgs = args.length <= 2 ? <String>[] : args.sublist(2);
 
   final ShellLauncher launcher =
       new ShellLauncher(shellExecutablePath,