Handle crashing engine.

When the engine dies unexpectedly during test execution, we have to
terminate any tests running in that engine. Previously, they would just
hang. For some reason that I was never able to satisfactorily explain,
the WebSocket doesn't die in a way I can detect in this case. So
instead, we hand in a future that we only complete when we detect the
server subprocess ends.

packages/flutter_tools/lib/src/test/json_socket.dart

@@ -7,11 +7,12 @@ import 'dart:convert';
 import 'dart:io';
 
 class JSONSocket {
-  JSONSocket(WebSocket socket)
+  JSONSocket(WebSocket socket, this.unusualTermination)
     : _socket = socket, stream = socket.map(JSON.decode).asBroadcastStream();
 
   final WebSocket _socket;
   final Stream stream;
+  final Future unusualTermination;
 
   void send(dynamic data) {
     _socket.add(JSON.encode(data));

packages/flutter_tools/lib/src/test/loader.dart

@@ -90,6 +90,7 @@ void main() {
 ''');
 
   Completer<Iterable<RemoteTest>> completer = new Completer<Iterable<RemoteTest>>();
+  Completer deathCompleter = new Completer();
 
   Process process = await _startProcess(
     listenerFile.path,
@@ -138,6 +139,7 @@ void main() {
         if (kExpectAllTestsToCloseCleanly && output != '')
           print('Unexpected failure after test claimed to pass:\n$output');
       }
+      deathCompleter.complete();
     } catch (e) {
       // Throwing inside this block causes all kinds of hard-to-debug issues
       // like stack overflows and hangs. So catch everything just in case.
@@ -145,7 +147,7 @@ void main() {
     }
   });
 
-  JSONSocket socket = new JSONSocket(await info.socket);
+  JSONSocket socket = new JSONSocket(await info.socket, deathCompleter.future);
 
   await cleanupTempDirectory();
 

packages/flutter_tools/lib/src/test/remote_test.dart

@@ -4,6 +4,7 @@
 
 import 'dart:async';
 
+import 'package:stack_trace/stack_trace.dart';
 import 'package:test/src/backend/live_test.dart';
 import 'package:test/src/backend/live_test_controller.dart';
 import 'package:test/src/backend/metadata.dart';
@@ -29,8 +30,8 @@ class RemoteTest extends Test {
     StreamSubscription subscription;
 
     controller = new LiveTestController(suite, this, () async {
-      controller.setState(const State(Status.running, Result.success));
 
+      controller.setState(const State(Status.running, Result.success));
       _socket.send({'command': 'run', 'index': _index});
 
       subscription = _socket.stream.listen((message) {
@@ -51,6 +52,17 @@ class RemoteTest extends Test {
           controller.completer.complete();
         }
       });
+
+      _socket.unusualTermination.then((_) {
+        if (subscription != null) {
+          controller.addError(new Exception('Unexpected subprocess termination.'), new Trace.current());
+          controller.setState(new State(Status.complete, Result.error));
+          subscription.cancel();
+          subscription = null;
+          controller.completer.complete();
+        }
+      });
+
     }, () async {
       _socket.send({'command': 'close'});
       if (subscription != null) {