Handle closing of HTTP client and server

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.
 
 // Global constants.
 class _Const {
   // Bytes for "HTTP/1.0".
   static final HTTP10 = const [72, 84, 84, 80, 47, 49, 46, 48];
   // Bytes for "HTTP/1.1".
   static final HTTP11 = const [72, 84, 84, 80, 47, 49, 46, 49];
@@ skipping 670 matching lines @@
     _keepAlive = keepAlive;
   }
 
   int get statusCode() => _statusCode;
   void set statusCode(int statusCode) {
     if (_outputStream != null) return new HttpException("Header already sent");
     _statusCode = statusCode;
   }
 
   String get reasonPhrase() => _findReasonPhrase(_statusCode);
-  void set reasonPhrase(String reasonPhrase) => _reasonPhrase = reasonPhrase;
+  void set reasonPhrase(String reasonPhrase) {
+    if (_outputStream != null) return new HttpException("Header already sent");
+    _reasonPhrase = reasonPhrase;
+  }
 
   // Set a header on the response. NOTE: If the same header is set
   // more than once only the last one will be part of the response.
   void setHeader(String name, String value) {
     if (_outputStream != null) return new HttpException("Header already sent");
     _setHeader(name, value);
   }
 
   OutputStream get outputStream() {
     if (_state == DONE) throw new HttpException("Response closed");
@@ skipping 16 matching lines @@
   // 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.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();
     }
@@ skipping 172 matching lines @@
   }
 
   void set onError(void callback()) {
     _requestOrResponse._streamSetErrorHandler(callback);
   }
 
   _HttpRequestResponseBase _requestOrResponse;
 }
 
 
-class _HttpConnectionBase {
-  _HttpConnectionBase() : _sendBuffers = new Queue(),
+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(),
                           _httpParser = new HttpParser();
 
   void _connectionEstablished(Socket socket) {
     _socket = socket;
     // Register handler for socket events.
     _socket.onData = _onData;
     _socket.onClosed = _onClosed;
     _socket.onError = _onError;
   }
 
@@ skipping 12 matching lines @@
     if (bytesRead > 0) {
       int parsed = _httpParser.writeList(buffer, 0, bytesRead);
       if (parsed != bytesRead) {
         // TODO(sgjesse): Error handling.
         _socket.close();
       }
     }
   }
 
   void _onClosed() {
-    // Client closed socket for writing. Socket should still be open
-    // for writing the response.
-    _closing = true;
+    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();
   }
 
   void _onError() {
     // If an error occours, treat the socket as closed.
     _onClosed();
     if (_onErrorCallback != null) {
       _onErrorCallback("Connection closed while sending data to client.");
     }
   }
 
   void set onDisconnect(void callback()) {
     _onDisconnectCallback = callback;
   }
 
   void set onError(void callback(String errorMessage)) {
     _onErrorCallback = callback;
   }
 
+  int hashCode() => _socket.hashCode();
+
+  int _phase;
   Socket _socket;
   bool _closing = false;  // Is the socket closed by the client?
   HttpParser _httpParser;
 
   Queue _sendBuffers;
 
   Function _onDisconnectCallback;
   Function _onErrorCallback;
 }
 
 
 // HTTP server connection over a socket.
 class _HttpConnection extends _HttpConnectionBase {
   _HttpConnection() {
     // 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();
   }
 
   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);
     }
   }
 
   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;
+    }
     _request._onDataEnd();
   }
 
   HttpRequest _request;
   HttpResponse _response;
 
   // Callbacks.
   var requestReceived;
 }
 
 
 // HTTP server waiting for socket connections. The connections are
 // managed by the server and as requests are received the request.
 class _HttpServer implements HttpServer {
   void listen(String host, int port, [int backlog = 5]) {
 
     void onConnection(Socket socket) {
       // Accept the client connection.
       _HttpConnection connection = new _HttpConnection();
       connection._connectionEstablished(socket);
       connection.requestReceived = _onRequest;
       _connections.add(connection);
       void onDisconnect() {
-        for (int i = 0; i < _connections.length; i++) {
-          if (_connections[i] == connection) {
-            _connections.removeRange(i, 1);
-            break;
-          }
-        }
+        _connections.remove(connection);
       }
       connection.onDisconnect = onDisconnect;
       void onError(String errorMessage) {
         if (_onError != null) _onError(errorMessage);
       }
       connection.onError = onError;
     }
 
-    // TODO(ajohnsen): Use Set once Socket is Hashable.
-    _connections = new List<_HttpConnection>();
+    _connections = new Set<_HttpConnection>();
     _server = new ServerSocket(host, port, backlog);
     _server.onConnection = onConnection;
   }
 
   void close() => _server.close();
   int get port() => _server.port;
 
   void set onError(void handler(String errorMessage)) {
     _onError = handler;
   }
 
   void set onRequest(void handler(HttpRequest, HttpResponse)) {
     _onRequest = handler;
   }
 
   ServerSocket _server;  // The server listen socket.
-  List<_HttpConnection> _connections;  // List of currently connected clients.
+  Set<_HttpConnection> _connections;  // Set of currently connected clients.
   Function _onRequest;
   Function _onError;
 }
 
 
 class _HttpClientRequest
     extends _HttpRequestResponseBase implements HttpClientRequest {
   static final int START = 0;
   static final int HEADERS_SENT = 1;
   static final int DONE = 2;
@@ skipping 268 matching lines @@
 
   void _markReturned() {
     _socket.onData = null;
     _socket.onClosed = null;
     _socket.onError = null;
     _returnTime = new Date.now();
   }
 
   Duration _idleTime(Date now) => now.difference(_returnTime);
 
+  int hashCode() => _socket.hashCode();
+
   String _host;
   int _port;
   Socket _socket;
   Date _returnTime;
 }
 
 
 class _HttpClient implements HttpClient {
   static final int DEFAULT_EVICTION_TIMEOUT = 60000;
 
-  _HttpClient() : _openSockets = new Map(), _shutdown = false;
+  _HttpClient() : _openSockets = new Map(),
+                  _activeSockets = new Set(),
+                  _shutdown = false;
 
   HttpClientConnection open(
       String method, String host, int port, String path) {
     if (_shutdown) throw new HttpException("HttpClient shutdown");
     return _prepareHttpClientConnection(host, port, method, path);
   }
 
   HttpClientConnection get(String host, int port, String path) {
     return open("GET", host, port, path);
   }
 
   HttpClientConnection post(String host, int port, String path) {
     return open("POST", host, port, path);
   }
 
   void shutdown() {
-     _openSockets.forEach(
-         void _(String key, Queue<_SocketConnection> connections) {
-           while (!connections.isEmpty()) {
-             var socketConn = connections.removeFirst();
-             socketConn._socket.close();
-           }
-         });
+     _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) {
        _evictionTimer.cancel();
      }
      _shutdown = true;
   }
 
   String _connectionKey(String host, int port) {
     return "$host:$port";
   }
 
@@ skipping 15 matching lines @@
 
     // If there are active connections for this key get the first one
     // otherwise create a new one.
     Queue socketConnections = _openSockets[_connectionKey(host, port)];
     if (socketConnections == null || socketConnections.isEmpty()) {
       Socket socket = new Socket(host, port);
       socket.onConnect = () {
         socket.onError = null;
         _SocketConnection socketConn =
             new _SocketConnection(host, port, socket);
+        _activeSockets.add(socketConn);
         _connectionOpened(socketConn, connection);
       };
       socket.onError = () {
         if (_onError !== null) {
           _onError(HttpStatus.NETWORK_CONNECT_TIMEOUT_ERROR);
         }
       };
     } else {
       _SocketConnection socketConn = socketConnections.removeFirst();
+      _activeSockets.add(socketConn);
       new Timer((ignored) => _connectionOpened(socketConn, connection), 0);
 
       // Get rid of eviction timer if there are no more active connections.
       if (socketConnections.isEmpty()) {
         _evictionTimer.cancel();
         _evictionTimer = null;
       }
     }
 
     return connection;
@@ skipping 31 matching lines @@
                 } else {
                   break;
                 }
               }
             });
       }
       _evictionTimer = new Timer.repeating(_handleEviction, 10000);
     }
 
     // Return connection.
+    _activeSockets.remove(socketConn);
     sockets.addFirst(socketConn);
     socketConn._markReturned();
   }
 
   void set onError(void callback(int status)) {
     _onError = callback;
   }
 
   Function _onOpen;
   Function _onError;
   Map<String, Queue<_SocketConnection>> _openSockets;
+  Set<_SocketConnection> _activeSockets;
   Timer _evictionTimer;
   bool _shutdown;  // Has this HTTP client been shutdown?
 }
 
 
 class HttpUtil {
   static String decodeUrlEncodedString(String urlEncoded) {
     void invalidEscape() {
       // TODO(sgjesse): Handle the error.
     }
@@ skipping 43 matching lines @@
       } else {
         value = queryString.substring(currentPosition, position);
         currentPosition = position + 1;
       }
       result[HttpUtil.decodeUrlEncodedString(name)] =
         HttpUtil.decodeUrlEncodedString(value);
     }
     return result;
   }
 }