Initial web socket server implementation

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 _HttpHeaders implements HttpHeaders {
   _HttpHeaders() : _headers = new Map<String, List<String>>();
 
   List<String> operator[](String name) {
     name = name.toLowerCase();
     return _headers[name];
@@ skipping 343 matching lines @@
   _BufferList _buffer;
   Function _streamErrorHandler;
 }
 
 
 // HTTP response object for sending a HTTP response.
 class _HttpResponse extends _HttpRequestResponseBase implements HttpResponse {
   static final int START = 0;
   static final int HEADERS_SENT = 1;
   static final int DONE = 2;
+  static final int UPGRADED = 3;
 
   _HttpResponse(_HttpConnection httpConnection)
       : super(httpConnection),
         _statusCode = HttpStatus.OK,
         _state = START;
 
   void set contentLength(int contentLength) {
     if (_outputStream != null) throw new HttpException("Header already sent");
     _contentLength = contentLength;
   }
 
   int get statusCode() => _statusCode;
   void set statusCode(int statusCode) {
     if (_outputStream != null) throw new HttpException("Header already sent");
     _statusCode = statusCode;
   }
 
   String get reasonPhrase() => _findReasonPhrase(_statusCode);
   void set reasonPhrase(String reasonPhrase) {
     if (_outputStream != null) throw new HttpException("Header already sent");
     _reasonPhrase = reasonPhrase;
   }
 
   OutputStream get outputStream() {
-    if (_state == DONE) throw new HttpException("Response closed");
+    if (_state >= DONE) throw new HttpException("Response closed");
     if (_outputStream == null) {
       // Ensure that headers are written.
       if (_state == START) {
         _writeHeader();
       }
       _outputStream = new _HttpOutputStream(this);
     }
     return _outputStream;
   }
 
+  Socket detachSocket() {
+    if (_state >= DONE) throw new HttpException("Response closed");
+    // Ensure that headers are written.
+    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() {
     _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();
   }
 
@@ skipping 97 matching lines @@
     _writeCRLF();
 
     // Determine the value of the "Connection" header.
     if (_protocolVersion == "1.1" && !_persistentConnection) {
       _headers.set("Connection", "close");
     } else if (_protocolVersion == "1.0" && _persistentConnection) {
       _headers.set("Connection", "keep-alive");
     }
     // Determine the value of the "Transfer-Encoding" header based on
     // whether the content length is known.
-    if (_contentLength >= 0) {
+    if (_contentLength > 0) {
       _headers.set("Content-Length", _contentLength.toString());
     } else {
       _headers.set("Transfer-Encoding", "chunked");
     }
 
     // Write headers.
     bool allWritten = _writeHeaders();
     _state = HEADERS_SENT;
     return allWritten;
   }
@@ skipping 80 matching lines @@
   void set onError(void callback(Exception e)) {
     _requestOrResponse._streamSetErrorHandler(callback);
   }
 
   _HttpRequestResponseBase _requestOrResponse;
 }
 
 
 class _HttpConnectionBase implements Hashable {
   _HttpConnectionBase() : _sendBuffers = new Queue(),
-                          _httpParser = new _HttpParser();
+                          _httpParser = new _HttpParser() {
+    _hashCode = _nextHashCode;
+    _nextHashCode = (_nextHashCode + 1) & 0xFFFFFFF;
+  }
 
   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]) {
@@ skipping 16 matching lines @@
   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.
-        _close();
+      if (!_httpParser.upgrade) {
+        if (parsed != bytesRead) {
+          // TODO(sgjesse): Error handling.
+          _close();
+        }
       }
     }
   }
 
   void _onClosed() {
     _closing = true;
     _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();
     }
     _onConnectionClosed(e);
   }
 
+  Socket _detachSocket() {
+    _socket.onData = null;
+    // TODO(sgjesse): Handle getting the write handler when using output stream.
+    //_socket.onWrite = null;
+    _socket.onClosed = null;
+    _socket.onError = null;
+    Socket socket = _socket;
+    _socket = null;
+    if (onDetach) onDetach();
+    return socket;
+  }
+
   abstract void _onConnectionClosed(Exception e);
   abstract void _responseDone();
 
   void set _onNoPendingWrites(void callback()) {
     if (!_error) {
       _socket.outputStream.onNoPendingWrites = callback;
     }
   }
 
-  int hashCode() => _socket.hashCode();
+  int hashCode() => _hashCode;
 
   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 onDetach;
+
+  // Hash code for HTTP connection. Currently this is just a counter.
+  int _hashCode;
+  static int _nextHashCode = 0;
 }
 
 
 // HTTP server connection over a socket.
 class _HttpConnection extends _HttpConnectionBase {
   _HttpConnection(HttpServer this._server) {
     // Register HTTP parser callbacks.
     _httpParser.requestStart =
       (method, uri, version) => _onRequestStart(method, uri, version);
     _httpParser.responseStart =
@@ skipping 117 matching lines @@
     _closeServer = true;
   }
 
   void listenOn(ServerSocket serverSocket) {
     void onConnection(Socket socket) {
       // Accept the client connection.
       _HttpConnection connection = new _HttpConnection(this);
       connection._connectionEstablished(socket);
       connection.onRequestReceived = _handleRequest;
       connection.onClosed = () => _connections.remove(connection);
+      connection.onDetach = () => _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 139 matching lines @@
     data = _method.toString().charCodes();
     _httpConnection._write(data);
     _writeSP();
     data = _uri.toString().charCodes();
     _httpConnection._write(data);
     _writeSP();
     _httpConnection._write(_Const.HTTP11);
     _writeCRLF();
 
     // Determine the value of the "Transfer-Encoding" header based on
-    // whether the content length is known.
-    if (_contentLength >= 0) {
+    // whether the content length is known. If there is no content
+    // neither "Content-Length" nor "Transfer-Encoding" is set
+    if (_contentLength > 0) {
       _headers.set("Content-Length", _contentLength.toString());
-    } else {
+    } else if (_contentLength < 0) {
       _headers.set("Transfer-Encoding", "chunked");
     }
 
     // Write headers.
     _writeHeaders();
     _state = HEADERS_SENT;
   }
 
   String _method;
   String _uri;
@@ skipping 390 matching lines @@
     _activeSockets.remove(socketConn);
     sockets.addFirst(socketConn);
   }
 
   Function _onOpen;
   Map<String, Queue<_SocketConnection>> _openSockets;
   Set<_SocketConnection> _activeSockets;
   Timer _evictionTimer;
   bool _shutdown;  // Has this HTTP client been shutdown?
 }