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

pkg/analysis_server/test/integration/integration_tests.dart

 // Copyright (c) 2014, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 library test.integration.analysis;
 
 import 'dart:async';
 import 'dart:collection';
 import 'dart:convert';
 import 'dart:io';
@@ skipping 70 matching lines @@
    * received from the server with 'analyzing' set to false.
    *
    * The future will only be completed by 'server.status' notifications that are
    * received after this function call.  So it is safe to use this getter
    * multiple times in one test; each time it is used it will wait afresh for
    * analysis to finish.
    */
   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'];
       } catch (_) {
         // Status message was mal-formed or missing optional parameters.  That's
         // fine, since we'll detect a mal-formed status message below.
       }
       if (analysisComplete) {
         completer.complete(params);
         subscription.cancel();
       }
       expect(params, isServerStatusParams);
-    });
+    }));
     return completer.future;
   }
 
   /**
    * Print out any messages exchanged with the server.  If some messages have
    * already been exchanged with the server, they are printed out immediately.
    */
   void debugStdio() {
     server.debugStdio();
   }
 
   /**
    * The server is automatically started before every test, and a temporary
    * [sourceDirectory] is created.
    */
   Future setUp() {
     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'];
-      });
+      }));
     });
   }
 
   /**
    * After every test, the server stopped and [sourceDirectory] is deleted.
    */
   Future tearDown() {
     return server.kill().then((_) {
       sourceDirectory.deleteSync(recursive: true);
     });
@@ skipping 364 matching lines @@
     String serverPath = normalize(join(dirname(Platform.script.path), '..',
         '..', 'bin', 'server.dart'));
     List<String> arguments = [];
     if (debugServer) {
       arguments.add('--debug');
     }
     arguments.add(serverPath);
     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);
         expect(message, isMap);
         Map messageAsMap = message;
         if (messageAsMap.containsKey('id')) {
           expect(messageAsMap['id'], isString);
           String id = message['id'];
           Completer completer = server._pendingCommands[id];
           if (completer == null) {
@@ skipping 25 matching lines @@
           StreamController notificationController =
               server._notificationControllers[event];
           if (notificationController != null) {
             notificationController.add(messageAsMap['params']);
           }
           // Check that the message is well-formed.  We do this after calling
           // notificationController.add() so that we don't stall the test in the
           // 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;
     });
   }
 
   /**
    * Stop the server.
    */
   Future kill() {
     _process.kill();
     return _process.exitCode;
@@ skipping 42 matching lines @@
    * Record a message that was exchanged with the server, and print it out if
    * [debugStdio] has been called.
    */
   void _recordStdio(String line) {
     if (_debuggingStdio) {
       print(line);
     }
     _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
+    );