Consolidate observatory code (#3892)

* rename service_protocol.dart to protocol_discovery.dart

* add a wrapper around the obs. protocol

* use json-rpc in run

* consolidate obs. code; implement flutter run --benchmark

* review comments

packages/flutter_tools/lib/src/android/android_device.dart

@@ -14,7 +14,7 @@ import '../build_info.dart';
 import '../device.dart';
 import '../flx.dart' as flx;
 import '../globals.dart';
-import '../service_protocol.dart';
+import '../protocol_discovery.dart';
 import 'adb.dart';
 import 'android.dart';
 
@@ -243,14 +243,12 @@ class AndroidDevice extends Device {
 
     runCheckedSync(adbCommandForDevice(<String>['push', bundlePath, _deviceBundlePath]));
 
-    ServiceProtocolDiscovery observatoryDiscovery;
-    ServiceProtocolDiscovery diagnosticDiscovery;
+    ProtocolDiscovery observatoryDiscovery;
+    ProtocolDiscovery diagnosticDiscovery;
 
     if (options.debuggingEnabled) {
-      observatoryDiscovery = new ServiceProtocolDiscovery(
-        logReader, ServiceProtocolDiscovery.kObservatoryService);
-      diagnosticDiscovery = new ServiceProtocolDiscovery(
-        logReader, ServiceProtocolDiscovery.kDiagnosticService);
+      observatoryDiscovery = new ProtocolDiscovery(logReader, ProtocolDiscovery.kObservatoryService);
+      diagnosticDiscovery = new ProtocolDiscovery(logReader, ProtocolDiscovery.kDiagnosticService);
     }
 
     List<String> cmd = adbCommandForDevice(<String>[
@@ -295,13 +293,11 @@ class AndroidDevice extends Device {
         int observatoryLocalPort = await options.findBestObservatoryPort();
         // TODO(devoncarew): Remember the forwarding information (so we can later remove the
         // port forwarding).
-        await _forwardPort(ServiceProtocolDiscovery.kObservatoryService,
-            observatoryDevicePort, observatoryLocalPort);
+        await _forwardPort(ProtocolDiscovery.kObservatoryService, observatoryDevicePort, observatoryLocalPort);
         int diagnosticDevicePort = devicePorts[1];
         printTrace('diagnostic port = $diagnosticDevicePort');
         int diagnosticLocalPort = await options.findBestDiagnosticPort();
-        await _forwardPort(ServiceProtocolDiscovery.kDiagnosticService,
-            diagnosticDevicePort, diagnosticLocalPort);
+        await _forwardPort(ProtocolDiscovery.kDiagnosticService, diagnosticDevicePort, diagnosticLocalPort);
         return new LaunchResult.succeeded(
           observatoryPort: observatoryLocalPort,
           diagnosticPort: diagnosticLocalPort

packages/flutter_tools/lib/src/base/common.dart

@@ -2,11 +2,6 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-const int defaultObservatoryPort = 8100;
-const int defaultDiagnosticPort  = 8101;
-const int defaultDrivePort       = 8183;
-
-// Names of some of the Timeline events we care about
-const String flutterEngineMainEnterEventName = 'FlutterEngineMainEnter';
-const String frameworkInitEventName = 'Framework initialization';
-const String firstUsefulFrameEventName = 'Widgets completed first useful frame';
+const int kDefaultObservatoryPort = 8100;
+const int kDefaultDiagnosticPort  = 8101;
+const int kDefaultDrivePort       = 8183;

packages/flutter_tools/lib/src/base/utils.dart

@@ -3,6 +3,7 @@
 // found in the LICENSE file.
 
 import 'dart:async';
+import 'dart:convert';
 import 'dart:io';
 
 import 'package:crypto/crypto.dart';
@@ -60,6 +61,10 @@ File getUniqueFile(Directory dir, String baseName, String ext) {
   }
 }
 
+String toPrettyJson(Object jsonable) {
+  return new JsonEncoder.withIndent('  ').convert(jsonable) + '\n';
+}
+
 /// A class to maintain a list of items, fire events when items are added or
 /// removed, and calculate a diff of changes when a new list of items is
 /// available.

packages/flutter_tools/lib/src/cache.dart

@@ -12,7 +12,7 @@ import 'base/logger.dart';
 import 'base/os.dart';
 import 'globals.dart';
 
-/// A warpper around the `bin/cache/` directory.
+/// A wrapper around the `bin/cache/` directory.
 class Cache {
   /// [rootOverride] is configurable for testing.
   Cache({ Directory rootOverride }) {

packages/flutter_tools/lib/src/commands/analyze.dart

@@ -376,9 +376,8 @@ class AnalyzeCommand extends FlutterCommand {
       'time': (stopwatch.elapsedMilliseconds / 1000.0),
       'issues': errorCount
     };
-    JsonEncoder encoder = new JsonEncoder.withIndent('  ');
-    new File(benchmarkOut).writeAsStringSync(encoder.convert(data) + '\n');
-    printStatus('Analysis benchmark written to $benchmarkOut.');
+    new File(benchmarkOut).writeAsStringSync(toPrettyJson(data));
+    printStatus('Analysis benchmark written to $benchmarkOut ($data).');
   }
 }
 

packages/flutter_tools/lib/src/commands/drive.dart

@@ -62,7 +62,7 @@ class DriveCommand extends RunCommandBase {
     );
 
     argParser.addOption('debug-port',
-      defaultsTo: defaultDrivePort.toString(),
+      defaultsTo: kDefaultDrivePort.toString(),
       help: 'Listen to the given port for a debug connection.'
     );
   }

packages/flutter_tools/lib/src/commands/skia.dart

@@ -16,7 +16,7 @@ class SkiaCommand extends FlutterCommand {
     argParser.addOption('output-file', help: 'Write the Skia picture file to this path.');
     argParser.addOption('skiaserve', help: 'Post the picture to a skiaserve debugger at this URL.');
     argParser.addOption('diagnostic-port',
-        defaultsTo: defaultDiagnosticPort.toString(),
+        defaultsTo: kDefaultDiagnosticPort.toString(),
         help: 'Local port where the diagnostic server is listening.');
   }
 

packages/flutter_tools/lib/src/commands/trace.dart

@@ -8,10 +8,15 @@ import 'dart:io';
 
 import '../base/common.dart';
 import '../base/utils.dart';
-import '../device.dart';
 import '../globals.dart';
+import '../observatory.dart';
 import '../runner/flutter_command.dart';
 
+// Names of some of the Timeline events we care about.
+const String kFlutterEngineMainEnterEventName = 'FlutterEngineMainEnter';
+const String kFrameworkInitEventName = 'Framework initialization';
+const String kFirstUsefulFrameEventName = 'Widgets completed first useful frame';
+
 class TraceCommand extends FlutterCommand {
   TraceCommand() {
     argParser.addFlag('start', negatable: false, help: 'Start tracing.');
@@ -20,7 +25,7 @@ class TraceCommand extends FlutterCommand {
     argParser.addOption('duration',
         defaultsTo: '10', abbr: 'd', help: 'Duration in seconds to trace.');
     argParser.addOption('debug-port',
-        defaultsTo: defaultObservatoryPort.toString(),
+        defaultsTo: kDefaultObservatoryPort.toString(),
         help: 'Local port where the observatory is listening.');
   }
 
@@ -41,38 +46,104 @@ class TraceCommand extends FlutterCommand {
 
   @override
   Future<int> runInProject() async {
-    Device device = deviceForCommand;
     int observatoryPort = int.parse(argResults['debug-port']);
 
+    Tracing tracing;
+
+    try {
+      tracing = await Tracing.connect(observatoryPort);
+    } catch (error) {
+      printError('Error connecting to observatory: $error');
+      return 1;
+    }
+
     if ((!argResults['start'] && !argResults['stop']) ||
         (argResults['start'] && argResults['stop'])) {
       // Setting neither flags or both flags means do both commands and wait
       // duration seconds in between.
-      await device.startTracing(observatoryPort);
+      await tracing.startTracing();
       await new Future<Null>.delayed(
         new Duration(seconds: int.parse(argResults['duration'])),
-        () => _stopTracing(device, observatoryPort)
+        () => _stopTracing(tracing)
       );
     } else if (argResults['stop']) {
-      await _stopTracing(device, observatoryPort);
+      await _stopTracing(tracing);
     } else {
-      await device.startTracing(observatoryPort);
+      await tracing.startTracing();
     }
+
     return 0;
   }
 
-  Future<Null> _stopTracing(Device device, int observatoryPort) async {
-    Map<String, dynamic> timeline = await device.stopTracingAndDownloadTimeline(observatoryPort);
-
-    String outPath = argResults['out'];
+  Future<Null> _stopTracing(Tracing tracing) async {
+    Map<String, dynamic> timeline = await tracing.stopTracingAndDownloadTimeline();
     File localFile;
-    if (outPath != null) {
-      localFile = new File(outPath);
+
+    if (argResults['out'] != null) {
+      localFile = new File(argResults['out']);
     } else {
       localFile = getUniqueFile(Directory.current, 'trace', 'json');
     }
 
     await localFile.writeAsString(JSON.encode(timeline));
+
     printStatus('Trace file saved to ${localFile.path}');
   }
 }
+
+class Tracing {
+  Tracing(this.observatory);
+
+  static Future<Tracing> connect(int port) {
+    return Observatory.connect(port).then((Observatory observatory) => new Tracing(observatory));
+  }
+
+  final Observatory observatory;
+
+  Future<Null> startTracing() async {
+    await observatory.setVMTimelineFlags(<String>['Compiler', 'Dart', 'Embedder', 'GC']);
+    await observatory.clearVMTimeline();
+  }
+
+  /// Stops tracing; optionally wait for first frame.
+  Future<Map<String, dynamic>> stopTracingAndDownloadTimeline({
+    bool waitForFirstFrame: false
+  }) async {
+    Response timeline;
+
+    if (!waitForFirstFrame) {
+      // Stop tracing immediately and get the timeline
+      await observatory.setVMTimelineFlags(<String>[]);
+      timeline = await observatory.getVMTimeline();
+    } else {
+      Completer<Null> whenFirstFrameRendered = new Completer<Null>();
+
+      observatory.onTimelineEvent.listen((Event timelineEvent) {
+        List<Map<String, dynamic>> events = timelineEvent['timelineEvents'];
+        for (Map<String, dynamic> event in events) {
+          if (event['name'] == kFirstUsefulFrameEventName)
+            whenFirstFrameRendered.complete();
+        }
+      });
+      await observatory.streamListen('Timeline');
+
+      await whenFirstFrameRendered.future.timeout(
+        const Duration(seconds: 10),
+        onTimeout: () {
+          printError(
+            'Timed out waiting for the first frame event. Either the '
+            'application failed to start, or the event was missed because '
+            '"flutter run" took too long to subscribe to timeline events.'
+          );
+          return null;
+        }
+      );
+
+      timeline = await observatory.getVMTimeline();
+
+      await observatory.setVMTimelineFlags(<String>[]);
+    }
+
+    return timeline.response;
+  }
+}

packages/flutter_tools/lib/src/device.dart

@@ -6,9 +6,6 @@ import 'dart:async';
 import 'dart:io';
 import 'dart:math' as math;
 
-import 'package:json_rpc_2/json_rpc_2.dart' as rpc;
-import 'package:web_socket_channel/io.dart';
-
 import 'android/android_device.dart';
 import 'application_package.dart';
 import 'base/common.dart';
@@ -225,82 +222,6 @@ abstract class Device {
         '${getNameForTargetPlatform(device.platform)}$supportIndicator');
     }
   }
-
-
-  Future<rpc.Peer> _connectToObservatory(int observatoryPort) async {
-    Uri uri = new Uri(scheme: 'ws', host: '127.0.0.1', port: observatoryPort, path: 'ws');
-    WebSocket ws = await WebSocket.connect(uri.toString());
-    rpc.Peer peer = new rpc.Peer(new IOWebSocketChannel(ws));
-    peer.listen();
-    return peer;
-  }
-
-  Future<Null> startTracing(int observatoryPort) async {
-    rpc.Client client;
-    try {
-      client = await _connectToObservatory(observatoryPort);
-    } catch (e) {
-      printError('Error connecting to observatory: $e');
-      return;
-    }
-
-    await client.sendRequest('_setVMTimelineFlags',
-        <String, dynamic>{'recordedStreams': <String>['Compiler', 'Dart', 'Embedder', 'GC']}
-    );
-    await client.sendRequest('_clearVMTimeline');
-  }
-
-  /// Stops tracing, optionally waiting
-  Future<Map<String, dynamic>> stopTracingAndDownloadTimeline(int observatoryPort, {bool waitForFirstFrame: false}) async {
-    rpc.Peer peer;
-    try {
-      peer = await _connectToObservatory(observatoryPort);
-    } catch (e) {
-      printError('Error connecting to observatory: $e');
-      return null;
-    }
-
-    Future<Map<String, dynamic>> fetchTimeline() async {
-      return await peer.sendRequest('_getVMTimeline');
-    }
-
-    Map<String, dynamic> timeline;
-
-    if (!waitForFirstFrame) {
-      // Stop tracing immediately and get the timeline
-      await peer.sendRequest('_setVMTimelineFlags', <String, dynamic>{'recordedStreams': '[]'});
-      timeline = await fetchTimeline();
-    } else {
-      Completer<Null> whenFirstFrameRendered = new Completer<Null>();
-      peer.registerMethod('streamNotify', (rpc.Parameters params) {
-        Map<String, dynamic> data = params.asMap;
-        if (data['streamId'] == 'Timeline') {
-          List<Map<String, dynamic>> events = data['event']['timelineEvents'];
-          for (Map<String, dynamic> event in events) {
-            if (event['name'] == firstUsefulFrameEventName) {
-              whenFirstFrameRendered.complete();
-            }
-          }
-        }
-      });
-      await peer.sendRequest('streamListen', <String, dynamic>{'streamId': 'Timeline'});
-      await whenFirstFrameRendered.future.timeout(
-        const Duration(seconds: 10),
-        onTimeout: () {
-          printError(
-            'Timed out waiting for the first frame event. Either the '
-            'application failed to start, or the event was missed because '
-            '"flutter run" took too long to subscribe to timeline events.'
-          );
-          return null;
-        }
-      );
-      timeline = await fetchTimeline();
-      await peer.sendRequest('_setVMTimelineFlags', <String, dynamic>{'recordedStreams': '[]'});
-    }
-
-    return timeline;
-  }
 }
 
 class DebuggingOptions {
@@ -332,7 +253,7 @@ class DebuggingOptions {
   Future<int> findBestObservatoryPort() {
     if (hasObservatoryPort)
       return new Future<int>.value(observatoryPort);
-    return findPreferredPort(observatoryPort ?? defaultObservatoryPort);
+    return findPreferredPort(observatoryPort ?? kDefaultObservatoryPort);
   }
 
   bool get hasDiagnosticPort => diagnosticPort != null;
@@ -340,7 +261,7 @@ class DebuggingOptions {
   /// Return the user specified diagnostic port. If that isn't available,
   /// return [defaultObservatoryPort], or a port close to that one.
   Future<int> findBestDiagnosticPort() {
-    return findPreferredPort(diagnosticPort ?? defaultDiagnosticPort);
+    return findPreferredPort(diagnosticPort ?? kDefaultDiagnosticPort);
   }
 }
 

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

@@ -15,7 +15,7 @@ import '../build_info.dart';
 import '../device.dart';
 import '../flx.dart' as flx;
 import '../globals.dart';
-import '../service_protocol.dart';
+import '../protocol_discovery.dart';
 import 'mac.dart';
 
 const String _xcrunPath = '/usr/bin/xcrun';
@@ -395,20 +395,17 @@ class IOSSimulator extends Device {
     RegExp versionExp = new RegExp(r'iPhone ([0-9])+');
     Match match = versionExp.firstMatch(name);
 
-    if (match == null) {
     // Not an iPhone. All available non-iPhone simulators are compatible.
+    if (match == null)
       return true;
-    }
 
-    if (int.parse(match.group(1)) > 5) {
     // iPhones 6 and above are always fine.
+    if (int.parse(match.group(1)) > 5)
       return true;
-    }
 
     // The 's' subtype of 5 is compatible.
-    if (name.contains('iPhone 5s')) {
+    if (name.contains('iPhone 5s'))
       return true;
-    }
 
     _supportMessage = "The simulator version is too old. Choose an iPhone 5s or above.";
     return false;
@@ -447,12 +444,10 @@ class IOSSimulator extends Device {
     if (!(await _setupUpdatedApplicationBundle(app)))
       return new LaunchResult.failed();
 
-    ServiceProtocolDiscovery observatoryDiscovery;
+    ProtocolDiscovery observatoryDiscovery;
 
-    if (debuggingOptions.debuggingEnabled) {
-      observatoryDiscovery = new ServiceProtocolDiscovery(
-        logReader, ServiceProtocolDiscovery.kObservatoryService);
-    }
+    if (debuggingOptions.debuggingEnabled)
+      observatoryDiscovery = new ProtocolDiscovery(logReader, ProtocolDiscovery.kObservatoryService);
 
     // Prepare launch arguments.
     List<String> args = <String>[

packages/flutter_tools/lib/src/observatory.dart

+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:io';
+
+import 'package:json_rpc_2/json_rpc_2.dart' as rpc;
+import 'package:web_socket_channel/io.dart';
+
+class Observatory {
+  Observatory._(this.peer, this.port) {
+    peer.registerMethod('streamNotify', (rpc.Parameters event) {
+      _handleStreamNotify(event.asMap);
+    });
+  }
+
+  static Future<Observatory> connect(int port) async {
+    Uri uri = new Uri(scheme: 'ws', host: '127.0.0.1', port: port, path: 'ws');
+    WebSocket ws = await WebSocket.connect(uri.toString());
+    rpc.Peer peer = new rpc.Peer(new IOWebSocketChannel(ws));
+    peer.listen();
+    return new Observatory._(peer, port);
+  }
+
+  final rpc.Peer peer;
+  final int port;
+
+  Map<String, StreamController<Event>> _eventControllers = <String, StreamController<Event>>{};
+
+  bool get isClosed => peer.isClosed;
+  Future<Null> get done => peer.done;
+
+  // Events
+
+  // IsolateStart, IsolateRunnable, IsolateExit, IsolateUpdate, ServiceExtensionAdded
+  Stream<Event> get onIsolateEvent => _getEventController('Isolate').stream;
+  Stream<Event> get onTimelineEvent => _getEventController('Timeline').stream;
+
+  // Listen for a specific event name.
+  Stream<Event> onEvent(String streamName) => _getEventController(streamName).stream;
+
+  StreamController<Event> _getEventController(String eventName) {
+    StreamController<Event> controller = _eventControllers[eventName];
+    if (controller == null) {
+      controller = new StreamController<Event>.broadcast();
+      _eventControllers[eventName] = controller;
+    }
+    return controller;
+  }
+
+  void _handleStreamNotify(Map<String, dynamic> data) {
+    Event event = new Event(data['event']);
+    _getEventController(data['streamId']).add(event);
+  }
+
+  // Requests
+
+  Future<Response> sendRequest(String method, [Map<String, dynamic> args]) {
+    return peer.sendRequest(method, args).then((dynamic result) => new Response(result));
+  }
+
+  Future<Response> streamListen(String streamId) {
+    return sendRequest('streamListen', <String, dynamic>{
+      'streamId': streamId
+    });
+  }
+
+  Future<VM> getVM() {
+    return peer.sendRequest('getVM').then((dynamic result) {
+      return new VM(result);
+    });
+  }
+
+  Future<Response> isolateReload(String isolateId) {
+    return sendRequest('isolateReload', <String, dynamic>{
+      'isolateId': isolateId
+    });
+  }
+
+  Future<Response> clearVMTimeline() => sendRequest('_clearVMTimeline');
+
+  Future<Response> setVMTimelineFlags(List<String> recordedStreams) {
+    assert(recordedStreams != null);
+
+    return sendRequest('_setVMTimelineFlags', <String, dynamic> {
+      'recordedStreams': recordedStreams
+    });
+  }
+
+  Future<Response> getVMTimeline() => sendRequest('_getVMTimeline');
+
+  // Flutter extension methods.
+
+  Future<Response> flutterExit(String isolateId) {
+    return peer.sendRequest('ext.flutter.exit', <String, dynamic>{
+      'isolateId': isolateId
+    }).then((dynamic result) => new Response(result));
+  }
+}
+
+class Response {
+  Response(this.response);
+
+  final Map<String, dynamic> response;
+
+  dynamic operator[](String key) => response[key];
+
+  @override
+  String toString() => response.toString();
+}
+
+class VM extends Response {
+  VM(Map<String, dynamic> response) : super(response);
+
+  List<dynamic> get isolates => response['isolates'];
+}
+
+class Event {
+  Event(this.event);
+
+  final Map<String, dynamic> event;
+
+  String get kind => event['kind'];
+
+  dynamic operator[](String key) => event[key];
+
+  @override
+  String toString() => event.toString();
+}

packages/flutter_tools/lib/src/service_protocol.dart → packages/flutter_tools/lib/src/protocol_discovery.dart

@@ -7,9 +7,9 @@ import 'dart:async';
 import 'device.dart';
 
 /// Discover service protocol ports on devices.
-class ServiceProtocolDiscovery {
+class ProtocolDiscovery {
   /// [logReader] - a [DeviceLogReader] to look for service messages in.
-  ServiceProtocolDiscovery(DeviceLogReader logReader, String serviceName)
+  ProtocolDiscovery(DeviceLogReader logReader, String serviceName)
       : _logReader = logReader, _serviceName = serviceName {
     assert(_logReader != null);
     _subscription = _logReader.logLines.listen(_onLine);

packages/flutter_tools/test/all.dart

@@ -23,8 +23,8 @@ import 'install_test.dart' as install_test;
 import 'listen_test.dart' as listen_test;
 import 'logs_test.dart' as logs_test;
 import 'os_utils_test.dart' as os_utils_test;
+import 'protocol_discovery_test.dart' as protocol_discovery_test;
 import 'run_test.dart' as run_test;
-import 'service_protocol_test.dart' as service_protocol_test;
 import 'stop_test.dart' as stop_test;
 import 'toolchain_test.dart' as toolchain_test;
 import 'trace_test.dart' as trace_test;
@@ -47,8 +47,8 @@ void main() {
   listen_test.main();
   logs_test.main();
   os_utils_test.main();
+  protocol_discovery_test.main();
   run_test.main();
-  service_protocol_test.main();
   stop_test.main();
   toolchain_test.main();
   trace_test.main();

packages/flutter_tools/test/service_protocol_test.dart → packages/flutter_tools/test/protocol_discovery_test.dart

@@ -3,9 +3,9 @@
 // found in the LICENSE file.
 
 import 'dart:async';
-import 'package:test/test.dart';
 
-import 'package:flutter_tools/src/service_protocol.dart';
+import 'package:flutter_tools/src/protocol_discovery.dart';
+import 'package:test/test.dart';
 
 import 'src/mocks.dart';
 
@@ -13,8 +13,8 @@ void main() {
   group('service_protocol', () {
     test('Discovery Heartbeat', () async {
       MockDeviceLogReader logReader = new MockDeviceLogReader();
-      ServiceProtocolDiscovery discoverer =
-          new ServiceProtocolDiscovery(logReader, ServiceProtocolDiscovery.kObservatoryService);
+      ProtocolDiscovery discoverer =
+          new ProtocolDiscovery(logReader, ProtocolDiscovery.kObservatoryService);
 
       // Get next port future.
       Future<int> nextPort = discoverer.nextPort();