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

sdk/lib/io/http_impl.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;
 
 class _HttpIncoming extends Stream<List<int>> {
   final int _transferLength;
   final Completer _dataCompleter = new Completer();
   Stream<List<int>> _stream;
@@ skipping 1053 matching lines @@
 }
 
 class _HttpClientConnection {
   final String key;
   final Socket _socket;
   final bool _proxyTunnel;
   final _HttpParser _httpParser;
   StreamSubscription _subscription;
   final _HttpClient _httpClient;
   bool _dispose = false;
+  bool closed = false;

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

Private? (also key above).

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

This is used from another class (HttpClient). Also, this class is never exposed
to the user. Same with key.

 
   Completer<_HttpIncoming> _nextResponseCompleter;
   Future _streamFuture;
 
   _HttpClientConnection(String this.key,
                         Socket this._socket,
                         _HttpClient this._httpClient,
                         [this._proxyTunnel = false])
       : _httpParser = new _HttpParser.responseParser() {
     _socket.pipe(_httpParser);
 
     // Set up handlers on the parser here, so we are sure to get 'onDone' from
     // the parser.
     _subscription = _httpParser.listen(
         (incoming) {
           // Only handle one incoming response at the time. Keep the
           // stream paused until the response have been processed.
           _subscription.pause();
           // We assume the response is not here, until we have send the request.

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

Shouldn't this be an error instead of an assert? A server not conforming to the
protocol should not hit an assert.

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

Done.

           assert(_nextResponseCompleter != null);

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

Do we still need this temp variable after the completion changed to be always
async?

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

Done.

           var completer = _nextResponseCompleter;
           _nextResponseCompleter = null;
           completer.complete(incoming);
         },
         onError: (error) {
           if (_nextResponseCompleter != null) {
             _nextResponseCompleter.completeError(error);
             _nextResponseCompleter = null;
           }
         },
         onDone: () {
+          if (_nextResponseCompleter != null) {
+            _nextResponseCompleter.completeError(new HttpException(
+                "Connection closed before response was received"));
+            _nextResponseCompleter = null;
+          }
           close();
         });
   }
 
   _HttpClientRequest send(Uri uri, int port, String method, _Proxy proxy) {
+    if (closed) {
+      throw new HttpException("Socket closed before request was sent");
+    }
     // Start with pausing the parser.
     _subscription.pause();
     _ProxyCredentials proxyCreds;  // Credentials used to authorize proxy.
     _SiteCredentials creds;  // Credentials used to authorize this request.
     var outgoing = new _HttpOutgoing(_socket);
     // Create new request object, wrapping the outgoing connection.
     var request = new _HttpClientRequest(outgoing,
                                          uri,
                                          method,
                                          proxy,
@@ skipping 100 matching lines @@
         });
     return request;
   }
 
   Future<Socket> detachSocket() {
     return _streamFuture.then(
         (_) => new _DetachedSocket(_socket, _httpParser.detachIncoming()));
   }
 
   void destroy() {
+    closed = true;
     _httpClient._connectionClosed(this);
     _socket.destroy();
   }
 
   void close() {
+    closed = true;
     _httpClient._connectionClosed(this);
     _streamFuture
           // TODO(ajohnsen): Add timeout.
         .then((_) => _socket.destroy());
   }
 
   Future<_HttpClientConnection> createProxyTunnel(host, port, proxy) {
     _HttpClientRequest request =
         send(new Uri.fromComponents(domain: host, port: port),
              port,
@@ skipping 155 matching lines @@
       // TODO(sgjesse): Keep a map of these as normally only a few
       // configuration strings will be used.
       try {
         proxyConf = new _ProxyConfiguration(_findProxy(uri));
       } catch (error, stackTrace) {
         return new Future.error(error, stackTrace);
       }
     }
     return _getConnection(uri.domain, port, proxyConf, isSecure)
         .then((info) {
-          return info.connection.send(uri,
-                                      port,
-                                      method.toUpperCase(),
-                                      info.proxy);
+          send(info) {
+            return info.connection.send(uri,
+                                        port,
+                                        method.toUpperCase(),
+                                        info.proxy);
+          }
+          // If the connection was closed before the request was sent, create
+          // and use another connection.
+          if (info.connection.closed) {
+            return _getConnection(uri.domain, port, proxyConf, isSecure)
+                .then(send);
+          }
+          return send(info);
         });
   }
 
   Future<HttpClientRequest> _openUrlFromRequest(String method,
                                                 Uri uri,
                                                 _HttpClientRequest previous) {
     // If the new URI is relative (to either '/' or some sub-path),
     // construct a full URI from the previous one.
     // See http://tools.ietf.org/html/rfc3986#section-4.2
     replaceComponents({scheme, domain, port, path}) {
@@ skipping 811 matching lines @@
 
 
 class _RedirectInfo implements RedirectInfo {
   const _RedirectInfo(int this.statusCode,
                       String this.method,
                       Uri this.location);
   final int statusCode;
   final String method;
   final Uri location;
 }