Refactor the close and error handling of HTTP connections

runtime/bin/http_impl.dart

Index: runtime/bin/http_impl.dart
diff --git a/runtime/bin/http_impl.dart b/runtime/bin/http_impl.dart
index 921d7cb61077dba34b67f07fef4d7f47af702a56..d210ed92935238f12fa897fada90a555ae36a5b1 100644
--- a/runtime/bin/http_impl.dart
+++ b/runtime/bin/http_impl.dart
@@ -260,6 +260,16 @@ class _HttpResponse extends _HttpRequestResponseBase implements HttpResponse {
     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);
@@ -270,17 +280,7 @@ class _HttpResponse extends _HttpRequestResponseBase implements HttpResponse {
   }
 
   void _streamClose() {
-    _httpConnection._phase = _HttpConnectionBase.PHASE_IDLE;
-    _state = DONE;
-    // Stop tracking no pending write events.
-    _httpConnection.outputStream.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.outputStream.close();
-    }
+    _responseEnd();
   }
 
   void _streamSetNoPendingWriteHandler(callback()) {
@@ -470,12 +470,7 @@ class _HttpOutputStream extends _BaseOutputStream implements OutputStream {
 
 
 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) {
@@ -508,49 +503,29 @@ class _HttpConnectionBase implements Hashable {
   }
 
   void _onClosed() {
-    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);
-  }
-
-  abstract void _propagateError(Exception e);
-
-  void set onDisconnect(void callback()) {
-    _onDisconnectCallback = callback;
+    _onConnectionClosed(e);
   }
 
-  void set onError(void callback(Exception e)) {
-    _onErrorCallback = callback;
-  }
+  abstract void _onConnectionClosed(Exception e);
+  abstract void _responseDone();
 
   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;
 }
 
 
@@ -567,13 +542,42 @@ class _HttpConnection extends _HttpConnectionBase {
         (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;
+    }
+  }
+
   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);
@@ -590,8 +594,8 @@ class _HttpConnection extends _HttpConnectionBase {
   void _onHeadersComplete() {
     _request._onHeadersComplete();
     _response.keepAlive = _httpParser.keepAlive;
-    if (requestReceived != null) {
-      requestReceived(_request, _response);
+    if (onRequestReceived != null) {
+      onRequestReceived(_request, _response);
     }
   }
 
@@ -599,22 +603,20 @@ class _HttpConnection extends _HttpConnectionBase {
     _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);
-    }
-    if (_response != null && _response._streamErrorHandler != null) {
-      _response._streamErrorHandler(e);
+  void _responseDone() {
+    // If the connection is closing then close the output stream to
+    // fully close the socket.
+    if (_closing) {
+      outputStream.close();
     }
+    _response = null;
   }
 
   HttpServer _server;
@@ -622,7 +624,9 @@ class _HttpConnection extends _HttpConnectionBase {
   HttpResponse _response;
 
   // Callbacks.
-  var requestReceived;
+  Function onRequestReceived;
+  Function onClosed;
+  Function onError;
 }
 
 
@@ -640,10 +644,12 @@ class _HttpServer implements HttpServer {
     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) {
         if (_onError != null) _onError(e);
+        _connections.remove(connection);

[Anders Johnsen, 2012/04/03 05:36:37]

So, here the order matters. In the future when we ask for current connections,
do we want include this connection if called from onError. The problem is that
it's not valid at this point.

[Søren Gjesse, 2012/04/10 13:10:08]

Moved the remove up before the _onError call. In general when asking for the
current connections you can still get connections which are not valid.

       };
       connection._connectionEstablished(socket);
       _connections.add(connection);
@@ -773,11 +779,6 @@ class _HttpClientRequest
     _httpConnection.outputStream.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.outputStream.close();
-    }
   }
 
   void _streamSetNoPendingWriteHandler(callback()) {
@@ -887,6 +888,7 @@ class _HttpClientResponse
 
   void _onDataEnd() {
     if (_inputStream != null) _inputStream._closeReceived();
+    _connection._responseDone();
   }
 
   // Delegate functions for the HttpInputStream implementation.
@@ -935,18 +937,22 @@ class _HttpClientConnection
         (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) {
@@ -957,6 +963,28 @@ class _HttpClientConnection
     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.
   }
@@ -977,15 +1005,8 @@ class _HttpClientConnection
     _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();
   }
 
@@ -997,15 +1018,13 @@ class _HttpClientConnection
     _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;