Add an integration test to verify that server I/O is asynchronous.

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 34 matching lines @@
   HashMap<String, List<AnalysisError>> currentAnalysisErrors =
       new HashMap<String, List<AnalysisError>>();
 
   /**
    * True if the teardown process should skip sending a "server.shutdown"
    * request (e.g. because the server is known to have already shutdown).
    */
   bool skipShutdown = false;
 
   /**
-   * Data associated with the "server.connected" notification that was received
-   * when the server started up.
-   */
-  var serverConnectedParams;
-
-  /**
    * True if we are currently subscribed to [SERVER_STATUS] updates.
    */
   bool _subscribedToServerStatus = false;
 
   AbstractAnalysisServerIntegrationTest() {
     initializeInttestMixin();
   }
 
   /**
    * Write a source file with the given absolute [pathname] and [contents].
@@ skipping 78 matching lines @@
     sourceDirectory = Directory.systemTemp.createTempSync('analysisServer');
 
     onAnalysisErrors.listen((AnalysisErrorsParams params) {
       currentAnalysisErrors[params.file] = params.errors;
     });
     Completer serverConnected = new Completer();
     onServerConnected.listen((_) {
       expect(serverConnected.isCompleted, isFalse);
       serverConnected.complete();
     });
-    return server.start(dispatchNotification).then((params) {
-      serverConnectedParams = params;
+    return server.start().then((_) {
+      server.listenToOutput(dispatchNotification);
       server.exitCode.then((_) {
         skipShutdown = true;
       });
       return serverConnected.future;
     });
   }
 
   /**
    * After every test, the server is stopped and [sourceDirectory] is deleted.
    */
@@ skipping 510 matching lines @@
       }
       pathname = parent;
     }
     return dirname(pathname);
   }
 
   /**
    * Start the server.  If [debugServer] is true, the server will be started
    * with "--debug", allowing a debugger to be attached.
    */
-  Future start(NotificationProcessor notificationProcessor, {bool debugServer:
-      false}) {
+  Future start({bool debugServer: false}) {
     if (_process != null) {
       throw new Exception('Process already started');
     }
     _time.start();
-    // TODO(paulberry): move the logic for finding the script, the dart
-    // executable, and the package root into a shell script.
     String dartBinary = Platform.executable;
     String rootDir =
         findRoot(Platform.script.toFilePath(windows: Platform.isWindows));
     String serverPath = normalize(join(rootDir, 'bin', 'server.dart'));
     List<String> arguments = [];
     if (debugServer) {
       arguments.add('--debug');
     }
     if (Platform.packageRoot.isNotEmpty) {
       arguments.add('--package-root=${Platform.packageRoot}');
     }
     arguments.add('--checked');
     arguments.add(serverPath);
     return Process.start(dartBinary, arguments).then((Process process) {
       _process = process;
-      process.stdout.transform(
-          (new Utf8Codec()).decoder).transform(new LineSplitter()).listen((String line) {
-        String trimmedLine = line.trim();
-        _recordStdio('RECV: $trimmedLine');
-        var message;
-        try {
-          message = JSON.decoder.convert(trimmedLine);
-        } catch (exception) {
-          _badDataFromServer();
-          return;
-        }
-        expect(message, isMap);
-        Map messageAsMap = message;
-        if (messageAsMap.containsKey('id')) {
-          expect(messageAsMap['id'], isString);
-          String id = message['id'];
-          Completer completer = _pendingCommands[id];
-          if (completer == null) {
-            fail('Unexpected response from server: id=$id');
-          } else {
-            _pendingCommands.remove(id);
-          }
-          if (messageAsMap.containsKey('error')) {
-            // TODO(paulberry): propagate the error info to the completer.
-            completer.completeError(
-                new UnimplementedError(
-                    'Server responded with an error: ${JSON.encode(message)}'));
-          } else {
-            completer.complete(messageAsMap['result']);
-          }
-          // Check that the message is well-formed.  We do this after calling
-          // completer.complete() or completer.completeError() so that we don't
-          // stall the test in the event of an error.
-          expect(message, isResponse);
-        } else {
-          // Message is a notification.  It should have an event and possibly
-          // params.
-          expect(messageAsMap, contains('event'));
-          expect(messageAsMap['event'], isString);
-          notificationProcessor(messageAsMap['event'], 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.transform(
-          (new Utf8Codec()).decoder).transform(new LineSplitter()).listen((String line) {
-        String trimmedLine = line.trim();
-        _recordStdio('ERR:  $trimmedLine');
-        _badDataFromServer();
-      });
       process.exitCode.then((int code) {
         _recordStdio('TERMINATED WITH EXIT CODE $code');
         if (code != 0) {
           _badDataFromServer();
         }
       });
     });
   }
 
   /**
+   * Start listening to output from the server, and deliver notifications to
+   * [notificationProcessor].
+   */
+  void listenToOutput(NotificationProcessor notificationProcessor) {
+    _process.stdout.transform(
+        (new Utf8Codec()).decoder).transform(new LineSplitter()).listen((String line) {
+      String trimmedLine = line.trim();
+      _recordStdio('RECV: $trimmedLine');
+      var message;
+      try {
+        message = JSON.decoder.convert(trimmedLine);
+      } catch (exception) {
+        _badDataFromServer();
+        return;
+      }
+      expect(message, isMap);
+      Map messageAsMap = message;
+      if (messageAsMap.containsKey('id')) {
+        expect(messageAsMap['id'], isString);
+        String id = message['id'];
+        Completer completer = _pendingCommands[id];
+        if (completer == null) {
+          fail('Unexpected response from server: id=$id');
+        } else {
+          _pendingCommands.remove(id);
+        }
+        if (messageAsMap.containsKey('error')) {
+          // TODO(paulberry): propagate the error info to the completer.
+          completer.completeError(
+              new UnimplementedError(
+                  'Server responded with an error: ${JSON.encode(message)}'));
+        } else {
+          completer.complete(messageAsMap['result']);
+        }
+        // Check that the message is well-formed.  We do this after calling
+        // completer.complete() or completer.completeError() so that we don't
+        // stall the test in the event of an error.
+        expect(message, isResponse);
+      } else {
+        // Message is a notification.  It should have an event and possibly
+        // params.
+        expect(messageAsMap, contains('event'));
+        expect(messageAsMap['event'], isString);
+        notificationProcessor(messageAsMap['event'], 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.transform(
+        (new Utf8Codec()).decoder).transform(new LineSplitter()).listen((String line) {
+      String trimmedLine = line.trim();
+      _recordStdio('ERR:  $trimmedLine');
+      _badDataFromServer();
+    });
+  }
+
+  /**
    * Future that completes when the server process exits.
    */
   Future<int> get exitCode => _process.exitCode;
 
   /**
    * Stop the server.
    */
   Future kill() {
     debugStdio();
     _recordStdio('PROCESS FORCIBLY TERMINATED');
@@ skipping 34 matching lines @@
     if (_debuggingStdio) {
       return;
     }
     _debuggingStdio = true;
     for (String line in _recordedStdio) {
       print(line);
     }
   }
 
   /**
+   * Return a future that will complete when all commands that have been sent
+   * to the server so far have been flushed to the OS buffer.
+   */
+  Future flushCommands() {
+    return _process.stdin.flush();
+  }
+
+  /**
    * Deal with bad data received from the server.
    */
   void _badDataFromServer() {
     if (_receivedBadDataFromServer) {
       // We're already dealing with it.
       return;
     }
     _receivedBadDataFromServer = true;
     debugStdio();
     // Give the server 1 second to continue outputting bad data before we kill
@@ skipping 12 matching lines @@
    */
   void _recordStdio(String line) {
     double elapsedTime = _time.elapsedTicks / _time.frequency;
     line = "$elapsedTime: $line";
     if (_debuggingStdio) {
       print(line);
     }
     _recordedStdio.add(line);
   }
 }