Abort analysis server integration tests in the event of async errors.

pkg/analysis_server/test/integration/integration_tests.dart

Index: pkg/analysis_server/test/integration/integration_tests.dart
diff --git a/pkg/analysis_server/test/integration/integration_tests.dart b/pkg/analysis_server/test/integration/integration_tests.dart
index de0af377fad10b105dd3c25c43727f8139ba8990..585b7ac7208b78fc0e205d287d0521125f4b5b6e 100644
--- a/pkg/analysis_server/test/integration/integration_tests.dart
+++ b/pkg/analysis_server/test/integration/integration_tests.dart
@@ -88,7 +88,8 @@ abstract class AbstractAnalysisServerIntegrationTest {
   Future get analysisFinished {
     Completer completer = new Completer();
     StreamSubscription subscription;
-    subscription = server.onNotification(SERVER_STATUS).listen((params) {
+    subscription = server.onNotification(SERVER_STATUS).listen(wrapAsync(
+        (params) {
       bool analysisComplete = false;
       try {
         analysisComplete = !params['analysis']['analyzing'];
@@ -101,7 +102,7 @@ abstract class AbstractAnalysisServerIntegrationTest {
         subscription.cancel();
       }
       expect(params, isServerStatusParams);
-    });
+    }));
     return completer.future;
   }
 
@@ -121,11 +122,11 @@ abstract class AbstractAnalysisServerIntegrationTest {
     sourceDirectory = Directory.systemTemp.createTempSync('analysisServer');
     return Server.start().then((Server server) {
       this.server = server;
-      server.onNotification(ANALYSIS_ERRORS).listen((params) {
+      server.onNotification(ANALYSIS_ERRORS).listen(wrapAsync((params) {
         expect(params, isMap);
         expect(params['file'], isString);
         currentAnalysisErrors[params['file']] = params['errors'];
-      });
+      }));
     });
   }
 
@@ -510,7 +511,7 @@ class Server {
     return Process.start(dartBinary, arguments).then((Process process) {
       Server server = new Server._(process);
       process.stdout.transform((new Utf8Codec()).decoder).transform(
-          new LineSplitter()).listen((String line) {
+          new LineSplitter()).listen(wrapAsync((String line) {
         String trimmedLine = line.trim();
         server._recordStdio('RECV: $trimmedLine');
         var message = JSON.decoder.convert(trimmedLine);
@@ -556,10 +557,10 @@ class Server {
           // event of an error.
           expect(message, isNotification);
         }
-      });
-      process.stderr.listen((List<int> data) {
+      }));
+      process.stderr.listen(wrapAsync((List<int> data) {
         fail('Unexpected output from stderr');
-      });
+      }));
       return server;
     });
   }
@@ -622,3 +623,12 @@ class Server {
     _recordedStdio.add(line);
   }
 }
+
+/**
+ * Wrap a callback function that will be called an unknown number of times.
+ * This won't prevent the integration test from completing, but it will ensure
+ * that any exceptions thrown from the callback are properly reported as unit
+ * test failures.
+ */
+Function wrapAsync(Function callback) => expectAsync(callback, count: 0, max: -1
+    );