Fix a number of HTTP issues

sdk/lib/io/http_impl.dart

 // Copyright (c) 2012, 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.
 
 // The close queue handles graceful closing of HTTP connections. When
 // a connection is added to the queue it will enter a wait state
 // waiting for all data written and possibly socket shutdown from
 // peer.
 class _CloseQueue {
   _CloseQueue() : _q = new Set<_HttpConnectionBase>();
@@ skipping 740 matching lines @@
         _httpParser.streamData(buffer);
       }
     };
     _socket.onClosed = _httpParser.streamDone;
     _socket.onError = _httpParser.streamError;
     // Ignore errors in the socket output stream as this is getting
     // the same errors as the socket itself.
     _socket.outputStream.onError = (e) => null;
   }
 
-  bool _write(List<int> data, [bool copyBuffer = false]) {
-    return _socket.outputStream.write(data, copyBuffer);
-  }
-
-  bool _writeFrom(List<int> buffer, [int offset, int len]) {
-    return _socket.outputStream.writeFrom(buffer, offset, len);
-  }
-
-  bool _flush() {
-    _socket.outputStream.flush();
-  }
-
-  bool _close() {
-    _socket.outputStream.close();
-  }
-
-  bool _destroy() {
-    _socket.close();
-  }
-
-  DetachedSocket _detachSocket() {
-    _socket.onData = null;
-    _socket.onClosed = null;
-    _socket.onError = null;
-    _socket.outputStream.onNoPendingWrites = null;
-    Socket socket = _socket;
-    _socket = null;
-    if (onDetach != null) onDetach();
-    return new _DetachedSocket(socket, _httpParser.readUnparsedData());
-  }
+  bool _write(List<int> data, [bool copyBuffer = false]);
+  bool _writeFrom(List<int> buffer, [int offset, int len]);
+  bool _flush();
+  bool _close();
+  bool _destroy();
+  DetachedSocket _detachSocket();
 
   HttpConnectionInfo get connectionInfo {
     if (_socket == null) return null;
     try {
       _HttpConnectionInfo info = new _HttpConnectionInfo();
       info.remoteHost = _socket.remoteHost;
       info.remotePort = _socket.remotePort;
       info.localPort = _socket.port;
       return info;
     } catch (e) { }
@@ skipping 27 matching lines @@
     _httpParser.requestStart = _onRequestReceived;
     _httpParser.dataReceived = _onDataReceived;
     _httpParser.dataEnd = _onDataEnd;
     _httpParser.error = _onError;
     _httpParser.closed = _onClosed;
     _httpParser.responseStart = (statusCode, reasonPhrase, version) {
       assert(false);
     };
   }
 
+  void _bufferData(List<int> data, [bool copyBuffer = false]) {
+    if (_buffer == null) _buffer = new _BufferList();
+    if (copyBuffer) data = data.getRange(0, data.length);
+    _buffer.add(data);
+  }
+
+  void _releaseBuffer() {

[Mads Ager (google), 2012/12/05 16:02:18]

Maybe name this something like _writeBufferedResponse to make it clear that it
writes and does not just drop the buffer?

[Søren Gjesse, 2012/12/06 17:23:08]

Done.

+    if (_buffer != null) {
+      while (!_buffer.isEmpty) {
+        var data = _buffer.first;
+        _socket.outputStream.write(data, false);
+        _buffer.removeBytes(data.length);
+      }
+      _buffer = null;
+    }
+  }
+
+  bool _write(List<int> data, [bool copyBuffer = false]) {
+    if (_isRequestDone) {
+      return _socket.outputStream.write(data, copyBuffer);
+    } else {
+      _bufferData(data, copyBuffer);
+      return false;
+    }
+  }
+
+  bool _writeFrom(List<int> data, [int offset, int len]) {
+    if (_isRequestDone) {
+      return _socket.outputStream.writeFrom(data, offset, len);
+    } else {
+      if (offset == null) offset = 0;
+      if (len == null) len = buffer.length - offset;
+      _bufferData(data.getRange(offset, len), false);
+      return false;
+    }
+  }
+
+  bool _flush() {
+    _socket.outputStream.flush();
+  }
+
+  bool _close() {
+    _socket.outputStream.close();
+  }
+
+  bool _destroy() {
+    _socket.close();
+  }
+
   void _onClosed() {
     _state |= _HttpConnectionBase.READ_CLOSED;
     _checkDone();
   }
 
+  DetachedSocket _detachSocket() {
+    _socket.onData = null;
+    _socket.onClosed = null;
+    _socket.onError = null;
+    _socket.outputStream.onNoPendingWrites = null;
+    _releaseBuffer();
+    Socket socket = _socket;
+    _socket = null;
+    if (onDetach != null) onDetach();
+    return new _DetachedSocket(socket, _httpParser.readUnparsedData());
+  }
+
   void _onError(e) {
     onError(e);
     // Propagate the error to the streams.
     if (_request != null && _request._streamErrorHandler != null) {
       _request._streamErrorHandler(e);
     }
     if (_response != null && _response._streamErrorHandler != null) {
       _response._streamErrorHandler(e);
     }
     if (_socket != null) _socket.close();
@@ skipping 33 matching lines @@
       bool close =
           !_response.persistentConnection ||
           (_response._protocolVersion == "1.0" && _response._contentLength < 0);
       _request = null;
       _response = null;
       if (close) {
         _server._closeQueue.add(this);
       } else {
         _state = _HttpConnectionBase.IDLE;
       }
+    } else if (_isResponseDone) {
+      // If the response is closed before the request is fully read
+      // close this connection. If there is buffered output
+      // (e.g. error response for invalid request where the server did
+      // not care to read the request body) this is send.
+      assert(!_isRequestDone);
+      _releaseBuffer();
+      _close();
+      _server._closeQueue.add(this);
     }
   }
 
   void _onDataEnd(bool close) {
+    // Start sending queued response if any.
+    _releaseBuffer();
+    _state |= _HttpConnectionBase.REQUEST_DONE;
     _request._onDataEnd();
-    _state |= _HttpConnectionBase.REQUEST_DONE;
-    _checkDone();
   }
 
   void _responseClosed() {
     _state |= _HttpConnectionBase.RESPONSE_DONE;
     _checkDone();
   }
 
   HttpServer _server;
   HttpRequest _request;
   HttpResponse _response;
 
+  // Buffer for data written before full response have been processed.

[Mads Ager (google), 2012/12/05 16:02:18]

have -> has

[Søren Gjesse, 2012/12/06 17:23:08]

Done.

+  _BufferList _buffer;
+
   // Callbacks.
   Function onRequestReceived;
   Function onError;
 }
 
 
 class _RequestHandlerRegistration {
   _RequestHandlerRegistration(Function this._matcher, Function this._handler);
   Function _matcher;
   Function _handler;
@@ skipping 190 matching lines @@
     if (_done) throw new HttpException("Request closed");
     if (_outputStream == null) {
       _outputStream = new _HttpOutputStream(this);
     }
     return _outputStream;
   }
 
   // Delegate functions for the HttpOutputStream implementation.
   bool _streamWrite(List<int> buffer, bool copyBuffer) {
     if (_done) throw new HttpException("Request closed");
+    _emptyBody = false;

[Mads Ager (google), 2012/12/05 16:02:18]

Should this be 

_emptyBody = buffer.length != 0 || _emptyBody;

?

[Søren Gjesse, 2012/12/06 17:23:08]

Good point, however it should be

_emptyBody = buffer.length == 0 && _emptyBody;

     return _write(buffer, copyBuffer);
   }
 
   bool _streamWriteFrom(List<int> buffer, int offset, int len) {
     if (_done) throw new HttpException("Request closed");
+    _emptyBody = false;

[Mads Ager (google), 2012/12/05 16:02:18]

Ditto

[Søren Gjesse, 2012/12/06 17:23:08]

Done.

     return _writeList(buffer, offset, len);
   }
 
   void _streamFlush() {
     _httpConnection._flush();
   }
 
   void _streamClose() {
     _ensureHeadersSent();
     _state = _HttpRequestResponseBase.DONE;
@@ skipping 70 matching lines @@
     // Write headers.
     _writeHeaders();
     _state = _HttpRequestResponseBase.HEADER_SENT;
   }
 
   String _method;
   Uri _uri;
   _HttpClientConnection _connection;
   _HttpOutputStream _outputStream;
   Function _streamErrorHandler;
+  bool _emptyBody = true;
 }
 
 
 class _HttpClientResponse
     extends _HttpRequestResponseBase implements HttpClientResponse {
   _HttpClientResponse(_HttpClientConnection connection)
       : super(connection) {
     _connection = connection;
   }
 
@@ skipping 188 matching lines @@
 }
 
 
 class _HttpClientConnection
     extends _HttpConnectionBase implements HttpClientConnection {
 
   _HttpClientConnection(_HttpClient this._client) {
     _httpParser = new _HttpParser.responseParser();
   }
 
+  bool _write(List<int> data, [bool copyBuffer = false]) {
+    return _socket.outputStream.write(data, copyBuffer);
+  }
+
+  bool _writeFrom(List<int> data, [int offset, int len]) {
+    return _socket.outputStream.writeFrom(data, offset, len);
+  }
+
+  bool _flush() {
+    _socket.outputStream.flush();
+  }
+
+  bool _close() {
+    _socket.outputStream.close();
+  }
+
+  bool _destroy() {
+    _socket.close();
+  }
+
+  DetachedSocket _detachSocket() {
+    _socket.onData = null;
+    _socket.onClosed = null;
+    _socket.onError = null;
+    _socket.outputStream.onNoPendingWrites = null;
+    Socket socket = _socket;
+    _socket = null;
+    if (onDetach != null) onDetach();
+    return new _DetachedSocket(socket, _httpParser.readUnparsedData());
+  }
+
   void _connectionEstablished(_SocketConnection socketConn) {
     super._connectionEstablished(socketConn._socket);
     _socketConn = socketConn;
     // Register HTTP parser callbacks.
     _httpParser.responseStart = _onResponseReceived;
     _httpParser.dataReceived = _onDataReceived;
     _httpParser.dataEnd = _onDataEnd;
     _httpParser.error = _onError;
     _httpParser.closed = _onClosed;
     _httpParser.requestStart = (method, uri, version) { assert(false); };
@@ skipping 28 matching lines @@
 
   void _requestClosed() {
     _state |= _HttpConnectionBase.REQUEST_DONE;
     _checkSocketDone();
   }
 
   HttpClientRequest open(String method, Uri uri) {
     _method = method;
     // Tell the HTTP parser the method it is expecting a response to.
     _httpParser.responseToMethod = method;
-    _request = new _HttpClientRequest(method, uri, this);
+    // If the connection already have a request this is a retry of a

[Mads Ager (google), 2012/12/05 16:02:18]

have -> has

[Søren Gjesse, 2012/12/06 17:23:08]

Done.

+    // request. In this case the request object is reused to ensure
+    // that the same headers are send.
+    if (_request != null) {
+      _request._method = method;
+      _request._uri = uri;
+      _request._headers._mutable = true;
+      _request._state = _HttpRequestResponseBase.START;
+    } else {
+      _request = new _HttpClientRequest(method, uri, this);
+    }
     _response = new _HttpClientResponse(this);
     return _request;
   }
 
   DetachedSocket detachSocket() {
     return _detachSocket();
   }
 
   void _onClosed() {
     _state |= _HttpConnectionBase.READ_CLOSED;
     _checkSocketDone();
   }
 
   void _onError(e) {
     // Cancel any pending data in the HTTP parser.
     _httpParser.cancel();
     if (_socketConn != null) {
       _client._closeSocketConnection(_socketConn);
     }
-    // Report the error.
-    if (_response != null && _response._streamErrorHandler != null) {
-       _response._streamErrorHandler(e);
-    } else if (_onErrorCallback != null) {
-      _onErrorCallback(e);
+
+    // If it looks as if we got a bad connection from the connection
+    // pool and the request can be retried do a retry.
+    if (_socketConn != null && _socketConn._fromPool && _request._emptyBody) {
+      String method = _request._method;
+      Uri uri = _request._uri;
+      _socketConn = null;
+
+      // Retry the URL using the same connection instance.
+      _httpParser.restart();
+      _client._openUrl(method, uri, this);
     } else {
-      throw e;
+      // Report the error.
+      if (_response != null && _response._streamErrorHandler != null) {
+         _response._streamErrorHandler(e);
+      } else if (_onErrorCallback != null) {
+        _onErrorCallback(e);
+      } else {
+        throw e;
+      }
     }
   }
 
   void _onResponseReceived(int statusCode,
                            String reasonPhrase,
                            String version,
                            _HttpHeaders headers) {
     _response._onResponseReceived(statusCode, reasonPhrase, version, headers);
   }
 
@@ skipping 20 matching lines @@
   void set onResponse(void handler(HttpClientResponse response)) {
     _onResponse = handler;
   }
 
   void set onError(void callback(e)) {
     _onErrorCallback = callback;
   }
 
   void _doRetry(_RedirectInfo retry) {
     assert(_socketConn == null);
-    _request = null;
-    _response = null;
 
     // Retry the URL using the same connection instance.
     _state = _HttpConnectionBase.IDLE;
     _client._openUrl(retry.method, retry.location, this);
   }
 
   void _retry() {
     var retry = new _RedirectInfo(_response.statusCode, _method, _request._uri);
     // The actual retry is postponed until both response and request
     // are done.
@@ skipping 54 matching lines @@
 
 
 // Class for holding keep-alive sockets in the cache for the HTTP
 // client together with the connection information.
 class _SocketConnection {
   _SocketConnection(String this._host,
                     int this._port,
                     Socket this._socket);
 
   void _markReturned() {
-    _socket.onData = null;
-    _socket.onClosed = null;
-    _socket.onError = null;
+    // Any activity on the socket while waiting in the pool will
+    // invalidate the connection os that it is not reused.

[Mads Ager (google), 2012/12/05 16:02:18]

os -> so

[Søren Gjesse, 2012/12/06 17:23:08]

Done.

+    _socket.onData = _invalidate;
+    _socket.onClosed = _invalidate;
+    _socket.onError = (_) => _invalidate();
     _returnTime = new Date.now();
     _httpClientConnection = null;
   }
 
+  void _markRetreived() {

[Mads Ager (google), 2012/12/05 16:02:18]

_markRetrieved

[Søren Gjesse, 2012/12/06 17:23:08]

Done.

+    _socket.onData = null;
+    _socket.onClosed = null;
+    _socket.onError = null;
+    _httpClientConnection = null;
+  }
+
   void _close() {
     _socket.onData = null;
     _socket.onClosed = null;
     _socket.onError = null;
     _httpClientConnection = null;
     _socket.close();
   }
 
   Duration _idleTime(Date now) => now.difference(_returnTime);
 
+  bool get _fromPool => _returnTime != null;
+
+  void _invalidate() {
+    _valid = false;
+    _close();
+  }
+
   int get hashCode => _socket.hashCode;
 
   String _host;
   int _port;
   Socket _socket;
   Date _returnTime;
+  bool _valid = true;
   HttpClientConnection _httpClientConnection;
 }
 
 class _ProxyConfiguration {
   static const String PROXY_PREFIX = "PROXY ";
   static const String DIRECT_PREFIX = "DIRECT";
 
   _ProxyConfiguration(String configuration) : proxies = new List<_Proxy>() {
     if (configuration == null) {
       throw new HttpException("Invalid proxy configuration $configuration");
@@ skipping 189 matching lines @@
         connectPort = port;
       } else {
         connectHost = proxy.host;
         connectPort = proxy.port;
       }
 
       // If there are active connections for this key get the first one
       // otherwise create a new one.
       String key = _connectionKey(connectHost, connectPort);
       Queue socketConnections = _openSockets[key];
-      // Remove active connections that are of the wrong type (HTTP or HTTPS).
+      // Remove active connections that are not valid any more or of
+      // the wrong type (HTTP or HTTPS).
       while (socketConnections != null &&
              !socketConnections.isEmpty &&
-             secure != (socketConnections.first._socket is SecureSocket)) {
-        socketConnection.removeFirst()._close();
+             (!socketConnections.first._valid ||
+              secure != (socketConnections.first._socket is SecureSocket))) {
+        socketConnections.removeFirst()._close();
       }
       if (socketConnections == null || socketConnections.isEmpty) {
         Socket socket = secure ? new SecureSocket(connectHost, connectPort) :
                                  new Socket(connectHost, connectPort);
         // Until the connection is established handle connection errors
         // here as the HttpClientConnection object is not yet associated
         // with the socket.
         socket.onError = (e) {
           proxyIndex++;
           if (proxyIndex < proxyConfiguration.proxies.length) {
@@ skipping 12 matching lines @@
           // HttpClientConnection object which will be associated with
           // the connected socket.
           socket.onError = null;
           _SocketConnection socketConn =
               new _SocketConnection(connectHost, connectPort, socket);
           _activeSockets.add(socketConn);
           _connectionOpened(socketConn, connection, !proxy.isDirect);
         };
       } else {
         _SocketConnection socketConn = socketConnections.removeFirst();
+        socketConn._markRetreived();
         _activeSockets.add(socketConn);
         new Timer(0, (ignored) =>
                   _connectionOpened(socketConn, connection, !proxy.isDirect));
 
         // Get rid of eviction timer if there are no more active connections.
         if (socketConnections.isEmpty) _openSockets.remove(key);
         if (_openSockets.isEmpty) _cancelEvictionTimer();
       }
     }
 
@@ skipping 253 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;
 }