Rewrite parts of http-parser, to better handle connection errors and pausing.

sdk/lib/io/http_parser.dart

 // Copyright (c) 2013, 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.
 
 part of dart.io;
 
 // Global constants.
 class _Const {
   // Bytes for "HTTP".
   static const HTTP = const [72, 84, 84, 80];
@@ skipping 179 matching lines @@
   factory _HttpParser.requestParser() {
     return new _HttpParser._(true);
   }
 
   factory _HttpParser.responseParser() {
     return new _HttpParser._(false);
   }
 
   _HttpParser._(this._requestParser) {
     _controller = new StreamController<_HttpIncoming>(
-          onListen: _updateParsePauseState,
-          onPause: _updateParsePauseState,
-          onResume: _updateParsePauseState,
-          onCancel: _updateParsePauseState);
+          onListen: () {
+            _socketSubscription.resume();
+          },
+          onPause: () {
+            _paused = true;
+            _pauseStateChanged();
+          },
+          onResume: () {
+            _paused = false;
+            _pauseStateChanged();
+          },
+          onCancel: () {
+            try {
+              _socketSubscription.cancel();
+            } catch (e) {
+            }
+          });
     _reset();
   }
 
 
   StreamSubscription<_HttpIncoming> listen(void onData(_HttpIncoming event),
                                            {void onError(error),
                                             void onDone(),
                                             bool cancelOnError}) {
     return _controller.stream.listen(onData,
                                      onError: onError,
                                      onDone: onDone,
                                      cancelOnError: cancelOnError);
   }
 
   Future<_HttpParser> addStream(Stream<List<int>> stream) {
     // Listen to the stream and handle data accordingly. When a
     // _HttpIncoming is created, _dataPause, _dataResume, _dataDone is
     // given to provide a way of controlling the parser.
     // TODO(ajohnsen): Remove _dataPause, _dataResume and _dataDone and clean up
     // how the _HttpIncoming signals the parser.
     var completer = new Completer();
     _socketSubscription = stream.listen(
         _onData,
         onError: _onError,
         onDone: () {
           completer.complete(this);
         });
+    _socketSubscription.pause();
     return completer.future;
   }
 
   Future<_HttpParser> close() {
     _onDone();
     return new Future.value(this);
   }
 
   void _parse() {
     try {
@@ skipping 19 matching lines @@
     if (_state == _State.CLOSED) {
       throw new HttpParserException("Data on closed connection");
     }
     if (_state == _State.FAILURE) {
       throw new HttpParserException("Data on failed connection");
     }
     while (_buffer != null &&
            _index < _buffer.length &&
            _state != _State.FAILURE &&
            _state != _State.UPGRADED) {
-      if (_paused) {
+      if (_bodyPaused) {

[Søren Gjesse, 2013/05/17 07:09:28]

Don't we need to check for both _paused and _bodyPaused? Without we just go
ahead and parse the whole header evne though the stream of incoming is paused -
maybe that's fine - but could deserve a comment.

[Anders Johnsen, 2013/05/17 08:27:59]

Done.

         _parserCalled = false;
         return;
       }
       int byte = _buffer[_index++];
       switch (_state) {
         case _State.START:
           if (byte == _Const.HTTP[0]) {
             // Start parsing method or HTTP version.
             _httpVersionIndex = 1;
             _state = _State.METHOD_OR_RESPONSE_HTTP_VERSION;
@@ skipping 274 matching lines @@
                     new String.fromCharCodes(_uri_or_reason_phrase));
           } else {
             _incoming.statusCode = _statusCode;
             _incoming.reasonPhrase =
                 new String.fromCharCodes(_uri_or_reason_phrase);
           }
           _method_or_status_code.clear();
           _uri_or_reason_phrase.clear();
           if (_connectionUpgrade) {
             _incoming.upgraded = true;
+            _parserCalled = false;
             _controller.add(_incoming);
-            break;
+            return;
           }
           if (_transferLength == 0 ||
               (_messageType == _MessageType.RESPONSE &&
                (_noMessageBody || _responseToMethod == "HEAD"))) {
             _reset();
             var tmp = _incoming;
             _closeIncoming();
             _controller.add(tmp);
             break;
           } else if (_chunked) {
             _state = _State.CHUNK_SIZE;
             _remainingContent = 0;
           } else if (_transferLength > 0) {
             _remainingContent = _transferLength;
             _state = _State.BODY;
           } else {
             // Neither chunked nor content length. End of body
             // indicated by close.
             _state = _State.BODY;
           }
+          _parserCalled = false;
           _controller.add(_incoming);
-          break;
+          return;
 
         case _State.CHUNK_SIZE_STARTING_CR:
           _expect(byte, _CharCode.CR);
           _state = _State.CHUNK_SIZE_STARTING_LF;
           break;
 
         case _State.CHUNK_SIZE_STARTING_LF:
           _expect(byte, _CharCode.LF);
           _state = _State.CHUNK_SIZE;
           break;
@@ skipping 258 matching lines @@
     } else if (0x61 <= byte && byte <= 0x66) {
       return byte - 0x61 + 10;  // a - f
     } else {
       throw new HttpParserException("Failed to parse HTTP");
     }
   }
 
   void _createIncoming(int transferLength) {
     assert(_incoming == null);
     assert(_bodyController == null);
+    assert(!_bodyPaused);
+    var incoming;
     _bodyController = new StreamController<List<int>>(
-        onListen: _bodySubscriptionStateChange,
-        onPause: _updateParsePauseState,
-        onResume: _updateParsePauseState,
-        onCancel: _bodySubscriptionStateChange);
-    _incoming = new _HttpIncoming(
+        onListen: () {
+          if (incoming != _incoming) return;
+          assert(_bodyPaused);
+          _bodyPaused = false;
+          _pauseStateChanged();
+        },
+        onPause: () {
+          if (incoming != _incoming) return;
+          assert(!_bodyPaused);
+          _bodyPaused = true;
+          _pauseStateChanged();
+        },
+        onResume: () {
+          if (incoming != _incoming) return;
+          assert(_bodyPaused);
+          _bodyPaused = false;
+          _pauseStateChanged();
+        },
+        onCancel: () {
+          if (incoming != _incoming) return;
+          if (_socketSubscription != null) {
+            _socketSubscription.cancel();
+          }
+          _closeIncoming();
+          _controller.close();
+        });
+    incoming = _incoming = new _HttpIncoming(
         _headers, transferLength, _bodyController.stream);
-    _pauseParsing();  // Needed to handle detaching - don't start on the body!
+    _bodyPaused = true;
+    _pauseStateChanged();
   }
 
   void _closeIncoming() {
     // Ignore multiple close (can happend in re-entrance).
     if (_incoming == null) return;
     var tmp = _incoming;
+    tmp.close();
     _incoming = null;
-    tmp.close();
     if (_bodyController != null) {
       _bodyController.close();
       _bodyController = null;
     }
-    _updateParsePauseState();
+    _bodyPaused = false;
+    _pauseStateChanged();
   }
 
-  void _continueParsing() {
-    _paused = false;
-    if (!_parserCalled && _buffer != null) _parse();
-  }
+  void _pauseStateChanged() {
+    void update(bool pause) {
+      if (pause && !_socketSubscription.isPaused) {
+        _socketSubscription.pause();
+      } else if (!pause && _socketSubscription.isPaused) {
+        _socketSubscription.resume();
+      }
+    }
 
-  void _pauseParsing() {
-    _paused = true;
-  }
-
-  void _bodySubscriptionStateChange() {
-    if (_incoming != null && !_bodyController.hasListener) {
-      _closeIncoming();
-    } else {
-      _updateParsePauseState();
-    }
-  }
-
-  void _updateParsePauseState() {
-    if (_bodyController != null) {
-      if (_bodyController.hasListener && !_bodyController.isPaused) {
-        _continueParsing();
-      } else {
-        _pauseParsing();
+    if (_incoming != null) {
+      if (!_bodyPaused && !_parserCalled) {
+        _parse();
       }
     } else {
-      if (_controller.hasListener && !_controller.isPaused) {
-        _continueParsing();
-      } else {
-        _pauseParsing();
+      if (!_paused && !_parserCalled) {
+        _parse();
       }
     }
   }
 
   void error(error, [stackTrace]) {
     if (_socketSubscription != null) _socketSubscription.cancel();
     _state = _State.FAILURE;
     _controller.addError(error, stackTrace);
     _controller.close();
   }
@@ skipping 23 matching lines @@
 
   bool _noMessageBody;
   String _responseToMethod;  // Indicates the method used for the request.
   int _remainingContent;
 
   _HttpHeaders _headers;
 
   // The current incoming connection.
   _HttpIncoming _incoming;
   StreamSubscription _socketSubscription;
-  bool _paused = false;
+  bool _paused = true;
+  bool _bodyPaused = false;
   Completer _pauseCompleter;
   StreamController<_HttpIncoming> _controller;
   StreamController<List<int>> _bodyController;
 }
 
 
 class HttpParserException implements Exception {
   const HttpParserException([String this.message = ""]);
   String toString() => "HttpParserException: $message";
   final String message;
 }