Refactor the close and error handling of HTTP connections

runtime/bin/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 _HttpRequestResponseBase {
   _HttpRequestResponseBase(_HttpConnectionBase this._httpConnection)
       : _contentLength = -1,
         _keepAlive = false,
         _headers = new Map();
 
@@ skipping 242 matching lines @@
     if (_outputStream == null) {
       // Ensure that headers are written.
       if (_state == START) {
         _writeHeader();
       }
       _outputStream = new _HttpOutputStream(this);
     }
     return _outputStream;
   }
 
+  void _responseEnd() {
+    _state = DONE;
+    // Stop tracking no pending write events.
+    _httpConnection.outputStream.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) {
     return _write(buffer, copyBuffer);
   }
 
   bool _streamWriteFrom(List<int> buffer, int offset, int len) {
     return _writeList(buffer, offset, len);
   }
 
   void _streamClose() {
-    _httpConnection._phase = _HttpConnectionBase.PHASE_IDLE;
-    _state = DONE;
-    // Stop tracking no pending write events.
-    _httpConnection._onNoPendingWrites = null;
-    // Ensure that any trailing data is written.
-    _writeDone();
-    // If the connection is closing then close the output stream to
-    // fully close the socket.
-    if (_httpConnection._closing) {
-      _httpConnection._close();
-    }
+    _responseEnd();
   }
 
   void _streamSetNoPendingWriteHandler(callback()) {
     if (_state != DONE) {
       _httpConnection._onNoPendingWrites = callback;
     }
   }
 
   void _streamSetCloseHandler(callback()) {
     // TODO(sgjesse): Handle this.
@@ skipping 168 matching lines @@
 
   void set onError(void callback(Exception e)) {
     _requestOrResponse._streamSetErrorHandler(callback);
   }
 
   _HttpRequestResponseBase _requestOrResponse;
 }
 
 
 class _HttpConnectionBase implements Hashable {
-  static final int PHASE_IDLE = 0;
-  static final int PHASE_REQUEST = 1;
-  static final int PHASE_RESPONSE = 2;
-
-  _HttpConnectionBase() : _phase = PHASE_IDLE,
-                          _sendBuffers = new Queue(),
+  _HttpConnectionBase() : _sendBuffers = new Queue(),
                           _httpParser = new _HttpParser();
 
   void _connectionEstablished(Socket socket) {
     _socket = socket;
     // Register handler for socket events.
     _socket.onData = _onData;
     _socket.onClosed = _onClosed;
     _socket.onError = _onError;
   }
 
   bool _write(List<int> data, [bool copyBuffer = false]) {
     if (!_error) {

[Anders Johnsen, 2012/04/10 08:18:53]

I think we should add "&& !_closing".

       return _socket.outputStream.write(data, copyBuffer);
     }
   }
 
   bool _writeFrom(List<int> buffer, [int offset, int len]) {
     if (!_error) {

[Anders Johnsen, 2012/04/10 08:18:53]

Same here.

       return _socket.outputStream.writeFrom(buffer, offset, len);
     }
   }
 
   bool _close() {
     _socket.close();
   }
 
   void _onData() {
     int available = _socket.available();
     if (available == 0) {
       return;
     }
 
     ByteArray buffer = new ByteArray(available);
     int bytesRead = _socket.readList(buffer, 0, available);
     if (bytesRead > 0) {
       int parsed = _httpParser.writeList(buffer, 0, bytesRead);
       if (parsed != bytesRead) {
         // TODO(sgjesse): Error handling.
         _socket.close();
       }
     }
   }
 
   void _onClosed() {

[Anders Johnsen, 2012/04/10 08:18:53]

And a "_closing = true;" here.

-    if (_phase != PHASE_IDLE) {
-      // Client closed socket for writing. Socket should still be open
-      // for writing the response.
-      _closing = true;
-    } else {
-      // The connection is currently not used by any request just close it.
-      _socket.close();
-    }
-    if (_onDisconnectCallback != null) _onDisconnectCallback();
+    _onConnectionClosed(null);
   }
 
   void _onError(Exception e) {
     // If an error occurs, make sure to close the socket if one is associated.
     _error = true;
     if (_socket != null) {
       _socket.close();
     }
-    if (_onErrorCallback != null) {
-      _onErrorCallback(e);
-    }
-    _propagateError(e);
+    _onConnectionClosed(e);
   }
 
-  abstract void _propagateError(Exception e);
-
-  void set onDisconnect(void callback()) {
-    _onDisconnectCallback = callback;
-  }
-
-  void set onError(void callback(Exception e)) {
-    _onErrorCallback = callback;
-  }
+  abstract void _onConnectionClosed(Exception e);
+  abstract void _responseDone();
 
   void set _onNoPendingWrites(void callback()) {
     if (!_error) {
       _socket.outputStream.onNoPendingWrites = callback;
     }
   }
 
   int hashCode() => _socket.hashCode();
 
-  int _phase;
   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;
 
   Queue _sendBuffers;
-
-  Function _onDisconnectCallback;
-  Function _onErrorCallback;
 }
 
 
 // HTTP server connection over a socket.
 class _HttpConnection extends _HttpConnectionBase {
   _HttpConnection(HttpServer this._server) {
     // Register HTTP parser callbacks.
     _httpParser.requestStart =
         (method, uri) => _onRequestStart(method, uri);
     _httpParser.responseStart =
         (statusCode, reasonPhrase) =>
             _onResponseStart(statusCode, reasonPhrase);
     _httpParser.headerReceived =
         (name, value) => _onHeaderReceived(name, value);
     _httpParser.headersComplete = () => _onHeadersComplete();
     _httpParser.dataReceived = (data) => _onDataReceived(data);
-    _httpParser.dataEnd = () => _onDataEnd();
+    _httpParser.dataEnd = (close) => _onDataEnd(close);
     _httpParser.error = (e) => _onError(e);
   }
 
+  void _onConnectionClosed(Exception e) {
+    if (e != null && onError != null) {
+      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 currently not processing any request just close the socket.
+    if (_httpParser.isIdle) {
+      _socket.close();
+      if (onClosed != null && e == null) {
+        // Don't call onClosed if onError has been called.
+        onClosed();
+      }
+      return;
+    }
+
+    // Processing a request.
+    if (e == null) {
+      // Indicate connection close to the HTTP parser.
+      _httpParser.connectionClosed();
+      _closing = true;

[Anders Johnsen, 2012/04/10 08:18:53]

And remove it here, since we mark it in super class.

+    }
+  }
+
   void _onRequestStart(String method, String uri) {
     // Create new request and response objects for this request.
-    _phase = PHASE_REQUEST;
     _request = new _HttpRequest(this);
     _response = new _HttpResponse(this);
     _request._onRequestStart(method, uri);
   }
 
   void _onResponseStart(int statusCode, String reasonPhrase) {
     // TODO(sgjesse): Error handling.
   }
 
   void _onHeaderReceived(String name, String value) {
     _request._onHeaderReceived(name, value);
   }
 
   void _onHeadersComplete() {
     _request._onHeadersComplete();
     _response.keepAlive = _httpParser.keepAlive;
-    if (requestReceived != null) {
-      requestReceived(_request, _response);
+    if (onRequestReceived != null) {
+      onRequestReceived(_request, _response);
     }
   }
 
   void _onDataReceived(List<int> data) {
     _request._onDataReceived(data);
   }
 
-  void _onDataEnd() {
-    // Phase might already have gone to PHASE_IDLE if the response is
-    // sent without waiting for request body.
-    if (_phase == PHASE_REQUEST) {
-      _phase = PHASE_RESPONSE;
+  void _onDataEnd(bool close) {
+    if (_request != null) {
+      _request._onDataEnd();
     }
-    _request._onDataEnd();
+    _request = null;
   }
 
-  void _propagateError(Exception e) {
-    if (_request != null && _request._streamErrorHandler != null) {
-      _request._streamErrorHandler(e);
+  void _responseDone() {
+    // If the connection is closing then close the output stream to
+    // fully close the socket.
+    if (_closing) {
+      outputStream.close();
     }
-    if (_response != null && _response._streamErrorHandler != null) {
-      _response._streamErrorHandler(e);
-    }
+    _response = null;
   }
 
   HttpServer _server;
   HttpRequest _request;
   HttpResponse _response;
 
   // Callbacks.
-  var requestReceived;
+  Function onRequestReceived;
+  Function onClosed;
+  Function onError;
 }
 
 
 // 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>();
 
   void listen(String host, int port, [int backlog = 5]) {
     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);
-      connection.requestReceived = _onRequest;
-      connection.onDisconnect = () => _connections.remove(connection);
+      connection._connectionEstablished(socket);
+      connection.onRequestReceived = _onRequest;
+      connection.onClosed = () => _connections.remove(connection);
       connection.onError = (e) {
+        _connections.remove(connection);
         if (_onError != null) _onError(e);
       };
       connection._connectionEstablished(socket);
       _connections.add(connection);
     }
     serverSocket.onConnection = onConnection;
     _server = serverSocket;
     _closeServer = false;
   }
 
@@ skipping 110 matching lines @@
   bool _streamWriteFrom(List<int> buffer, int offset, int len) {
     return _writeList(buffer, offset, len);
   }
 
   void _streamClose() {
     _state = DONE;
     // Stop tracking no pending write events.
     _httpConnection._onNoPendingWrites = null;
     // Ensure that any trailing data is written.
     _writeDone();
-    // If the connection is closing then close the output stream to
-    // fully close the socket.
-    if (_httpConnection._closing) {
-      _httpConnection._close();
-    }
   }
 
   void _streamSetNoPendingWriteHandler(callback()) {
     if (_state != DONE) {
       _httpConnection._onNoPendingWrites = callback;
     }
   }
 
   void _streamSetCloseHandler(callback()) {
     // TODO(sgjesse): Handle this.
@@ skipping 88 matching lines @@
     }
   }
 
   void _onDataReceived(List<int> data) {
     _buffer.add(data);
     if (_inputStream != null) _inputStream._dataReceived();
   }
 
   void _onDataEnd() {
     if (_inputStream != null) _inputStream._closeReceived();
+    _connection._responseDone();
   }
 
   // Delegate functions for the HttpInputStream implementation.
   int _streamAvailable() {
     return _buffer.length;
   }
 
   List<int> _streamRead(int bytesToRead) {
     return _buffer.readBytes(bytesToRead);
   }
@@ skipping 28 matching lines @@
     // Register HTTP parser callbacks.
     _httpParser.requestStart =
         (method, uri) => _onRequestStart(method, uri);
     _httpParser.responseStart =
         (statusCode, reasonPhrase) =>
             _onResponseStart(statusCode, reasonPhrase);
     _httpParser.headerReceived =
         (name, value) => _onHeaderReceived(name, value);
     _httpParser.headersComplete = () => _onHeadersComplete();
     _httpParser.dataReceived = (data) => _onDataReceived(data);
-    _httpParser.dataEnd = () => _onDataEnd();
+    _httpParser.dataEnd = (closed) => _onDataEnd(closed);
     _httpParser.error = (e) => _onError(e);
     // Tell the HTTP parser the method it is expecting a response to.
     _httpParser.responseToMethod = _method;
-
-    onDisconnect = _onDisconnected;
   }
 
-  void _propagateError(Exception e) {
-    if (_response != null && _response._streamErrorHandler != null) {
-      _response._streamErrorHandler(e);
+  void _responseDone() {
+    if (_closing) {
+      if (_socket != null) {
+        _socket.close();
+      }
+    } else {
+      _client._returnSocketConnection(_socketConn);
     }
+    _socket = null;
+    _socketConn = null;
   }
 
   HttpClientRequest open(String method, String uri) {
     _method = method;
     _request = new _HttpClientRequest(method, uri, this);
     _request.keepAlive = true;
     _response = new _HttpClientResponse(this);
     return _request;
   }
 
+  void _onConnectionClosed(Exception e) {
+    // Socket is closed either due to an error or due to normal socket close.
+    if (e != null) {
+      if (_onErrorCallback != null) {
+        _onErrorCallback(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();
+    }
+  }
+
   void _onRequestStart(String method, String uri) {
     // TODO(sgjesse): Error handling.
   }
 
   void _onResponseStart(int statusCode, String reasonPhrase) {
     _response._onResponseStart(statusCode, reasonPhrase);
   }
 
   void _onHeaderReceived(String name, String value) {
     _response._onHeaderReceived(name, value);
   }
 
   void _onHeadersComplete() {
     _response._onHeadersComplete();
   }
 
   void _onDataReceived(List<int> data) {
     _response._onDataReceived(data);
   }
 
-  void _onDataEnd() {
-    onDisconnect = null;
-    if (_response.headers["connection"] == "close") {
-      _socket.close();
-    } else {
-      _client._returnSocketConnection(_socketConn);
-    }
-    _socket = null;
-    _socketConn = null;
+  void _onDataEnd(bool close) {
+    if (close) _closing = true;
     _response._onDataEnd();
   }
 
   void set onRequest(void handler(HttpClientRequest request)) {
     _onRequest = handler;
   }
 
   void set onResponse(void handler(HttpClientResponse response)) {
     _onResponse = handler;
   }
 
-  void _onDisconnected() {
-    if (_onErrorCallback !== null) {
-      _onErrorCallback(new HttpException(
-          "Client disconnected before response was received."));
-    }
+  void set onError(void callback(Exception e)) {
+    _onErrorCallback = callback;
   }
 
   Function _onRequest;
   Function _onResponse;
+  Function _onErrorCallback;
 
   _HttpClient _client;
   _SocketConnection _socketConn;
   HttpClientRequest _request;
   HttpClientResponse _response;
   String _method;
 
   // Callbacks.
   var requestReceived;
 }
@@ skipping 190 matching lines @@
     sockets.addFirst(socketConn);
     socketConn._markReturned();
   }
 
   Function _onOpen;
   Map<String, Queue<_SocketConnection>> _openSockets;
   Set<_SocketConnection> _activeSockets;
   Timer _evictionTimer;
   bool _shutdown;  // Has this HTTP client been shutdown?
 }