Coalesce analysis server output messages into a single write call.

pkg/analysis_server/lib/src/channel/byte_stream_channel.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 channel.byte_stream;
 
 import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
@@ skipping 42 matching lines @@
     return responseStream.firstWhere((Response response) => response.id == id);
   }
 }
 
 /**
  * Instances of the class [ByteStreamServerChannel] implement a
  * [ServerCommunicationChannel] that uses a stream and a sink (typically,
  * standard input and standard output) to communicate with clients.
  */
 class ByteStreamServerChannel implements ServerCommunicationChannel {
+  /**
+   * Value of [_outputState] indicating that there is no outstanding data in
+   * [_pendingOutput], and that the most recent flush of [_output] has
+   * completed.
+   */
+  static const int _STATE_IDLE = 0;
+
+  /**
+   * Value of [_outputState] indicating that there is outstanding data in
+   * [_pendingOutput], and that the most recent flush of [_output] has
+   * completed; therefore a microtask has been scheduled to send the data.
+   */
+  static const int _STATE_MICROTASK_PENDING = 1;
+
+  /**
+   * Value of [_outputState] indicating that data has been sent to the
+   * [_output] stream and flushed, but the flush has not completed, so we must
+   * wait for it to complete before sending more data.  There may or may not be
+   * outstanding data in [_pendingOutput].
+   */
+  static const int _STATE_FLUSH_PENDING = 2;
+
   final Stream input;
-  final IOSink output;
+
+  final IOSink _output;
 
   /**
    * Completer that will be signalled when the input stream is closed.
    */
   final Completer _closed = new Completer();
 
-  ByteStreamServerChannel(this.input, this.output);
+  /**
+   * State of the output stream (see constants above).
+   */
+  int _outputState = _STATE_IDLE;
+
+  /**
+   * List of strings that need to be sent to [_output] at the next available
+   * opportunity.
+   */
+  List<String> _pendingOutput = <String>[];
+
+  /**
+   * True if [close] has been called.
+   */
+  bool _closeRequested = false;
+
+  ByteStreamServerChannel(this.input, this._output);
 
   /**
    * Future that will be completed when the input stream is closed.
    */
   Future get closed {
     return _closed.future;
   }
 
   @override
   void close() {
-    output.flush().then((_) {
-      if (!_closed.isCompleted) {
+    if (!_closeRequested) {
+      _closeRequested = true;
+      if (_outputState == _STATE_IDLE) {
+        assert(!_closed.isCompleted);
         _closed.complete();
+      } else {
+        // Nothing to do.  [_flushCompleted] will call _closed.complete() after
+        // the flush completes.
       }
-    });
+    }
   }
 
   @override
   void listen(void onRequest(Request request), {Function onError, void
       onDone()}) {
     input.transform(
         (new Utf8Codec()).decoder).transform(
             new LineSplitter()).listen(
                 (String data) => _readRequest(data, onRequest),
                 onError: onError,
                 onDone: () {
       close();
       onDone();
     });
   }
 
   @override
   void sendNotification(Notification notification) {
     // Don't send any further notifications after the communication channel is
     // closed.
-    if (_closed.isCompleted) {
+    if (_closeRequested) {
       return;
     }
     ServerCommunicationChannel.ToJson.start();
     String jsonEncoding = JSON.encode(notification.toJson());
     ServerCommunicationChannel.ToJson.stop();
-    output.write(jsonEncoding + '\n');
+    _outputLine(jsonEncoding);
   }
 
   @override
   void sendResponse(Response response) {
     // Don't send any further responses after the communication channel is
     // closed.
-    if (_closed.isCompleted) {
+    if (_closeRequested) {
       return;
     }
     ServerCommunicationChannel.ToJson.start();
     String jsonEncoding = JSON.encode(response.toJson());
     ServerCommunicationChannel.ToJson.stop();
-    output.write(jsonEncoding + '\n');
+    _outputLine(jsonEncoding);
   }
 
   /**
+   * Callback invoked after a flush of [_output] completes.  Closes the stream
+   * if necessary.  Otherwise schedules additional pending output.
+   */
+  void _flushCompleted(_) {
+    assert(_outputState == _STATE_FLUSH_PENDING);
+    if (_pendingOutput.isNotEmpty) {
+      _output.write(_pendingOutput.join());
+      _output.flush().then(_flushCompleted);
+      _pendingOutput.clear();
+      // Since we've done another flush, stay in _STATE_FLUSH_PENDING.
+    } else {
+      _outputState = _STATE_IDLE;
+      if (_closeRequested) {
+        assert(!_closed.isCompleted);
+        _closed.complete();
+      }
+    }
+  }
+
+  /**
+   * Microtask that writes pending output to the output stream and flushes it.
+   */
+  void _microtask() {
+    assert(_outputState == _STATE_MICROTASK_PENDING);
+    _output.write(_pendingOutput.join());
+    _output.flush().then(_flushCompleted);
+    _pendingOutput.clear();
+    _outputState = _STATE_FLUSH_PENDING;
+  }
+
+  /**
+   * Send the string [s] to [_output] followed by a newline.
+   */
+  void _outputLine(String s) {
+    _pendingOutput.add(s);
+    _pendingOutput.add('\n');
+    if (_outputState == _STATE_IDLE) {
+      // Don't send the output just yet; schedule a microtask to do it, so that
+      // if caller decides to output additional lines, they will get sent in
+      // the same call to _output.write().
+      new Future.microtask(_microtask);
+      _outputState = _STATE_MICROTASK_PENDING;
+    }
+  }
+
+  /**
    * Read a request from the given [data] and use the given function to handle
    * the request.
    */
   void _readRequest(Object data, void onRequest(Request request)) {
     // Ignore any further requests after the communication channel is closed.
     if (_closed.isCompleted) {
       return;
     }
     // Parse the string as a JSON descriptor and process the resulting
     // structure as a request.
     ServerCommunicationChannel.FromJson.start();
     Request request = new Request.fromString(data);
     ServerCommunicationChannel.FromJson.stop();
     if (request == null) {
       sendResponse(new Response.invalidRequestFormat());
       return;
     }
     onRequest(request);
   }
 }