Revert "Attach looks at future observatory URIs (#44637)" (#45211)

This reverts commit 6d77996d.

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

@@ -846,9 +846,25 @@ 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 and filter logs by the "flutter" tag.
-    final List<String> args = <String>['shell', '-x', 'logcat', '-v', 'time', '-s', 'flutter'];
+    // 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;
+    }
     processUtils.start(device.adbCommandForDevice(args)).then<void>((Process process) {
       _process = process;
       // We expect logcat streams to occasionally contain invalid utf-8,
@@ -898,7 +914,18 @@ 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 || line.length == timeMatch.end) {
+    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) {
       return;
     }
     // Chop off the time.

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

@@ -4,8 +4,6 @@
 
 import 'dart:async';
 
-import 'package:meta/meta.dart';
-
 import '../artifacts.dart';
 import '../base/common.dart';
 import '../base/context.dart';
@@ -202,14 +200,16 @@ class AttachCommand extends FlutterCommand {
             notifyingLogger: NotifyingLogger(), logToStdout: true)
       : null;
 
-    Stream<Uri> observatoryUri;
+    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,7 +218,8 @@ class AttachCommand extends FlutterCommand {
         FuchsiaIsolateDiscoveryProtocol isolateDiscoveryProtocol;
         try {
           isolateDiscoveryProtocol = device.getIsolateDiscoveryProtocol(module);
-          observatoryUri = Stream<Uri>.fromFuture(isolateDiscoveryProtocol.uri).asBroadcastStream();
+          observatoryUri = await isolateDiscoveryProtocol.uri;
+          printStatus('Done.'); // FYI, this message is used as a sentinel in tests.
         } catch (_) {
           isolateDiscoveryProtocol?.dispose();
           final List<ForwardedPort> ports = device.portForwarder.forwardedPorts.toList();
@@ -228,55 +229,85 @@ class AttachCommand extends FlutterCommand {
           rethrow;
         }
       } else if ((device is IOSDevice) || (device is IOSSimulator)) {
-        observatoryUri = Stream<Uri>
-          .fromFuture(
-            MDnsObservatoryDiscovery.instance.getObservatoryUri(
-              appId,
-              device,
-              usesIpv6: usesIpv6,
-              deviceVmservicePort: deviceVmservicePort,
-            )
-          ).asBroadcastStream();
+        observatoryUri = await MDnsObservatoryDiscovery.instance.getObservatoryUri(
+          appId,
+          device,
+          usesIpv6: usesIpv6,
+          deviceVmservicePort: deviceVmservicePort,
+        );
       }
       // If MDNS discovery fails or we're not on iOS, fallback to ProtocolDiscovery.
       if (observatoryUri == null) {
-        final ProtocolDiscovery observatoryDiscovery =
-          ProtocolDiscovery.observatory(
+        ProtocolDiscovery observatoryDiscovery;
+        try {
+          observatoryDiscovery = ProtocolDiscovery.observatory(
             device.getLogReader(),
             portForwarder: device.portForwarder,
             ipv6: ipv6,
             devicePort: deviceVmservicePort,
             hostPort: hostVmservicePort,
           );
-        printStatus('Waiting for a connection from Flutter on ${device.name}...');
-        observatoryUri = observatoryDiscovery.uris;
-        // Determine ipv6 status from the scanned logs.
-        usesIpv6 = observatoryDiscovery.ipv6;
+          printStatus('Waiting for a connection from Flutter on ${device.name}...');
+          observatoryUri = await observatoryDiscovery.uri;
+          // 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 = Stream<Uri>
-        .fromFuture(
-          buildObservatoryUri(
-            device,
-            debugUri?.host ?? hostname,
-            devicePort ?? debugUri.port,
-            hostVmservicePort,
-            debugUri?.path,
-          )
-        ).asBroadcastStream();
+      observatoryUri = await buildObservatoryUri(
+        device,
+        debugUri?.host ?? hostname,
+        devicePort ?? debugUri.port,
+        hostVmservicePort,
+        debugUri?.path,
+      );
     }
-
-    terminal.usesTerminalUi = daemon == null;
-
     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();
+      }
+
       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(
@@ -293,34 +324,20 @@ class AttachCommand extends FlutterCommand {
         }
         result = await app.runner.waitForAppToFinish();
         assert(result != null);
-        return;
-      }
-      while (true) {
-        final ResidentRunner runner = await createResidentRunner(
-          observatoryUris: observatoryUri,
-          device: device,
-          flutterProject: flutterProject,
-          usesIpv6: usesIpv6,
-        );
+      } else {
         final Completer<void> onAppStart = Completer<void>.sync();
-        TerminalHandler terminalHandler;
         unawaited(onAppStart.future.whenComplete(() {
-          terminalHandler = TerminalHandler(runner)
+          TerminalHandler(runner)
             ..setupTerminal()
             ..registerSignalHandlers();
         }));
         result = await runner.attach(
           appStartedCompleter: onAppStart,
         );
-        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}...');
+      }
+      if (result != 0) {
+        throwToolExit(null, exitCode: result);
       }
     } finally {
       final List<ForwardedPort> ports = device.portForwarder.forwardedPorts.toList();
@@ -330,52 +347,6 @@ 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,7 +11,6 @@ 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';
@@ -266,20 +265,17 @@ class DevFSException implements Exception {
 
 class _DevFSHttpWriter {
   _DevFSHttpWriter(this.fsName, VMService serviceProtocol)
-    : httpAddress = serviceProtocol.httpAddress,
-      _client = (context.get<HttpClientFactory>() == null)
-        ? HttpClient()
-        : context.get<HttpClientFactory>()();
+    : httpAddress = serviceProtocol.httpAddress;
 
   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,23 +17,16 @@ class ProtocolDiscovery {
     this.logReader,
     this.serviceName, {
     this.portForwarder,
-    this.throttleDuration,
     this.hostPort,
     this.devicePort,
     this.ipv6,
-  }) : assert(logReader != null)
-  {
-    _deviceLogSubscription = logReader.logLines.listen(
-      _handleLine,
-      onDone: _stopScrapingLogs,
-    );
-    _uriStreamController = StreamController<Uri>.broadcast();
+  }) : assert(logReader != null) {
+    _deviceLogSubscription = logReader.logLines.listen(_handleLine);
   }
 
   factory ProtocolDiscovery.observatory(
     DeviceLogReader logReader, {
     DevicePortForwarder portForwarder,
-    Duration throttleDuration = const Duration(milliseconds: 200),
     @required int hostPort,
     @required int devicePort,
     @required bool ipv6,
@@ -43,7 +36,6 @@ class ProtocolDiscovery {
       logReader,
       kObservatoryService,
       portForwarder: portForwarder,
-      throttleDuration: throttleDuration,
       hostPort: hostPort,
       devicePort: devicePort,
       ipv6: ipv6,
@@ -57,70 +49,50 @@ class ProtocolDiscovery {
   final int devicePort;
   final bool ipv6;
 
-  /// The time to wait before forwarding a new observatory URIs from [logReader].
-  final Duration throttleDuration;
+  final Completer<Uri> _completer = Completer<Uri>();
 
   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.
   ///
-  /// 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);
+  /// 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);
   }
 
   Future<void> cancel() => _stopScrapingLogs();
 
   Future<void> _stopScrapingLogs() async {
-    await _uriStreamController?.close();
     await _deviceLogSubscription?.cancel();
     _deviceLogSubscription = null;
   }
 
-  Match _getPatternMatch(String line) {
+  void _handleLine(String line) {
+    Uri uri;
     final RegExp r = RegExp('${RegExp.escape(serviceName)} listening on ((http|\/\/)[a-zA-Z0-9:/=_\\-\.\\[\\]]+)');
-    return r.firstMatch(line);
-  }
+    final Match match = r.firstMatch(line);
 
-  Uri _getObservatoryUri(String line) {
-    final Match match = _getPatternMatch(line);
     if (match != null) {
-      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);
+      try {
+        uri = Uri.parse(match[1]);
+      } on FormatException catch (error, stackTrace) {
+        _stopScrapingLogs();
+        _completer.completeError(error, stackTrace);
+      }
     }
     if (uri == null) {
       return;
     }
-    if (devicePort != null && uri.port != devicePort) {
+    if (devicePort != null  &&  uri.port != devicePort) {
       printTrace('skipping potential observatory $uri due to device port mismatch');
       return;
     }
-    _uriStreamController.add(uri);
+
+    assert(!_completer.isCompleted);
+    _stopScrapingLogs();
+    _completer.complete(uri);
   }
 
   Future<Uri> _forwardPort(Uri deviceUri) async {
@@ -138,43 +110,7 @@ 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,20 +131,16 @@ class FlutterDevice {
 
   final Device device;
   final ResidentCompiler generator;
-  Stream<Uri> observatoryUris;
+  List<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
@@ -158,50 +154,23 @@ class FlutterDevice {
     ReloadSources reloadSources,
     Restart restart,
     CompileExpression compileExpression,
-  }) {
-    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,
-        );
-      } 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;
-      }
-      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;
+  }) 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],
+        reloadSources: reloadSources,
+        restart: restart,
+        compileExpression: compileExpression,
+      );
+      printTrace('Successfully connected to service protocol: ${observatoryUris[i]}');
+    }
+    vmServices = localVmServices;
+    device.getLogReader(app: package).connectedVMServices = vmServices;
   }
 
   Future<void> refreshViews() async {
@@ -252,7 +221,6 @@ class FlutterDevice {
     if (flutterViews.any((FlutterView view) {
       return view != null &&
              view.uiIsolate != null &&
-             view.uiIsolate.pauseEvent != null &&
              view.uiIsolate.pauseEvent.isPauseEvent;
       }
     )) {
@@ -463,13 +431,9 @@ class FlutterDevice {
       return 2;
     }
     if (result.hasObservatory) {
-      observatoryUris = Stream<Uri>
-        .value(result.observatoryUri)
-        .asBroadcastStream();
+      observatoryUris = <Uri>[result.observatoryUri];
     } else {
-      observatoryUris = const Stream<Uri>
-        .empty()
-        .asBroadcastStream();
+      observatoryUris = <Uri>[];
     }
     return 0;
   }
@@ -527,13 +491,9 @@ class FlutterDevice {
       return 2;
     }
     if (result.hasObservatory) {
-      observatoryUris = Stream<Uri>
-        .value(result.observatoryUri)
-        .asBroadcastStream();
+      observatoryUris = <Uri>[result.observatoryUri];
     } else {
-      observatoryUris = const Stream<Uri>
-        .empty()
-        .asBroadcastStream();
+      observatoryUris = <Uri>[];
     }
     return 0;
   }
@@ -653,21 +613,14 @@ 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;
-  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;
-    });
-  }
+  bool hotMode ;
 
   String get dillOutputPath => _dillOutputPath ?? fs.path.join(artifactDirectory.path, 'app.dill');
   String getReloadPath({ bool fullRestart }) => mainPath + (fullRestart ? '' : '.incremental') + '.dill';
@@ -678,9 +631,6 @@ 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,
@@ -912,8 +862,6 @@ abstract class ResidentRunner {
       throw 'The service protocol is not enabled.';
     }
 
-    _finished ??= Completer<int>();
-
     bool viewFound = false;
     for (FlutterDevice device in flutterDevices) {
       await device.connect(
@@ -964,25 +912,22 @@ abstract class ResidentRunner {
       // User requested the application exit.
       return;
     }
-    if (_finished == null || _finished.isCompleted) {
+    if (_finished.isCompleted) {
       return;
     }
     printStatus('Lost connection to device.');
     _finished.complete(0);
-    _finished = null;
   }
 
   void appFinished() {
-    if (_finished == null || _finished.isCompleted) {
+    if (_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();
@@ -1100,33 +1045,15 @@ 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);
-
-    _addSignalHandler(io.ProcessSignal.SIGINT, _cleanUp);
-    _addSignalHandler(io.ProcessSignal.SIGTERM, _cleanUp);
+    signals.addHandler(io.ProcessSignal.SIGINT, _cleanUp);
+    signals.addHandler(io.ProcessSignal.SIGTERM, _cleanUp);
     if (!residentRunner.supportsServiceProtocol || !residentRunner.supportsRestart) {
       return;
     }
-    _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();
+    signals.addHandler(io.ProcessSignal.SIGUSR1, _handleSignal);
+    signals.addHandler(io.ProcessSignal.SIGUSR2, _handleSignal);
   }
 
   /// 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.vmServices.first.httpAddress,
+        httpUri: flutterDevices.first.observatoryUris.first,
         wsUri: flutterDevices.first.vmServices.first.wsAddress,
       ));
     }
@@ -184,8 +184,9 @@ class ColdRunner extends ResidentRunner {
     for (FlutterDevice device in flutterDevices) {
       final String dname = device.device.name;
       if (device.observatoryUris != null) {
-        for (VMService vm in device.vmServices) {
-          printStatus('An Observatory debugger and profiler on $dname is available at: ${vm.wsAddress}');
+        for (Uri uri in device.observatoryUris) {
+          printStatus('An Observatory debugger and profiler on $dname is available at $uri');
+          haveAnything = true;
         }
       }
     }

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.vmServices.first.httpAddress,
+            httpUri: flutterDevices.first.observatoryUris.first,
             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 (VMService vm in device.vmServices) {
-        printStatus('An Observatory debugger and profiler on $dname is available at: ${vm.wsAddress}');
+      for (Uri uri in device.observatoryUris) {
+        printStatus('An Observatory debugger and profiler on $dname is available at: $uri');
       }
     }
     final String quitMessage = _didAttach

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

@@ -10,7 +10,6 @@ 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';
@@ -160,7 +159,6 @@ void main() {
       verify(httpRequest.close()).called(kFailedAttempts + 1);
     }, overrides: <Type, Generator>{
       FileSystem: () => fs,
-      HttpClientFactory: () => () => httpClient,
       ProcessManager: () => FakeProcessManager.any(),
     });
   });
@@ -210,7 +208,6 @@ void main() {
       expect(report.success, true);
     }, overrides: <Type, Generator>{
       FileSystem: () => fs,
-      HttpClient: () => () => HttpClient(),
       ProcessManager: () => FakeProcessManager.any(),
     });
 
@@ -313,7 +310,6 @@ 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,7 +6,6 @@ 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';
@@ -17,47 +16,40 @@ void main() {
     MockDeviceLogReader logReader;
     ProtocolDiscovery discoverer;
 
-    /// Performs test set-up functionality that must be performed as part of
-    /// the `test()` pass and not part of the `setUp()` pass.
-    ///
-    /// This exists to make sure we're not creating an error that tries to
-    /// cross an error-zone boundary. Our use of `testUsingContext()` runs the
-    /// test code inside an error zone, but the `setUp()` code is not run in
-    /// any zone. This creates the potential for errors that try to cross
-    /// error-zone boundaries, which are considered uncaught.
-    ///
-    /// This also exists for cases where our initialization requires access to
-    /// a `Context` object, which is only set up inside the zone.
-    ///
-    /// These issues do not pertain to real code and are a test-only concern,
-    /// because in real code, the zone is set up in `main()`.
-    ///
-    /// See also: [runZoned]
-    void initialize({
-      int devicePort,
-      Duration throttleDuration = const Duration(milliseconds: 200),
-    }) {
-      logReader = MockDeviceLogReader();
-      discoverer = ProtocolDiscovery.observatory(
-        logReader,
-        ipv6: false,
-        hostPort: null,
-        devicePort: devicePort,
-        throttleDuration: throttleDuration,
-      );
-    }
-
-    testUsingContext('returns non-null uri future', () async {
-      initialize();
-      expect(discoverer.uri, isNotNull);
-    });
-
     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.
+      ///
+      /// This exists to make sure we're not creating an error that tries to
+      /// cross an error-zone boundary. Our use of `testUsingContext()` runs the
+      /// test code inside an error zone, but the `setUp()` code is not run in
+      /// any zone. This creates the potential for errors that try to cross
+      /// error-zone boundaries, which are considered uncaught.
+      ///
+      /// This also exists for cases where our initialization requires access to
+      /// a `Context` object, which is only set up inside the zone.
+      ///
+      /// These issues do not pertain to real code and are a test-only concern,
+      /// because in real code, the zone is set up in `main()`.
+      ///
+      /// See also: [runZoned]
+      void initialize() {
+        logReader = MockDeviceLogReader();
+        discoverer = ProtocolDiscovery.observatory(logReader, ipv6: false, hostPort: null, devicePort: devicePort);
+      }
+
       tearDown(() {
         discoverer.cancel();
         logReader.dispose();
       });
 
+      testUsingContext('returns non-null uri future', () async {
+        initialize();
+        expect(discoverer.uri, isNotNull);
+      });
+
       testUsingContext('discovers uri if logs already produced output', () async {
         initialize();
         logReader.addLine('HELLO WORLD');
@@ -86,7 +78,9 @@ void main() {
 
       testUsingContext('uri throws if logs produce bad line', () async {
         initialize();
-        logReader.addLine('Observatory listening on http://127.0.0.1:apple');
+        Timer.run(() {
+          logReader.addLine('Observatory listening on http://127.0.0.1:apple');
+        });
         expect(discoverer.uri, throwsA(isFormatException));
       });
 
@@ -129,7 +123,8 @@ void main() {
       });
 
       testUsingContext('skips uri if port does not match the requested vmservice - requested last', () async {
-        initialize(devicePort: 12346);
+        devicePort = 12346;
+        initialize();
         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=/');
@@ -139,7 +134,8 @@ void main() {
       });
 
       testUsingContext('skips uri if port does not match the requested vmservice - requested first', () async {
-        initialize(devicePort: 12346);
+        devicePort = 12346;
+        initialize();
         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=/');
@@ -147,86 +143,6 @@ 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', () {
@@ -248,7 +164,7 @@ void main() {
         expect('$uri', 'http://127.0.0.1:99/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        await logReader.dispose();
+        logReader.dispose();
       });
 
       testUsingContext('specified port', () async {
@@ -269,7 +185,7 @@ void main() {
         expect('$uri', 'http://127.0.0.1:1243/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        await logReader.dispose();
+        logReader.dispose();
       });
 
       testUsingContext('specified port zero', () async {
@@ -290,7 +206,7 @@ void main() {
         expect('$uri', 'http://127.0.0.1:99/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        await logReader.dispose();
+        logReader.dispose();
       });
 
       testUsingContext('ipv6', () async {
@@ -311,7 +227,7 @@ void main() {
         expect('$uri', 'http://[::1]:54777/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        await logReader.dispose();
+        logReader.dispose();
       });
 
       testUsingContext('ipv6 with Ascii Escape code', () async {
@@ -332,7 +248,7 @@ void main() {
         expect('$uri', 'http://[::1]:54777/PTwjm8Ii8qg=/');
 
         await discoverer.cancel();
-        await logReader.dispose();
+        logReader.dispose();
       });
     });
   });

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

@@ -95,7 +95,9 @@ 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).thenAnswer((_) => Stream<Uri>.value(testUri));
+    when(mockFlutterDevice.observatoryUris).thenReturn(<Uri>[
+      testUri,
+    ]);
     when(mockFlutterDevice.connect(
       reloadSources: anyNamed('reloadSources'),
       restart: anyNamed('restart'),
@@ -634,7 +636,7 @@ void main() {
     final TestFlutterDevice flutterDevice = TestFlutterDevice(
       mockDevice,
       <FlutterView>[],
-      observatoryUris: Stream<Uri>.value(testUri),
+      observatoryUris: <Uri>[ testUri ]
     );
 
     await flutterDevice.connect();
@@ -655,7 +657,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, { Stream<Uri> observatoryUris })
+  TestFlutterDevice(Device device, this.views, { List<Uri> observatoryUris })
     : super(device, buildMode: BuildMode.debug, trackWidgetCreation: false) {
     _observatoryUris = observatoryUris;
   }
@@ -664,8 +666,8 @@ class TestFlutterDevice extends FlutterDevice {
   final List<FlutterView> views;
 
   @override
-  Stream<Uri> get observatoryUris => _observatoryUris;
-  Stream<Uri> _observatoryUris;
+  List<Uri> get observatoryUris => _observatoryUris;
+  List<Uri> _observatoryUris;
 }
 
 class ThrowingForwardingFileSystem extends ForwardingFileSystem {

packages/flutter_tools/test/src/mocks.dart

@@ -533,12 +533,6 @@ 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;
 }
@@ -569,33 +563,16 @@ class MockDeviceLogReader extends DeviceLogReader {
   @override
   String get name => 'MockLogReader';
 
-  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;
-  }
+  final StreamController<String> _linesController = StreamController<String>.broadcast();
 
   @override
   Stream<String> get logLines => _linesController.stream;
 
-  void addLine(String line) {
-    if (_linesController.hasListener) {
-      _linesController.add(line);
-    } else {
-      _lineQueue.add(line);
-    }
-  }
+  void addLine(String line) => _linesController.add(line);
 
   @override
-  Future<void> dispose() async {
-    _lineQueue.clear();
-    await _linesController.close();
+  void dispose() {
+    _linesController.close();
   }
 }