Attach looks at future observatory URIs (#44637)

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

@@ -846,25 +846,9 @@ class _AdbLogReader extends DeviceLogReader {
   @override
   String get name => device.name;
 
-  DateTime _timeOrigin;
-
-  DateTime _adbTimestampToDateTime(String adbTimestamp) {
-    // The adb timestamp format is: mm-dd hours:minutes:seconds.milliseconds
-    // Dart's DateTime parse function accepts this format so long as we provide
-    // the year, resulting in:
-    // yyyy-mm-dd hours:minutes:seconds.milliseconds.
-    return DateTime.parse('${DateTime.now().year}-$adbTimestamp');
-  }
-
   void _start() {
-    // Start the adb logcat process.
-    final List<String> args = <String>['shell', '-x', 'logcat', '-v', 'time'];
-    final String lastTimestamp = device.lastLogcatTimestamp;
-    if (lastTimestamp != null) {
-      _timeOrigin = _adbTimestampToDateTime(lastTimestamp);
-    } else {
-      _timeOrigin = null;
-    }
+    // Start the adb logcat process and filter logs by the "flutter" tag.
+    final List<String> args = <String>['shell', '-x', 'logcat', '-v', 'time', '-s', 'flutter'];
     processUtils.start(device.adbCommandForDevice(args)).then<void>((Process process) {
       _process = process;
       // We expect logcat streams to occasionally contain invalid utf-8,
@@ -914,18 +898,7 @@ class _AdbLogReader extends DeviceLogReader {
   // mm-dd hh:mm:ss.milliseconds Priority/Tag( PID): ....
   void _onLine(String line) {
     final Match timeMatch = AndroidDevice._timeRegExp.firstMatch(line);
-    if (timeMatch == null) {
-      return;
-    }
-    if (_timeOrigin != null) {
-      final String timestamp = timeMatch.group(0);
-      final DateTime time = _adbTimestampToDateTime(timestamp);
-      if (!time.isAfter(_timeOrigin)) {
-        // Ignore log messages before the origin.
-        return;
-      }
-    }
-    if (line.length == timeMatch.end) {
+    if (timeMatch == null || line.length == timeMatch.end) {
       return;
     }
     // Chop off the time.

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

@@ -4,6 +4,8 @@
 
 import 'dart:async';
 
+import 'package:meta/meta.dart';
+
 import '../artifacts.dart';
 import '../base/common.dart';
 import '../base/context.dart';
@@ -200,16 +202,14 @@ class AttachCommand extends FlutterCommand {
             notifyingLogger: NotifyingLogger(), logToStdout: true)
       : null;
 
-    Uri observatoryUri;
+    Stream<Uri> observatoryUri;
     bool usesIpv6 = ipv6;
     final String ipv6Loopback = InternetAddress.loopbackIPv6.address;
     final String ipv4Loopback = InternetAddress.loopbackIPv4.address;
     final String hostname = usesIpv6 ? ipv6Loopback : ipv4Loopback;
 
-    bool attachLogger = false;
     if (devicePort == null && debugUri == null) {
       if (device is FuchsiaDevice) {
-        attachLogger = true;
         final String module = stringArg('module');
         if (module == null) {
           throwToolExit('\'--module\' is required for attaching to a Fuchsia device');
@@ -218,8 +218,7 @@ class AttachCommand extends FlutterCommand {
         FuchsiaIsolateDiscoveryProtocol isolateDiscoveryProtocol;
         try {
           isolateDiscoveryProtocol = device.getIsolateDiscoveryProtocol(module);
-          observatoryUri = await isolateDiscoveryProtocol.uri;
-          printStatus('Done.'); // FYI, this message is used as a sentinel in tests.
+          observatoryUri = Stream<Uri>.fromFuture(isolateDiscoveryProtocol.uri).asBroadcastStream();
         } catch (_) {
           isolateDiscoveryProtocol?.dispose();
           final List<ForwardedPort> ports = device.portForwarder.forwardedPorts.toList();
@@ -229,18 +228,20 @@ class AttachCommand extends FlutterCommand {
           rethrow;
         }
       } else if ((device is IOSDevice) || (device is IOSSimulator)) {
-        observatoryUri = await MDnsObservatoryDiscovery.instance.getObservatoryUri(
+        observatoryUri = Stream<Uri>
+          .fromFuture(
+            MDnsObservatoryDiscovery.instance.getObservatoryUri(
               appId,
               device,
               usesIpv6: usesIpv6,
               deviceVmservicePort: deviceVmservicePort,
-        );
+            )
+          ).asBroadcastStream();
       }
       // If MDNS discovery fails or we're not on iOS, fallback to ProtocolDiscovery.
       if (observatoryUri == null) {
-        ProtocolDiscovery observatoryDiscovery;
-        try {
-          observatoryDiscovery = ProtocolDiscovery.observatory(
+        final ProtocolDiscovery observatoryDiscovery =
+          ProtocolDiscovery.observatory(
             device.getLogReader(),
             portForwarder: device.portForwarder,
             ipv6: ipv6,
@@ -248,66 +249,34 @@ class AttachCommand extends FlutterCommand {
             hostPort: hostVmservicePort,
           );
         printStatus('Waiting for a connection from Flutter on ${device.name}...');
-          observatoryUri = await observatoryDiscovery.uri;
+        observatoryUri = observatoryDiscovery.uris;
         // Determine ipv6 status from the scanned logs.
         usesIpv6 = observatoryDiscovery.ipv6;
-          printStatus('Done.'); // FYI, this message is used as a sentinel in tests.
-        } catch (error) {
-          throwToolExit('Failed to establish a debug connection with ${device.name}: $error');
-        } finally {
-          await observatoryDiscovery?.cancel();
-        }
       }
     } else {
-      observatoryUri = await buildObservatoryUri(
+      observatoryUri = Stream<Uri>
+        .fromFuture(
+          buildObservatoryUri(
             device,
             debugUri?.host ?? hostname,
             devicePort ?? debugUri.port,
             hostVmservicePort,
             debugUri?.path,
-      );
-    }
-    try {
-      final bool useHot = getBuildInfo().isDebug;
-      final FlutterDevice flutterDevice = await FlutterDevice.create(
-        device,
-        flutterProject: flutterProject,
-        trackWidgetCreation: boolArg('track-widget-creation'),
-        fileSystemRoots: stringsArg('filesystem-root'),
-        fileSystemScheme: stringArg('filesystem-scheme'),
-        viewFilter: stringArg('isolate-filter'),
-        target: stringArg('target'),
-        targetModel: TargetModel(stringArg('target-model')),
-        buildMode: getBuildMode(),
-        dartDefines: dartDefines,
-      );
-      flutterDevice.observatoryUris = <Uri>[ observatoryUri ];
-      final List<FlutterDevice> flutterDevices =  <FlutterDevice>[flutterDevice];
-      final DebuggingOptions debuggingOptions = DebuggingOptions.enabled(getBuildInfo());
-      terminal.usesTerminalUi = daemon == null;
-      final ResidentRunner runner = useHot ?
-          hotRunnerFactory.build(
-            flutterDevices,
-            target: targetFile,
-            debuggingOptions: debuggingOptions,
-            packagesFilePath: globalResults['packages'] as String,
-            projectRootPath: stringArg('project-root'),
-            dillOutputPath: stringArg('output-dill'),
-            ipv6: usesIpv6,
-            flutterProject: flutterProject,
           )
-        : ColdRunner(
-            flutterDevices,
-            target: targetFile,
-            debuggingOptions: debuggingOptions,
-            ipv6: usesIpv6,
-          );
-      if (attachLogger) {
-        flutterDevice.startEchoingDeviceLog();
+        ).asBroadcastStream();
     }
 
+    terminal.usesTerminalUi = daemon == null;
+
+    try {
       int result;
       if (daemon != null) {
+        final ResidentRunner runner = await createResidentRunner(
+          observatoryUris: observatoryUri,
+          device: device,
+          flutterProject: flutterProject,
+          usesIpv6: usesIpv6,
+        );
         AppInstance app;
         try {
           app = await daemon.appDomain.launch(
@@ -324,21 +293,35 @@ class AttachCommand extends FlutterCommand {
         }
         result = await app.runner.waitForAppToFinish();
         assert(result != null);
-      } else {
+        return;
+      }
+      while (true) {
+        final ResidentRunner runner = await createResidentRunner(
+          observatoryUris: observatoryUri,
+          device: device,
+          flutterProject: flutterProject,
+          usesIpv6: usesIpv6,
+        );
         final Completer<void> onAppStart = Completer<void>.sync();
+        TerminalHandler terminalHandler;
         unawaited(onAppStart.future.whenComplete(() {
-          TerminalHandler(runner)
+          terminalHandler = TerminalHandler(runner)
             ..setupTerminal()
             ..registerSignalHandlers();
         }));
         result = await runner.attach(
           appStartedCompleter: onAppStart,
         );
-        assert(result != null);
-      }
         if (result != 0) {
           throwToolExit(null, exitCode: result);
         }
+        terminalHandler?.stop();
+        assert(result != null);
+        if (runner.exited || !runner.isWaitingForObservatory) {
+          break;
+        }
+        printStatus('Waiting for a new connection from Flutter on ${device.name}...');
+      }
     } finally {
       final List<ForwardedPort> ports = device.portForwarder.forwardedPorts.toList();
       for (ForwardedPort port in ports) {
@@ -347,6 +330,52 @@ class AttachCommand extends FlutterCommand {
     }
   }
 
+  Future<ResidentRunner> createResidentRunner({
+    @required Stream<Uri> observatoryUris,
+    @required Device device,
+    @required FlutterProject flutterProject,
+    @required bool usesIpv6,
+  }) async {
+    assert(observatoryUris != null);
+    assert(device != null);
+    assert(flutterProject != null);
+    assert(usesIpv6 != null);
+
+    final FlutterDevice flutterDevice = await FlutterDevice.create(
+      device,
+      flutterProject: flutterProject,
+      trackWidgetCreation: boolArg('track-widget-creation'),
+      fileSystemRoots: stringsArg('filesystem-root'),
+      fileSystemScheme: stringArg('filesystem-scheme'),
+      viewFilter: stringArg('isolate-filter'),
+      target: stringArg('target'),
+      targetModel: TargetModel(stringArg('target-model')),
+      buildMode: getBuildMode(),
+      dartDefines: dartDefines,
+    );
+    flutterDevice.observatoryUris = observatoryUris;
+    final List<FlutterDevice> flutterDevices =  <FlutterDevice>[flutterDevice];
+    final DebuggingOptions debuggingOptions = DebuggingOptions.enabled(getBuildInfo());
+
+    return getBuildInfo().isDebug
+      ? hotRunnerFactory.build(
+          flutterDevices,
+          target: targetFile,
+          debuggingOptions: debuggingOptions,
+          packagesFilePath: globalResults['packages'] as String,
+          projectRootPath: stringArg('project-root'),
+          dillOutputPath: stringArg('output-dill'),
+          ipv6: usesIpv6,
+          flutterProject: flutterProject,
+        )
+      : ColdRunner(
+          flutterDevices,
+          target: targetFile,
+          debuggingOptions: debuggingOptions,
+          ipv6: usesIpv6,
+        );
+  }
+
   Future<void> _validateArguments() async { }
 }
 

packages/flutter_tools/lib/src/devfs.dart

@@ -11,6 +11,7 @@ import 'asset.dart';
 import 'base/context.dart';
 import 'base/file_system.dart';
 import 'base/io.dart';
+import 'base/net.dart';
 import 'build_info.dart';
 import 'bundle.dart';
 import 'compile.dart';
@@ -265,17 +266,20 @@ class DevFSException implements Exception {
 
 class _DevFSHttpWriter {
   _DevFSHttpWriter(this.fsName, VMService serviceProtocol)
-    : httpAddress = serviceProtocol.httpAddress;
+    : httpAddress = serviceProtocol.httpAddress,
+      _client = (context.get<HttpClientFactory>() == null)
+        ? HttpClient()
+        : context.get<HttpClientFactory>()();
 
   final String fsName;
   final Uri httpAddress;
+  final HttpClient _client;
 
   static const int kMaxInFlight = 6;
 
   int _inFlight = 0;
   Map<Uri, DevFSContent> _outstanding;
   Completer<void> _completer;
-  final HttpClient _client = HttpClient();
 
   Future<void> write(Map<Uri, DevFSContent> entries) async {
     _client.maxConnectionsPerHost = kMaxInFlight;

packages/flutter_tools/lib/src/protocol_discovery.dart

@@ -17,16 +17,23 @@ class ProtocolDiscovery {
     this.logReader,
     this.serviceName, {
     this.portForwarder,
+    this.throttleDuration,
     this.hostPort,
     this.devicePort,
     this.ipv6,
-  }) : assert(logReader != null) {
-    _deviceLogSubscription = logReader.logLines.listen(_handleLine);
+  }) : assert(logReader != null)
+  {
+    _deviceLogSubscription = logReader.logLines.listen(
+      _handleLine,
+      onDone: _stopScrapingLogs,
+    );
+    _uriStreamController = StreamController<Uri>.broadcast();
   }
 
   factory ProtocolDiscovery.observatory(
     DeviceLogReader logReader, {
     DevicePortForwarder portForwarder,
+    Duration throttleDuration = const Duration(milliseconds: 200),
     @required int hostPort,
     @required int devicePort,
     @required bool ipv6,
@@ -36,6 +43,7 @@ class ProtocolDiscovery {
       logReader,
       kObservatoryService,
       portForwarder: portForwarder,
+      throttleDuration: throttleDuration,
       hostPort: hostPort,
       devicePort: devicePort,
       ipv6: ipv6,
@@ -49,38 +57,61 @@ class ProtocolDiscovery {
   final int devicePort;
   final bool ipv6;
 
-  final Completer<Uri> _completer = Completer<Uri>();
+  /// The time to wait before forwarding a new observatory URIs from [logReader].
+  final Duration throttleDuration;
 
   StreamSubscription<String> _deviceLogSubscription;
+  StreamController<Uri> _uriStreamController;
 
   /// The discovered service URI.
+  /// Use [uris] instead.
+  // TODO(egarciad): replace `uri` for `uris`.
+  Future<Uri> get uri {
+    return uris.first;
+  }
+
+  /// The discovered service URIs.
   ///
-  /// Port forwarding is only attempted when this is invoked, in case we never
-  /// need to port forward.
-  Future<Uri> get uri async {
-    final Uri rawUri = await _completer.future;
-    return await _forwardPort(rawUri);
+  /// When a new observatory URI is available in [logReader],
+  /// the URIs are forwarded at most once every [throttleDuration].
+  ///
+  /// Port forwarding is only attempted when this is invoked,
+  /// for each observatory URI in the stream.
+  Stream<Uri> get uris {
+    return _uriStreamController.stream
+      .transform(_throttle<Uri>(
+        waitDuration: throttleDuration,
+      ))
+      .asyncMap<Uri>(_forwardPort);
   }
 
   Future<void> cancel() => _stopScrapingLogs();
 
   Future<void> _stopScrapingLogs() async {
+    await _uriStreamController?.close();
     await _deviceLogSubscription?.cancel();
     _deviceLogSubscription = null;
   }
 
-  void _handleLine(String line) {
-    Uri uri;
+  Match _getPatternMatch(String line) {
     final RegExp r = RegExp('${RegExp.escape(serviceName)} listening on ((http|\/\/)[a-zA-Z0-9:/=_\\-\.\\[\\]]+)');
-    final Match match = r.firstMatch(line);
+    return r.firstMatch(line);
+  }
 
+  Uri _getObservatoryUri(String line) {
+    final Match match = _getPatternMatch(line);
     if (match != null) {
-      try {
-        uri = Uri.parse(match[1]);
-      } on FormatException catch (error, stackTrace) {
-        _stopScrapingLogs();
-        _completer.completeError(error, stackTrace);
+      return Uri.parse(match[1]);
     }
+    return null;
+  }
+
+  void _handleLine(String line) {
+    Uri uri;
+    try {
+      uri = _getObservatoryUri(line);
+    } on FormatException catch(error, stackTrace) {
+      _uriStreamController.addError(error, stackTrace);
     }
     if (uri == null) {
       return;
@@ -89,10 +120,7 @@ class ProtocolDiscovery {
       printTrace('skipping potential observatory $uri due to device port mismatch');
       return;
     }
-
-    assert(!_completer.isCompleted);
-    _stopScrapingLogs();
-    _completer.complete(uri);
+    _uriStreamController.add(uri);
   }
 
   Future<Uri> _forwardPort(Uri deviceUri) async {
@@ -110,7 +138,43 @@ class ProtocolDiscovery {
     if (ipv6) {
       hostUri = hostUri.replace(host: InternetAddress.loopbackIPv6.host);
     }
-
     return hostUri;
   }
 }
+
+/// This transformer will produce an event at most once every [waitDuration].
+///
+/// For example, consider a `waitDuration` of `10ms`, and list of event names
+/// and arrival times: `a (0ms), b (5ms), c (11ms), d (21ms)`.
+/// The events `c` and `d` will be produced as a result.
+StreamTransformer<S, S> _throttle<S>({
+  @required Duration waitDuration,
+}) {
+  assert(waitDuration != null);
+
+  S latestLine;
+  int lastExecution;
+  Future<void> throttleFuture;
+
+  return StreamTransformer<S, S>
+    .fromHandlers(
+      handleData: (S value, EventSink<S> sink) {
+        latestLine = value;
+
+        final int currentTime = DateTime.now().millisecondsSinceEpoch;
+        lastExecution ??= currentTime;
+        final int remainingTime = currentTime - lastExecution;
+        final int nextExecutionTime = remainingTime > waitDuration.inMilliseconds
+          ? 0
+          : waitDuration.inMilliseconds - remainingTime;
+
+        throttleFuture ??= Future<void>
+          .delayed(Duration(milliseconds: nextExecutionTime))
+          .whenComplete(() {
+            sink.add(latestLine);
+            throttleFuture = null;
+            lastExecution = DateTime.now().millisecondsSinceEpoch;
+          });
+      }
+    );
+}

packages/flutter_tools/lib/src/resident_runner.dart

@@ -131,16 +131,20 @@ class FlutterDevice {
 
   final Device device;
   final ResidentCompiler generator;
-  List<Uri> observatoryUris;
+  Stream<Uri> observatoryUris;
   List<VMService> vmServices;
   DevFS devFS;
   ApplicationPackage package;
   List<String> fileSystemRoots;
   String fileSystemScheme;
   StreamSubscription<String> _loggingSubscription;
+  bool _isListeningForObservatoryUri;
   final String viewFilter;
   final bool trackWidgetCreation;
 
+  /// Whether the stream [observatoryUris] is still open.
+  bool get isWaitingForObservatory => _isListeningForObservatoryUri ?? false;
+
   /// If the [reloadSources] parameter is not null the 'reloadSources' service
   /// will be registered.
   /// The 'reloadSources' service can be used by other Service Protocol clients
@@ -154,23 +158,50 @@ class FlutterDevice {
     ReloadSources reloadSources,
     Restart restart,
     CompileExpression compileExpression,
-  }) async {
-    if (vmServices != null) {
-      return;
-    }
-    final List<VMService> localVmServices = List<VMService>(observatoryUris.length);
-    for (int i = 0; i < observatoryUris.length; i += 1) {
-      printTrace('Connecting to service protocol: ${observatoryUris[i]}');
-      localVmServices[i] = await VMService.connect(
-        observatoryUris[i],
+  }) {
+    final Completer<void> completer = Completer<void>();
+    StreamSubscription<void> subscription;
+    bool isWaitingForVm = false;
+
+    subscription = observatoryUris.listen((Uri observatoryUri) async {
+      // FYI, this message is used as a sentinel in tests.
+      printTrace('Connecting to service protocol: $observatoryUri');
+      isWaitingForVm = true;
+      VMService service;
+
+      try {
+        service = await VMService.connect(
+          observatoryUri,
           reloadSources: reloadSources,
           restart: restart,
           compileExpression: compileExpression,
         );
-      printTrace('Successfully connected to service protocol: ${observatoryUris[i]}');
+      } on Exception catch (exception) {
+        printTrace('Fail to connect to service protocol: $observatoryUri: $exception');
+        if (!completer.isCompleted && !_isListeningForObservatoryUri) {
+          completer.completeError('failed to connect to $observatoryUri');
+        }
+        return;
+      }
+      if (completer.isCompleted) {
+        return;
       }
-    vmServices = localVmServices;
+      printTrace('Successfully connected to service protocol: $observatoryUri');
+
+      vmServices = <VMService>[service];
       device.getLogReader(app: package).connectedVMServices = vmServices;
+      completer.complete();
+      await subscription.cancel();
+    }, onError: (dynamic error) {
+      printTrace('Fail to handle observatory URI: $error');
+    }, onDone: () {
+      _isListeningForObservatoryUri = false;
+      if (!completer.isCompleted && !isWaitingForVm) {
+        completer.completeError('connection to device ended too early');
+      }
+    });
+    _isListeningForObservatoryUri = true;
+    return completer.future;
   }
 
   Future<void> refreshViews() async {
@@ -221,6 +252,7 @@ class FlutterDevice {
     if (flutterViews.any((FlutterView view) {
       return view != null &&
              view.uiIsolate != null &&
+             view.uiIsolate.pauseEvent != null &&
              view.uiIsolate.pauseEvent.isPauseEvent;
       }
     )) {
@@ -431,9 +463,13 @@ class FlutterDevice {
       return 2;
     }
     if (result.hasObservatory) {
-      observatoryUris = <Uri>[result.observatoryUri];
+      observatoryUris = Stream<Uri>
+        .value(result.observatoryUri)
+        .asBroadcastStream();
     } else {
-      observatoryUris = <Uri>[];
+      observatoryUris = const Stream<Uri>
+        .empty()
+        .asBroadcastStream();
     }
     return 0;
   }
@@ -491,9 +527,13 @@ class FlutterDevice {
       return 2;
     }
     if (result.hasObservatory) {
-      observatoryUris = <Uri>[result.observatoryUri];
+      observatoryUris = Stream<Uri>
+        .value(result.observatoryUri)
+        .asBroadcastStream();
     } else {
-      observatoryUris = <Uri>[];
+      observatoryUris = const Stream<Uri>
+        .empty()
+        .asBroadcastStream();
     }
     return 0;
   }
@@ -613,14 +653,21 @@ abstract class ResidentRunner {
   /// The parent location of the incremental artifacts.
   @visibleForTesting
   final Directory artifactDirectory;
-  final Completer<int> _finished = Completer<int>();
   final String packagesFilePath;
   final String projectRootPath;
   final String mainPath;
   final AssetBundle assetBundle;
 
   bool _exited = false;
-  bool hotMode ;
+  Completer<int> _finished = Completer<int>();
+  bool hotMode;
+
+  /// Returns true if every device is streaming observatory URIs.
+  bool get isWaitingForObservatory {
+    return flutterDevices.every((FlutterDevice device) {
+      return device.isWaitingForObservatory;
+    });
+  }
 
   String get dillOutputPath => _dillOutputPath ?? fs.path.join(artifactDirectory.path, 'app.dill');
   String getReloadPath({ bool fullRestart }) => mainPath + (fullRestart ? '' : '.incremental') + '.dill';
@@ -631,6 +678,9 @@ abstract class ResidentRunner {
   bool get isRunningRelease => debuggingOptions.buildInfo.isRelease;
   bool get supportsServiceProtocol => isRunningDebug || isRunningProfile;
 
+  /// Returns [true] if the resident runner exited after invoking [exit()].
+  bool get exited => _exited;
+
   /// Whether this runner can hot restart.
   ///
   /// To prevent scenarios where only a subset of devices are hot restarted,
@@ -862,6 +912,8 @@ abstract class ResidentRunner {
       throw 'The service protocol is not enabled.';
     }
 
+    _finished ??= Completer<int>();
+
     bool viewFound = false;
     for (FlutterDevice device in flutterDevices) {
       await device.connect(
@@ -912,22 +964,25 @@ abstract class ResidentRunner {
       // User requested the application exit.
       return;
     }
-    if (_finished.isCompleted) {
+    if (_finished == null || _finished.isCompleted) {
       return;
     }
     printStatus('Lost connection to device.');
     _finished.complete(0);
+    _finished = null;
   }
 
   void appFinished() {
-    if (_finished.isCompleted) {
+    if (_finished == null || _finished.isCompleted) {
       return;
     }
     printStatus('Application finished.');
     _finished.complete(0);
+    _finished = null;
   }
 
   Future<int> waitForAppToFinish() async {
+    _finished ??= Completer<int>();
     final int exitCode = await _finished.future;
     assert(exitCode != null);
     await cleanupAtFinish();
@@ -1045,15 +1100,33 @@ class TerminalHandler {
     subscription = terminal.keystrokes.listen(processTerminalInput);
   }
 
+
+  final Map<io.ProcessSignal, Object> _signalTokens = <io.ProcessSignal, Object>{};
+
+  void _addSignalHandler(io.ProcessSignal signal, SignalHandler handler) {
+    _signalTokens[signal] = signals.addHandler(signal, handler);
+  }
+
   void registerSignalHandlers() {
     assert(residentRunner.stayResident);
-    signals.addHandler(io.ProcessSignal.SIGINT, _cleanUp);
-    signals.addHandler(io.ProcessSignal.SIGTERM, _cleanUp);
+
+    _addSignalHandler(io.ProcessSignal.SIGINT, _cleanUp);
+    _addSignalHandler(io.ProcessSignal.SIGTERM, _cleanUp);
     if (!residentRunner.supportsServiceProtocol || !residentRunner.supportsRestart) {
       return;
     }
-    signals.addHandler(io.ProcessSignal.SIGUSR1, _handleSignal);
-    signals.addHandler(io.ProcessSignal.SIGUSR2, _handleSignal);
+    _addSignalHandler(io.ProcessSignal.SIGUSR1, _handleSignal);
+    _addSignalHandler(io.ProcessSignal.SIGUSR2, _handleSignal);
+  }
+
+  /// Unregisters terminal signal and keystroke handlers.
+  void stop() {
+    assert(residentRunner.stayResident);
+    for (MapEntry<io.ProcessSignal, Object> entry in _signalTokens.entries) {
+      signals.removeHandler(entry.key, entry.value);
+    }
+    _signalTokens.clear();
+    subscription.cancel();
   }
 
   /// Returns [true] if the input has been handled by this function.

packages/flutter_tools/lib/src/run_cold.dart

@@ -83,7 +83,7 @@ class ColdRunner extends ResidentRunner {
     if (flutterDevices.first.observatoryUris != null) {
       // For now, only support one debugger connection.
       connectionInfoCompleter?.complete(DebugConnectionInfo(
-        httpUri: flutterDevices.first.observatoryUris.first,
+        httpUri: flutterDevices.first.vmServices.first.httpAddress,
         wsUri: flutterDevices.first.vmServices.first.wsAddress,
       ));
     }
@@ -184,9 +184,8 @@ class ColdRunner extends ResidentRunner {
     for (FlutterDevice device in flutterDevices) {
       final String dname = device.device.name;
       if (device.observatoryUris != null) {
-        for (Uri uri in device.observatoryUris) {
-          printStatus('An Observatory debugger and profiler on $dname is available at $uri');
-          haveAnything = true;
+        for (VMService vm in device.vmServices) {
+          printStatus('An Observatory debugger and profiler on $dname is available at: ${vm.wsAddress}');
         }
       }
     }

packages/flutter_tools/lib/src/run_hot.dart

@@ -180,7 +180,7 @@ class HotRunner extends ResidentRunner {
         // Only handle one debugger connection.
         connectionInfoCompleter.complete(
           DebugConnectionInfo(
-            httpUri: flutterDevices.first.observatoryUris.first,
+            httpUri: flutterDevices.first.vmServices.first.httpAddress,
             wsUri: flutterDevices.first.vmServices.first.wsAddress,
             baseUri: baseUris.first.toString(),
           ),
@@ -987,8 +987,8 @@ class HotRunner extends ResidentRunner {
     printStatus(message);
     for (FlutterDevice device in flutterDevices) {
       final String dname = device.device.name;
-      for (Uri uri in device.observatoryUris) {
-        printStatus('An Observatory debugger and profiler on $dname is available at: $uri');
+      for (VMService vm in device.vmServices) {
+        printStatus('An Observatory debugger and profiler on $dname is available at: ${vm.wsAddress}');
       }
     }
     final String quitMessage = _didAttach

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

@@ -10,6 +10,7 @@ import 'package:file/file.dart';
 import 'package:file/memory.dart';
 import 'package:flutter_tools/src/base/file_system.dart';
 import 'package:flutter_tools/src/base/io.dart';
+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';
@@ -159,6 +160,7 @@ void main() {
       verify(httpRequest.close()).called(kFailedAttempts + 1);
     }, overrides: <Type, Generator>{
       FileSystem: () => fs,
+      HttpClientFactory: () => () => httpClient,
       ProcessManager: () => FakeProcessManager.any(),
     });
   });
@@ -208,6 +210,7 @@ void main() {
       expect(report.success, true);
     }, overrides: <Type, Generator>{
       FileSystem: () => fs,
+      HttpClient: () => () => HttpClient(),
       ProcessManager: () => FakeProcessManager.any(),
     });
 
@@ -310,6 +313,7 @@ void main() {
       expect(devFS.lastCompiled, isNot(previousCompile));
     }, overrides: <Type, Generator>{
       FileSystem: () => fs,
+      HttpClient: () => () => HttpClient(),
       ProcessManager: () => FakeProcessManager.any(),
     });
   });

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

@@ -6,6 +6,7 @@ import 'dart:async';
 
 import 'package:flutter_tools/src/device.dart';
 import 'package:flutter_tools/src/protocol_discovery.dart';
+import 'package:quiver/testing/async.dart';
 
 import '../src/common.dart';
 import '../src/context.dart';
@@ -16,9 +17,6 @@ void main() {
     MockDeviceLogReader logReader;
     ProtocolDiscovery discoverer;
 
-    group('no port forwarding', () {
-      int devicePort;
-
     /// Performs test set-up functionality that must be performed as part of
     /// the `test()` pass and not part of the `setUp()` pass.
     ///
@@ -35,21 +33,31 @@ void main() {
     /// because in real code, the zone is set up in `main()`.
     ///
     /// See also: [runZoned]
-      void initialize() {
+    void initialize({
+      int devicePort,
+      Duration throttleDuration = const Duration(milliseconds: 200),
+    }) {
       logReader = MockDeviceLogReader();
-        discoverer = ProtocolDiscovery.observatory(logReader, ipv6: false, hostPort: null, devicePort: devicePort);
+      discoverer = ProtocolDiscovery.observatory(
+        logReader,
+        ipv6: false,
+        hostPort: null,
+        devicePort: devicePort,
+        throttleDuration: throttleDuration,
+      );
     }
 
-      tearDown(() {
-        discoverer.cancel();
-        logReader.dispose();
-      });
-
     testUsingContext('returns non-null uri future', () async {
       initialize();
       expect(discoverer.uri, isNotNull);
     });
 
+    group('no port forwarding', () {
+      tearDown(() {
+        discoverer.cancel();
+        logReader.dispose();
+      });
+
       testUsingContext('discovers uri if logs already produced output', () async {
         initialize();
         logReader.addLine('HELLO WORLD');
@@ -78,9 +86,7 @@ void main() {
 
       testUsingContext('uri throws if logs produce bad line', () async {
         initialize();
-        Timer.run(() {
         logReader.addLine('Observatory listening on http://127.0.0.1:apple');
-        });
         expect(discoverer.uri, throwsA(isFormatException));
       });
 
@@ -123,8 +129,7 @@ void main() {
       });
 
       testUsingContext('skips uri if port does not match the requested vmservice - requested last', () async {
-        devicePort = 12346;
-        initialize();
+        initialize(devicePort: 12346);
         final Future<Uri> uriFuture = discoverer.uri;
         logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12345/PTwjm8Ii8qg=/');
         logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12346/PTwjm8Ii8qg=/');
@@ -134,8 +139,7 @@ void main() {
       });
 
       testUsingContext('skips uri if port does not match the requested vmservice - requested first', () async {
-        devicePort = 12346;
-        initialize();
+        initialize(devicePort: 12346);
         final Future<Uri> uriFuture = discoverer.uri;
         logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12346/PTwjm8Ii8qg=/');
         logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12345/PTwjm8Ii8qg=/');
@@ -143,6 +147,86 @@ void main() {
         expect(uri.port, 12346);
         expect('$uri', 'http://127.0.0.1:12346/PTwjm8Ii8qg=/');
       });
+
+      testUsingContext('first uri in the stream is the last one from the log', () async {
+        initialize();
+        logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12346/PTwjm8Ii8qg=/');
+        logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12345/PTwjm8Ii8qg=/');
+        final Uri uri = await discoverer.uris.first;
+        expect(uri.port, 12345);
+        expect('$uri', 'http://127.0.0.1:12345/PTwjm8Ii8qg=/');
+      });
+
+      testUsingContext('first uri in the stream is the last one from the log that matches the port', () async {
+        initialize(devicePort: 12345);
+        logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12346/PTwjm8Ii8qg=/');
+        logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12345/PTwjm8Ii8qg=/');
+        logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12344/PTwjm8Ii8qg=/');
+        final Uri uri = await discoverer.uris.first;
+        expect(uri.port, 12345);
+        expect('$uri', 'http://127.0.0.1:12345/PTwjm8Ii8qg=/');
+      });
+
+      testUsingContext('uris in the stream are throttled', () async {
+        const Duration kThrottleDuration = Duration(milliseconds: 10);
+
+        FakeAsync().run((FakeAsync time) {
+          initialize(throttleDuration: kThrottleDuration);
+
+          final List<Uri> discoveredUris = <Uri>[];
+          discoverer.uris.listen((Uri uri) {
+            discoveredUris.add(uri);
+          });
+
+          logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12346/PTwjm8Ii8qg=/');
+          logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12345/PTwjm8Ii8qg=/');
+
+          time.elapse(kThrottleDuration);
+
+          logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12344/PTwjm8Ii8qg=/');
+          logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12343/PTwjm8Ii8qg=/');
+
+          time.elapse(kThrottleDuration);
+
+          expect(discoveredUris.length, 2);
+          expect(discoveredUris[0].port, 12345);
+          expect('${discoveredUris[0]}', 'http://127.0.0.1:12345/PTwjm8Ii8qg=/');
+          expect(discoveredUris[1].port, 12343);
+          expect('${discoveredUris[1]}', 'http://127.0.0.1:12343/PTwjm8Ii8qg=/');
+        });
+      });
+
+      testUsingContext('uris in the stream are throttled when they match the port', () async {
+        const Duration kThrottleTimeInMilliseconds = Duration(milliseconds: 10);
+
+        FakeAsync().run((FakeAsync time) {
+          initialize(
+            devicePort: 12345,
+            throttleDuration: kThrottleTimeInMilliseconds,
+          );
+
+          final List<Uri> discoveredUris = <Uri>[];
+          discoverer.uris.listen((Uri uri) {
+            discoveredUris.add(uri);
+          });
+
+          logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12346/PTwjm8Ii8qg=/');
+          logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12345/PTwjm8Ii8qg=/');
+
+          time.elapse(kThrottleTimeInMilliseconds);
+
+          logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12345/PTwjm8Ii8qc=/');
+          logReader.addLine('I/flutter : Observatory listening on http://127.0.0.1:12344/PTwjm8Ii8qf=/');
+
+          time.elapse(kThrottleTimeInMilliseconds);
+
+          expect(discoveredUris.length, 2);
+          expect(discoveredUris[0].port, 12345);
+          expect('${discoveredUris[0]}', 'http://127.0.0.1:12345/PTwjm8Ii8qg=/');
+          expect(discoveredUris[1].port, 12345);
+          expect('${discoveredUris[1]}', 'http://127.0.0.1:12345/PTwjm8Ii8qc=/');
+        });
+      });
     });
 
     group('port forwarding', () {
@@ -164,7 +248,7 @@ void main() {
         expect('$uri', 'http://127.0.0.1:99/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        logReader.dispose();
+        await logReader.dispose();
       });
 
       testUsingContext('specified port', () async {
@@ -185,7 +269,7 @@ void main() {
         expect('$uri', 'http://127.0.0.1:1243/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        logReader.dispose();
+        await logReader.dispose();
       });
 
       testUsingContext('specified port zero', () async {
@@ -206,7 +290,7 @@ void main() {
         expect('$uri', 'http://127.0.0.1:99/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        logReader.dispose();
+        await logReader.dispose();
       });
 
       testUsingContext('ipv6', () async {
@@ -227,7 +311,7 @@ void main() {
         expect('$uri', 'http://[::1]:54777/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        logReader.dispose();
+        await logReader.dispose();
       });
 
       testUsingContext('ipv6 with Ascii Escape code', () async {
@@ -248,7 +332,7 @@ void main() {
         expect('$uri', 'http://[::1]:54777/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        logReader.dispose();
+        await logReader.dispose();
       });
     });
   });

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

@@ -95,9 +95,7 @@ void main() {
     when(mockFlutterView.uiIsolate).thenReturn(mockIsolate);
     when(mockFlutterView.runFromSource(any, any, any)).thenAnswer((Invocation invocation) async {});
     when(mockFlutterDevice.stopEchoingDeviceLog()).thenAnswer((Invocation invocation) async { });
-    when(mockFlutterDevice.observatoryUris).thenReturn(<Uri>[
-      testUri,
-    ]);
+    when(mockFlutterDevice.observatoryUris).thenAnswer((_) => Stream<Uri>.value(testUri));
     when(mockFlutterDevice.connect(
       reloadSources: anyNamed('reloadSources'),
       restart: anyNamed('restart'),
@@ -636,7 +634,7 @@ void main() {
     final TestFlutterDevice flutterDevice = TestFlutterDevice(
       mockDevice,
       <FlutterView>[],
-      observatoryUris: <Uri>[ testUri ]
+      observatoryUris: Stream<Uri>.value(testUri),
     );
 
     await flutterDevice.connect();
@@ -657,7 +655,7 @@ class MockUsage extends Mock implements Usage {}
 class MockProcessManager extends Mock implements ProcessManager {}
 class MockServiceEvent extends Mock implements ServiceEvent {}
 class TestFlutterDevice extends FlutterDevice {
-  TestFlutterDevice(Device device, this.views, { List<Uri> observatoryUris })
+  TestFlutterDevice(Device device, this.views, { Stream<Uri> observatoryUris })
     : super(device, buildMode: BuildMode.debug, trackWidgetCreation: false) {
     _observatoryUris = observatoryUris;
   }
@@ -666,8 +664,8 @@ class TestFlutterDevice extends FlutterDevice {
   final List<FlutterView> views;
 
   @override
-  List<Uri> get observatoryUris => _observatoryUris;
-  List<Uri> _observatoryUris;
+  Stream<Uri> get observatoryUris => _observatoryUris;
+  Stream<Uri> _observatoryUris;
 }
 
 class ThrowingForwardingFileSystem extends ForwardingFileSystem {

packages/flutter_tools/test/src/mocks.dart

@@ -533,6 +533,12 @@ class MockAndroidDevice extends Mock implements AndroidDevice {
   @override
   bool isSupported() => true;
 
+  @override
+  bool get supportsHotRestart => true;
+
+  @override
+  bool get supportsFlutterExit => false;
+
   @override
   bool isSupportedForProject(FlutterProject flutterProject) => true;
 }
@@ -563,16 +569,33 @@ class MockDeviceLogReader extends DeviceLogReader {
   @override
   String get name => 'MockLogReader';
 
-  final StreamController<String> _linesController = StreamController<String>.broadcast();
+  StreamController<String> _cachedLinesController;
+
+  final List<String> _lineQueue = <String>[];
+  StreamController<String> get _linesController {
+    _cachedLinesController ??= StreamController<String>
+      .broadcast(onListen: () {
+        _lineQueue.forEach(_linesController.add);
+        _lineQueue.clear();
+     });
+    return _cachedLinesController;
+  }
 
   @override
   Stream<String> get logLines => _linesController.stream;
 
-  void addLine(String line) => _linesController.add(line);
+  void addLine(String line) {
+    if (_linesController.hasListener) {
+      _linesController.add(line);
+    } else {
+      _lineQueue.add(line);
+    }
+  }
 
   @override
-  void dispose() {
-    _linesController.close();
+  Future<void> dispose() async {
+    _lineQueue.clear();
+    await _linesController.close();
   }
 }