Refactoring of the HTTP library

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.
 
 class _HttpHeaders implements HttpHeaders {
   _HttpHeaders() : _headers = new Map<String, List<String>>();
 
   List<String> operator[](String name) {
     name = name.toLowerCase();
     return _headers[name];
@@ skipping 587 matching lines @@
   String value;
   Date expires;
   int maxAge;
   String domain;
   String path;
   bool httpOnly = false;
   bool secure = false;
 }
 
 
+// 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>();
+
+  void add(_HttpConnectionBase connection) {
+    void closeIfDone() {
+      // We only check for write closed here. This means that we are
+      // not waiting for the client to half-close the socket before
+      // fully closing the socket.
+      if (!connection._isWriteClosed) return;
+      _q.remove(connection);
+      connection._socket.close();
+      if (connection.onClosed != null) connection.onClosed();
+    }
+
+    // If the connection is already fully closed don't insert it into the queue.
+    if (connection._isFullyClosed) {
+      connection._socket.close();
+      if (connection.onClosed != null) connection.onClosed();
+      return;
+    }
+
+    _q.add(connection);
+
+    // If output stream is not closed for writing close it now and
+    // wait for callback when closed.
+    if (!connection._isWriteClosed) {
+      connection._socket.outputStream.close();
+      connection._socket.outputStream.onClosed = () {
+        connection._state |= _HttpConnectionBase.WRITE_CLOSED;
+        closeIfDone();
+      };
+    } else {
+      connection._socket.outputStream.onClosed = () { assert(false); };
+    }
+
+    // If socket is not closed for reading wait for callback.
+    if (!connection._isReadClosed) {
+      connection._socket.onClosed = () {
+        connection._state |= _HttpConnectionBase.READ_CLOSED;
+        // This is a nop, as we are not using the read closed
+        // information for anything. For both server and client
+        // connections the inbound message have been read to
+        // completion when the socket enters the close queue.
+      };
+    } else {
+      connection._socket.onClosed = () { assert(false); };
+    }
+
+    // Ignore any data on a socket in the close queue.
+    connection._socket.onData = connection._socket.read;
+
+    // If an error occurs immediately close the socket.
+    connection._socket.onError = (e) {
+      connection._state |= _HttpConnectionBase.WRITE_CLOSED;
+      closeIfDone();
+    };
+  }
+
+  void shutdown() {
+    _q.forEach((_HttpConnectionBase connection) {
+      connection._socket.close();
+    });
+  }
+
+  final Set<_HttpConnectionBase> _q;
+}
+
+
 class _HttpRequestResponseBase {
   final int START = 0;
   final int HEADER_SENT = 1;
   final int DONE = 2;
   final int UPGRADED = 3;
 
   _HttpRequestResponseBase(_HttpConnectionBase this._httpConnection)
       : _headers = new _HttpHeaders() {
     _state = START;
     _headResponse = false;
@@ skipping 71 matching lines @@
     bool allWritten = true;
     if (_contentLength < 0) {
       // Terminate the content if transfer encoding is chunked.
       allWritten = _httpConnection._write(_Const.END_CHUNKED);
     } else {
       if (!_headResponse && _bodyBytesWritten < _contentLength) {
         throw new HttpException("Sending less than specified content length");
       }
       assert(_headResponse || _bodyBytesWritten == _contentLength);
     }
-    // If we are done writing the response, and either the client has
-    // closed or the connection is not persistent, we can close. Also
-    // if using HTTP 1.0 and the content length was not known we must
-    // close to indicate end of body.
-    if (!persistentConnection || _httpConnection._closing ||
-        (_protocolVersion == "1.0" && _contentLength < 0)) {
-      _httpConnection._close();
-    }
     return allWritten;
   }
 
   bool _writeHeaders() {
     _headers._mutable = false;
     _headers._write(_httpConnection);
     // Terminate header.
     return _writeCRLF();
   }
 
@@ skipping 289 matching lines @@
     if (_state == START) {
       _writeHeader();
     }
     _state = UPGRADED;
     // Ensure that any trailing data is written.
     _writeDone();
     // Indicate to the connection that the response handling is done.
     return _httpConnection._detachSocket();
   }
 
-  void _responseEnd() {
-    _ensureHeadersSent();
-    _state = DONE;
-    // Stop tracking no pending write events.
-    _httpConnection._onNoPendingWrites = null;
-    // Ensure that any trailing data is written.
-    _writeDone();
-    // Indicate to the connection that the response handling is done.
-    _httpConnection._responseDone();
-  }
-
   // Delegate functions for the HttpOutputStream implementation.
   bool _streamWrite(List<int> buffer, bool copyBuffer) {
     if (_done) throw new HttpException("Response closed");
     return _write(buffer, copyBuffer);
   }
 
   bool _streamWriteFrom(List<int> buffer, int offset, int len) {
     if (_done) throw new HttpException("Response closed");
     return _writeList(buffer, offset, len);
   }
 
   void _streamFlush() {
     _httpConnection._flush();
   }
 
   void _streamClose() {
-    _responseEnd();
+    _ensureHeadersSent();
+    _state = DONE;
+    // Stop tracking no pending write events.
+    _httpConnection._onNoPendingWrites = null;
+    // Ensure that any trailing data is written.
+    _writeDone();
+    // Indicate to the connection that the response handling is done.
+    _httpConnection._responseClosed();
   }
 
   void _streamSetNoPendingWriteHandler(callback()) {
     if (_state != DONE) {
       _httpConnection._onNoPendingWrites = callback;
     }
   }
 
   void _streamSetCloseHandler(callback()) {
     // TODO(sgjesse): Handle this.
@@ skipping 196 matching lines @@
 
   void set onError(void callback(e)) {
     _requestOrResponse._streamSetErrorHandler(callback);
   }
 
   _HttpRequestResponseBase _requestOrResponse;
 }
 
 
 abstract class _HttpConnectionBase {
-  _HttpConnectionBase() : _httpParser = new _HttpParser(),
-                          hashCode = _nextHashCode {
+  static const int IDLE = 0;
+  static const int ACTIVE = 1;
+  static const int REQUEST_DONE = 2;
+  static const int RESPONSE_DONE = 4;
+  static const int ALL_DONE = REQUEST_DONE | RESPONSE_DONE;
+  static const int READ_CLOSED = 8;
+  static const int WRITE_CLOSED = 16;
+  static const int FULLY_CLOSED = READ_CLOSED | WRITE_CLOSED;
+
+  _HttpConnectionBase() : hashCode = _nextHashCode {
     _nextHashCode = (_nextHashCode + 1) & 0xFFFFFFF;
   }
 
+  bool get _isRequestDone => (_state & REQUEST_DONE) == REQUEST_DONE;
+  bool get _isResponseDone => (_state & RESPONSE_DONE) == RESPONSE_DONE;
+  bool get _isAllDone => (_state & ALL_DONE) == ALL_DONE;
+  bool get _isReadClosed => (_state & READ_CLOSED) == READ_CLOSED;
+  bool get _isWriteClosed => (_state & WRITE_CLOSED) == WRITE_CLOSED;
+  bool get _isFullyClosed => (_state & FULLY_CLOSED) == FULLY_CLOSED;
+
   void _connectionEstablished(Socket socket) {
     _socket = socket;
-    // Register handler for socket events.
-    _socket.onData = _onData;
-    _socket.onClosed = _onClosed;
-    _socket.onError = _onError;
+    // Register handlers for socket events. All socket events are
+    // passed to the HTTP parser.
+    _socket.onData = () {
+      List<int> buffer = _socket.read();
+      if (buffer != null) {
+        _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]) {
-    if (!_error && !_closing) {
-      return _socket.outputStream.write(data, copyBuffer);
-    }
+    return _socket.outputStream.write(data, copyBuffer);
   }
 
   bool _writeFrom(List<int> buffer, [int offset, int len]) {
-    if (!_error && !_closing) {
-      return _socket.outputStream.writeFrom(buffer, offset, len);
-    }
+    return _socket.outputStream.writeFrom(buffer, offset, len);
   }
 
   bool _flush() {
     _socket.outputStream.flush();
   }
 
   bool _close() {
-    _closing = true;
     _socket.outputStream.close();
   }
 
   bool _destroy() {
-    _closing = true;
     _socket.close();
   }
 
-  void _onData() {
-    List<int> buffer = _socket.read();
-    if (buffer != null) {
-      _httpParser.writeList(buffer, 0, buffer.length);
-    }
-  }
-
-  void _onClosed() {
-    _closing = true;
-    _onConnectionClosed(null);
-  }
-
-  void _onError(e) {
-    // If an error occurs, make sure to close the socket if one is associated.
-    _error = true;
-    if (_socket != null) {
-      _socket.close();
-    }
-    _onConnectionClosed(e);
-  }
-
   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());
   }
 
   HttpConnectionInfo get connectionInfo {
-    if (_socket == null || _closing || _error) return null;
+    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) { }
     return null;
   }
 
-  void _onConnectionClosed(e);
-  void _responseDone();
-
   void set _onNoPendingWrites(void callback()) {
-    if (!_error) {
-      _socket.outputStream.onNoPendingWrites = callback;
-    }
+    _socket.outputStream.onNoPendingWrites = callback;
   }
 
+  int _state = IDLE;
+
   Socket _socket;
-  bool _closing = false;  // Is the socket closed by the client?
-  bool _error = false;  // Is the socket closed due to an error?
   _HttpParser _httpParser;
 
+  // Callbacks.
   Function onDetach;
+  Function onClosed;
 
   // Hash code for HTTP connection. Currently this is just a counter.
   final int hashCode;
   static int _nextHashCode = 0;
 }
 
 
 // HTTP server connection over a socket.
 class _HttpConnection extends _HttpConnectionBase {
   _HttpConnection(HttpServer this._server) {
+    _httpParser = new _HttpParser.requestParser();
     // Register HTTP parser callbacks.
-    _httpParser.requestStart =
-      (method, uri, version) => _onRequestStart(method, uri, version);
-    _httpParser.responseStart =
-      (statusCode, reasonPhrase, version) =>
-      _onResponseStart(statusCode, reasonPhrase, version);
-    _httpParser.headerReceived =
-        (name, value) => _onHeaderReceived(name, value);
-    _httpParser.headersComplete = () => _onHeadersComplete();
-    _httpParser.dataReceived = (data) => _onDataReceived(data);
-    _httpParser.dataEnd = (close) => _onDataEnd(close);
-    _httpParser.error = (e) => _onError(e);
+    _httpParser.requestStart = _onRequestStart;
+    _httpParser.headerReceived = _onHeaderReceived;
+    _httpParser.headersComplete = _onHeadersComplete;
+    _httpParser.dataReceived = _onDataReceived;
+    _httpParser.dataEnd = _onDataEnd;
+    _httpParser.error = _onError;
+    _httpParser.closed = _onClosed;
+    _httpParser.responseStart = (statusCode, reasonPhrase, version) {
+      assert(false);
+    };
   }
 
-  void _onConnectionClosed(e) {
-    // Don't report errors when HTTP parser is in idle state. Clients
-    // can close the connection and cause a connection reset by peer
-    // error which is OK.
-    if (e != null && !_httpParser.isIdle) {
-      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);
-      }
+  void _onClosed() {
+    _state |= _HttpConnectionBase.READ_CLOSED;
+  }
+
+  void _onError(e) {
+    onError(e);
+    // Propagate the error to the streams.
+    if (_request != null && _request._streamErrorHandler != null) {
+      _request._streamErrorHandler(e);
     }
-
-    // If currently not processing any request close the socket when
-    // we are done writing the response.
-    if (_httpParser.isIdle) {
-      // If the httpParser is idle and we get an error from the
-      // connection we deal with that as a closed connection and not
-      // as an error. When the client disappears we get a connection
-      // reset by peer and that is OK.
-      if (e != null) {
-        onClosed();
-      } else {
-        _socket.outputStream.onClosed = () {
-          _destroy();
-          onClosed();
-        };
-        // If the client closes and we are done writing the response
-        // the connection should be closed.
-        if (_response == null) _close();
-      }
-    } else {
-      // Processing a request.
-      if (e == null) {
-        // Indicate connection close to the HTTP parser.
-        _httpParser.connectionClosed();
-      }
+    if (_response != null && _response._streamErrorHandler != null) {
+      _response._streamErrorHandler(e);
     }
+    if (_socket != null) _socket.close();
   }
 
   void _onRequestStart(String method, String uri, String version) {
+    _state = _HttpConnectionBase.ACTIVE;
     // Create new request and response objects for this request.
     _request = new _HttpRequest(this);
     _response = new _HttpResponse(this);
     _request._onRequestStart(method, uri, version);
     _request._protocolVersion = version;
     _response._protocolVersion = version;
     _response._headResponse = method == "HEAD";
   }
 
-  void _onResponseStart(int statusCode, String reasonPhrase, String version) {
-    // TODO(sgjesse): Error handling.
-  }
-
   void _onHeaderReceived(String name, String value) {
     _request._onHeaderReceived(name, value);
   }
 
   void _onHeadersComplete() {
     _request._onHeadersComplete();
     _response.persistentConnection = _httpParser.persistentConnection;
     if (onRequestReceived != null) {
       onRequestReceived(_request, _response);
     }
   }
 
   void _onDataReceived(List<int> data) {
     _request._onDataReceived(data);
   }
 
+  void _checkDone() {
+    if (_isAllDone) {
+      // If we are done writing the response, and either the client
+      // has closed or the connection is not persistent, we must
+      // close. Also if using HTTP 1.0 and the content length was not
+      // known we must close to indicate end of body.
+      bool close =
+          !_response.persistentConnection ||
+          (_response._protocolVersion == "1.0" && _response._contentLength < 0);
+      _request = null;
+      _response = null;
+      if (_isReadClosed || close) {
+        _server._closeQueue.add(this);
+      } else {
+        _state = _HttpConnectionBase.IDLE;
+      }
+    }
+  }
+
   void _onDataEnd(bool close) {
     _request._onDataEnd();
+    _state |= _HttpConnectionBase.REQUEST_DONE;
+    _checkDone();
   }
 
-  void _responseDone() {
-    // If the connection is closing then close the output stream to
-    // fully close the socket.
-    if (_closing) {
-      _socket.outputStream.onClosed = () {
-        _socket.close();
-        onClosed();
-      };
-    }
-    _response = null;
+  void _responseClosed() {
+    _state |= _HttpConnectionBase.RESPONSE_DONE;
+    _checkDone();
   }
 
   HttpServer _server;
   HttpRequest _request;
   HttpResponse _response;
 
   // Callbacks.
   Function onRequestReceived;
-  Function onClosed;
   Function onError;
 }
 
 
 class _RequestHandlerRegistration {
   _RequestHandlerRegistration(Function this._matcher, Function this._handler);
   Function _matcher;
   Function _handler;
 }
 
 // HTTP server waiting for socket connections. The connections are
 // managed by the server and as requests are received the request.
 class _HttpServer implements HttpServer {
   _HttpServer() : _connections = new Set<_HttpConnection>(),
-                  _handlers = new List<_RequestHandlerRegistration>();
+                  _handlers = new List<_RequestHandlerRegistration>(),
+                  _closeQueue = new _CloseQueue();
 
   void listen(String host, int port, {int backlog: 128}) {
     listenOn(new ServerSocket(host, port, backlog));
     _closeServer = true;
   }
 
   void listenOn(ServerSocket serverSocket) {
     void onConnection(Socket socket) {
       // Accept the client connection.
       _HttpConnection connection = new _HttpConnection(this);
@@ skipping 20 matching lines @@
                     void handler(HttpRequest request, HttpResponse response)) {
     _handlers.add(new _RequestHandlerRegistration(matcher, handler));
   }
 
   void set defaultRequestHandler(
       void handler(HttpRequest request, HttpResponse response)) {
     _defaultHandler = handler;
   }
 
   void close() {
+    _closeQueue.shutdown();
     if (_sessionManagerInstance != null) {
       _sessionManagerInstance.close();
       _sessionManagerInstance = null;
     }
     if (_server != null && _closeServer) {
       _server.close();
     }
     _server = null;
     for (_HttpConnection connection in _connections) {
       connection._destroy();
@@ skipping 50 matching lines @@
     return _sessionManagerInstance;
   }
 
 
   ServerSocket _server;  // The server listen socket.
   bool _closeServer = false;
   Set<_HttpConnection> _connections;  // Set of currently connected clients.
   List<_RequestHandlerRegistration> _handlers;
   Object _defaultHandler;
   Function _onError;
+  _CloseQueue _closeQueue;
   _HttpSessionManager _sessionManagerInstance;
 }
 
 
 class _HttpClientRequest
     extends _HttpRequestResponseBase implements HttpClientRequest {
   _HttpClientRequest(String this._method,
                      Uri this._uri,
                      _HttpClientConnection connection)
       : super(connection) {
@@ skipping 37 matching lines @@
     _httpConnection._flush();
   }
 
   void _streamClose() {
     _ensureHeadersSent();
     _state = DONE;
     // Stop tracking no pending write events.
     _httpConnection._onNoPendingWrites = null;
     // Ensure that any trailing data is written.
     _writeDone();
-    _connection._requestDone();
+    _connection._requestClosed();
   }
 
   void _streamSetNoPendingWriteHandler(callback()) {
     if (_state != DONE) {
       _httpConnection._onNoPendingWrites = callback;
     }
   }
 
   void _streamSetCloseHandler(callback()) {
     // TODO(sgjesse): Handle this.
@@ skipping 94 matching lines @@
     return _cookies;
   }
 
   InputStream get inputStream {
     if (_inputStream == null) {
       _inputStream = new _HttpInputStream(this);
     }
     return _inputStream;
   }
 
-  void _onRequestStart(String method, String uri, String version) {
-    // TODO(sgjesse): Error handling
-  }
-
   void _onResponseStart(int statusCode, String reasonPhrase, String version) {
     _statusCode = statusCode;
     _reasonPhrase = reasonPhrase;
   }
 
   void _onHeaderReceived(String name, String value) {
     _headers.add(name, value);
   }
 
   void _handleUnauthorized() {
@@ skipping 108 matching lines @@
       _connection._onResponse(this);
     }
   }
 
   void _onDataReceived(List<int> data) {
     _buffer.add(data);
     if (_inputStream != null) _inputStream._dataReceived();
   }
 
   void _onDataEnd() {
-    _connection._responseDone();
     if (_inputStream != null) {
       _inputStream._closeReceived();
     } else {
       inputStream._streamMarkedClosed = true;
     }
   }
 
   // Delegate functions for the HttpInputStream implementation.
   int _streamAvailable() {
     return _buffer.length;
@@ skipping 19 matching lines @@
   _HttpClientConnection _connection;
   _HttpInputStream _inputStream;
   _BufferList _buffer;
 
   Function _streamErrorHandler;
 }
 
 
 class _HttpClientConnection
     extends _HttpConnectionBase implements HttpClientConnection {
-  static const int NONE = 0;
-  static const int REQUEST_DONE = 1;
-  static const int RESPONSE_DONE = 2;
-  static const int ALL_DONE = REQUEST_DONE | RESPONSE_DONE;
 
-  _HttpClientConnection(_HttpClient this._client);
+  _HttpClientConnection(_HttpClient this._client) {
+    _httpParser = new _HttpParser.responseParser();
+  }
 
   void _connectionEstablished(_SocketConnection socketConn) {
     super._connectionEstablished(socketConn._socket);
     _socketConn = socketConn;
     // Register HTTP parser callbacks.
-    _httpParser.requestStart =
-      (method, uri, version) => _onRequestStart(method, uri, version);
-    _httpParser.responseStart =
-      (statusCode, reasonPhrase, version) =>
-      _onResponseStart(statusCode, reasonPhrase, version);
-    _httpParser.headerReceived =
-        (name, value) => _onHeaderReceived(name, value);
-    _httpParser.headersComplete = () => _onHeadersComplete();
-    _httpParser.dataReceived = (data) => _onDataReceived(data);
-    _httpParser.dataEnd = (closed) => _onDataEnd(closed);
-    _httpParser.error = (e) => _onError(e);
+    _httpParser.responseStart = _onResponseStart;
+    _httpParser.headerReceived = _onHeaderReceived;
+    _httpParser.headersComplete = _onHeadersComplete;
+    _httpParser.dataReceived = _onDataReceived;
+    _httpParser.dataEnd = _onDataEnd;
+    _httpParser.error = _onError;
+    _httpParser.closed = _onClosed;
+    _httpParser.requestStart = (method, uri, version) { assert(false); };
+    _state = _HttpConnectionBase.ACTIVE;
   }
 
-  bool get _isRequestDone => (_state & REQUEST_DONE) == REQUEST_DONE;
-  bool get _isResponseDone => (_state & RESPONSE_DONE) == RESPONSE_DONE;
-  bool get _isAllDone => (_state & ALL_DONE) == ALL_DONE;
-
   void _checkSocketDone() {
     if (_isAllDone) {
-      if (!_closing) {
+      // If we are done writing the response, and either the server
+      // has closed or the connection is not persistent, we must
+      // close.
+      if (_isReadClosed || !_response.persistentConnection) {
+        this.onClosed = () {
+          _client._closedSocketConnection(_socketConn);
+        };
+        _client._closeQueue.add(this);
+      } else {
         _client._returnSocketConnection(_socketConn);
-      }
-      _socket = null;
-      _socketConn = null;
-      assert(_pendingRedirect == null || _pendingRetry == null);
-      if (_pendingRedirect != null) {
-        _doRedirect(_pendingRedirect);
-        _pendingRedirect = null;
-      } else if (_pendingRetry != null) {
-        _doRetry(_pendingRetry);
-        _pendingRetry = null;
+        _socket = null;
+        _socketConn = null;
+        assert(_pendingRedirect == null || _pendingRetry == null);
+        if (_pendingRedirect != null) {
+          _doRedirect(_pendingRedirect);
+          _pendingRedirect = null;
+        } else if (_pendingRetry != null) {
+          _doRetry(_pendingRetry);
+          _pendingRetry = null;
+        }
       }
     }
   }
 
-  void _requestDone() {
-    _state |= REQUEST_DONE;
+  void _requestClosed() {
+    _state |= _HttpConnectionBase.REQUEST_DONE;
     _checkSocketDone();
   }
 
-  void _responseDone() {
-    if (_closing) {
-      if (_socket != null) {
-        _socket.close();
-      }
-    }
-    _state |= RESPONSE_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);
     _response = new _HttpClientResponse(this);
     return _request;
   }
 
   DetachedSocket detachSocket() {
     return _detachSocket();
   }
 
-  void _onConnectionClosed(e) {
+  void _onClosed() {
+    _state |= _HttpConnectionBase.READ_CLOSED;
+  }
+
+  void _onError(e) {
     // Socket is closed either due to an error or due to normal socket close.
-    if (e != null) {
-      if (_onErrorCallback != null) {
-        _onErrorCallback(e);
-      } else {
-        throw e;
-      }
+    if (_onErrorCallback != null) {
+      _onErrorCallback(e);
+    } else {
+      throw e;
     }
-    _closing = true;
-    if (e != null) {
-      // Propagate the error to the streams.
-      if (_response != null && _response._streamErrorHandler != null) {
-        _response._streamErrorHandler(e);
-      }
-      _responseDone();
-    } else {
-      // If there was no socket error the socket was closed
-      // normally. Indicate closing to the HTTP Parser as there might
-      // still be an HTTP error.
-      _httpParser.connectionClosed();
+    // Propagate the error to the streams.
+    if (_response != null && _response._streamErrorHandler != null) {
+       _response._streamErrorHandler(e);
+    }
+    if (_socketConn != null) {
+      _client._closeSocketConnection(_socketConn);
     }
   }
 
-  void _onRequestStart(String method, String uri, String version) {
-    // TODO(sgjesse): Error handling.
-  }
-
   void _onResponseStart(int statusCode, String reasonPhrase, String version) {
     _response._onResponseStart(statusCode, reasonPhrase, version);
   }
 
   void _onHeaderReceived(String name, String value) {
     _response._onHeaderReceived(name, value);
   }
 
   void _onHeadersComplete() {
     _response._onHeadersComplete();
   }
 
   void _onDataReceived(List<int> data) {
     _response._onDataReceived(data);
   }
 
   void _onDataEnd(bool close) {
-    if (close) _closing = true;
     _response._onDataEnd();
+    _state |= _HttpConnectionBase.RESPONSE_DONE;
+    _checkSocketDone();
   }
 
   void set onRequest(void handler(HttpClientRequest request)) {
     _onRequest = handler;
   }
 
   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 = NONE;
+    _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.
     if (_isAllDone) {
       _doRetry(retry);
     } else {
@@ skipping 23 matching lines @@
     // request are done.
     if (_isAllDone) {
       _doRedirect(redirect);
     } else {
       // Prepare for redirect.
       assert(_pendingRetry == null);
       _pendingRedirect = redirect;
     }
   }
 
-  int _state = NONE;
-
   List<RedirectInfo> get redirects => _redirects;
 
   Function _onRequest;
   Function _onResponse;
   Function _onErrorCallback;
 
   _HttpClient _client;
   _SocketConnection _socketConn;
   HttpClientRequest _request;
   HttpClientResponse _response;
@@ skipping 19 matching lines @@
                     int this._port,
                     Socket this._socket);
 
   void _markReturned() {
     _socket.onData = null;
     _socket.onClosed = null;
     _socket.onError = null;
     _returnTime = new Date.now();
   }
 
+  void _close() {
+    _socket.onData = null;
+    _socket.onClosed = null;
+    _socket.onError = null;
+    _socket.close();
+  }
+
   Duration _idleTime(Date now) => now.difference(_returnTime);
 
   int get hashCode => _socket.hashCode;
 
   String _host;
   int _port;
   Socket _socket;
   Date _returnTime;
 }
 
@@ skipping 49 matching lines @@
   final String host;
   final int port;
   final bool isDirect;
 }
 
 class _HttpClient implements HttpClient {
   static const int DEFAULT_EVICTION_TIMEOUT = 60000;
 
   _HttpClient() : _openSockets = new Map(),
                   _activeSockets = new Set(),
+                  _closeQueue = new _CloseQueue(),
                   credentials = new List<_Credentials>(),
                   _shutdown = false;
 
   HttpClientConnection open(
       String method, String host, int port, String path) {
     // TODO(sgjesse): The path set here can contain both query and
     // fragment. They should be cracked and set correctly.
     return _open(method, new Uri.fromComponents(
         scheme: "http", domain: host, port: port, path: path));
   }
@@ skipping 38 matching lines @@
   }
 
   void addCredentials(
       Uri url, String realm, HttpClientCredentials cr) {
     credentials.add(new _Credentials(url, realm, cr));
   }
 
   set findProxy(String f(Uri uri)) => _findProxy = f;
 
   void shutdown() {
-     _openSockets.forEach((String key, Queue<_SocketConnection> connections) {
-       while (!connections.isEmpty) {
-         _SocketConnection socketConn = connections.removeFirst();
-         socketConn._socket.close();
-       }
-     });
-     _activeSockets.forEach((_SocketConnection socketConn) {
-       socketConn._socket.close();
-     });
+    _closeQueue.shutdown();
+    _openSockets.forEach((String key, Queue<_SocketConnection> connections) {
+      while (!connections.isEmpty) {
+        _SocketConnection socketConn = connections.removeFirst();
+        socketConn._socket.close();
+      }
+    });
+    _activeSockets.forEach((_SocketConnection socketConn) {
+      socketConn._socket.close();
+    });
     if (_evictionTimer != null) _cancelEvictionTimer();
      _shutdown = true;
   }
 
   void _cancelEvictionTimer() {
     _evictionTimer.cancel();
     _evictionTimer = null;
   }
 
   String _connectionKey(String host, int port) {
@@ skipping 175 matching lines @@
         if (_openSockets.isEmpty) _cancelEvictionTimer();
       }
       _evictionTimer = new Timer.repeating(10000, _handleEviction);
     }
 
     // Return connection.
     _activeSockets.remove(socketConn);
     sockets.addFirst(socketConn);
   }
 
+  void _closeSocketConnection(_SocketConnection socketConn) {
+    socketConn._close();
+    _activeSockets.remove(socketConn);
+  }
+
+  void _closedSocketConnection(_SocketConnection socketConn) {
+    _activeSockets.remove(socketConn);
+  }
+
   _Credentials _findCredentials(Uri url, [_AuthenticationScheme scheme]) {
     // Look for credentials.
     _Credentials cr =
         credentials.reduce(null, (_Credentials prev, _Credentials value) {
           if (value.applies(url, scheme)) {
             if (prev == null) return value;
             return value.uri.path.length > prev.uri.path.length ? value : prev;
           } else {
             return prev;
           }
         });
     return cr;
   }
 
   void _removeCredentials(_Credentials cr) {
     int index = credentials.indexOf(cr);
     if (index != -1) {
       credentials.removeAt(index);
     }
   }
 
   Function _onOpen;
   Map<String, Queue<_SocketConnection>> _openSockets;
   Set<_SocketConnection> _activeSockets;
+  _CloseQueue _closeQueue;
   List<_Credentials> credentials;
   Timer _evictionTimer;
   Function _findProxy;
   Function _authenticate;
   bool _shutdown;  // Has this HTTP client been shutdown?
 }
 
 
 class _HttpConnectionInfo implements HttpConnectionInfo {
   String remoteHost;
@@ skipping 115 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;
 }