| Index: pkg/analysis_server/lib/src/channel/byte_stream_channel.dart
|
| diff --git a/pkg/analysis_server/lib/src/channel/byte_stream_channel.dart b/pkg/analysis_server/lib/src/channel/byte_stream_channel.dart
|
| index d1dfed4451810f1dd57da67ab8b97e18ef5e7722..2953701525d1a2f54fc159b5039bf9b825cc99e4 100644
|
| --- a/pkg/analysis_server/lib/src/channel/byte_stream_channel.dart
|
| +++ b/pkg/analysis_server/lib/src/channel/byte_stream_channel.dart
|
| @@ -60,15 +60,54 @@ class ByteStreamClientChannel implements ClientCommunicationChannel {
|
| * 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.
|
| @@ -79,11 +118,16 @@ class ByteStreamServerChannel implements ServerCommunicationChannel {
|
|
|
| @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
|
| @@ -104,26 +148,72 @@ class ByteStreamServerChannel implements ServerCommunicationChannel {
|
| 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;
|
| + }
|
| }
|
|
|
| /**
|
|
|