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.  closed.

[Mads Ager (google), 2012/11/12 09:03:29]

Finish comment.

[Søren Gjesse, 2012/11/12 12:36:38]

Done.

+class _CloseQueue {
+  _CloseQueue() : _q = new Set<_HttpConnectionBase>();
+
+  void add(_HttpConnectionBase connection) {
+    // 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);
+
+    void checkFullyClosed() {

[Mads Ager (google), 2012/11/12 09:03:29]

Maybe move the local function declaration to the top of the function so we have
first declarations then code?

Since the function only checks for write closed, maybe we should rename this to
closeIfDone or something?

[Søren Gjesse, 2012/11/12 12:36:38]

Done.

+      // 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. There seems to be not need for that.

[Mads Ager (google), 2012/11/12 09:03:29]

not -> no

What happens if we get the write close first and then a read close. Then we will
call checkFullyClosed twice and call close on the socket twice? Or can't we get
read close after write close?

[Søren Gjesse, 2012/11/12 12:36:38]

I have turned the read closed handler into a nop. As far as I can see we don't
care about write close. For both server and client connections the inbound
message have been read to completion when the socket enters the close queue. If
read closed have not been seen already we do expect it, but will still close
when writing is done.

+      if (!connection._isWriteClosed) return;
+      _q.remove(connection);
+      connection._socket.close();
+      if (connection.onClosed != null) connection.onClosed();
+    }
+
+    // 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;
+        checkFullyClosed();
+      };
+    } 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;
+        checkFullyClosed();
+      };
+    } 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.READ_CLOSED;
+      connection._state |= _HttpConnectionBase.WRITE_CLOSED;
+      checkFullyClosed();
+    };
+  }
+
+  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;
 }