Add support for Websocket protocols.

sdk/lib/io/websocket_impl.dart

 // Copyright (c) 2013, 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.
 
 part of dart.io;
 
 const String _webSocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
 
 class _WebSocketMessageType {
   static const int NONE = 0;
@@ skipping 375 matching lines @@
 
 class _WebSocketPong {
   final List<int> payload;
   _WebSocketPong([this.payload = null]);
 }
 
 
 class _WebSocketTransformerImpl implements WebSocketTransformer {
   final StreamController<WebSocket> _controller =
       new StreamController<WebSocket>(sync: true);
+  final Function _protocolSelector;
+
+  _WebSocketTransformerImpl(this._protocolSelector);
 
   Stream<WebSocket> bind(Stream<HttpRequest> stream) {
     stream.listen((request) {
-        _upgrade(request)
+        _upgrade(request, _protocolSelector)
             .then((WebSocket webSocket) => _controller.add(webSocket))
             .catchError(_controller.addError);
     });
 
     return _controller.stream;
   }
 
-  static Future<WebSocket> _upgrade(HttpRequest request) {
+  static Future<WebSocket> _upgrade(HttpRequest request, _protocolSelector) {
     var response = request.response;
     if (!_isUpgradeRequest(request)) {
-      // Send error response and drain the request.
-      request.listen((_) {}, onDone: () {
-        response.statusCode = HttpStatus.BAD_REQUEST;
-        response.contentLength = 0;
-        response.close();
-      });
+      // Send error response.
+      response.statusCode = HttpStatus.BAD_REQUEST;
+      response.close();
       return new Future.error(
           new WebSocketException("Invalid WebSocket upgrade request"));
     }

[Søren Gjesse, 2013/11/27 10:43:58]

Please add a newline before and after the local function.

[Anders Johnsen, 2013/11/27 11:43:30]

Done.

-
-    // Send the upgrade response.
-    response.statusCode = HttpStatus.SWITCHING_PROTOCOLS;
-    response.headers.add(HttpHeaders.CONNECTION, "Upgrade");
-    response.headers.add(HttpHeaders.UPGRADE, "websocket");
-    String key = request.headers.value("Sec-WebSocket-Key");
-    _SHA1 sha1 = new _SHA1();
-    sha1.add("$key$_webSocketGUID".codeUnits);
-    String accept = _CryptoUtils.bytesToBase64(sha1.close());
-    response.headers.add("Sec-WebSocket-Accept", accept);
-    response.headers.contentLength = 0;
-    return response.detachSocket()
-        .then((socket) => new _WebSocketImpl._fromSocket(socket, true));
+    Future upgrade(String protocol) {
+      // Send the upgrade response.
+      response.statusCode = HttpStatus.SWITCHING_PROTOCOLS;
+      response.headers.add(HttpHeaders.CONNECTION, "Upgrade");
+      response.headers.add(HttpHeaders.UPGRADE, "websocket");
+      String key = request.headers.value("Sec-WebSocket-Key");
+      _SHA1 sha1 = new _SHA1();
+      sha1.add("$key$_webSocketGUID".codeUnits);
+      String accept = _CryptoUtils.bytesToBase64(sha1.close());
+      response.headers.add("Sec-WebSocket-Accept", accept);
+      if (protocol != null && protocol.isNotEmpty) {
+        response.headers.add("Sec-WebSocket-Protocol", protocol);
+      }
+      response.headers.contentLength = 0;
+      return response.detachSocket()
+          .then((socket) => new _WebSocketImpl._fromSocket(
+                socket, protocol, true));
+    }
+    var protocols = request.headers['Sec-WebSocket-Protocol'];
+    if (protocols != null && _protocolSelector != null) {

[Søren Gjesse, 2013/11/27 10:43:58]

Please explain the handling of

Sec-WebSocket-Protocol: protocol1, protocol2

vs

Sec-WebSocket-Protocol: protocol1
Sec-WebSocket-Protocol: protocol2

here.

[Anders Johnsen, 2013/11/27 11:43:30]

Done.

+      protocols = _HttpParser._tokenizeFieldValue(protocols.join(', '));
+      return new Future(() => _protocolSelector(protocols))
+        .then(upgrade, onError: (error) {
+          response.statusCode = HttpStatus.INTERNAL_SERVER_ERROR;
+          response.close();
+          throw error;
+        });
+    } else {
+      return upgrade(null);
+    }
   }
 
   static bool _isUpgradeRequest(HttpRequest request) {
     if (request.method != "GET") {
       return false;
     }
     if (request.headers[HttpHeaders.CONNECTION] == null) {
       return false;
     }
     bool isUpgrade = false;
@@ skipping 275 matching lines @@
 
   void closeSocket() {
     _closed = true;
     _cancel();
     close();
   }
 }
 
 
 class _WebSocketImpl extends Stream implements WebSocket {
+  final String protocol;
+
   StreamController _controller;
   StreamSubscription _subscription;
   StreamSink _sink;
 
   final Socket _socket;
   final bool _serverSide;
   int _readyState = WebSocket.CONNECTING;
   bool _writeClosed = false;
   int _closeCode;
   String _closeReason;
   Duration _pingInterval;
   Timer _pingTimer;
   _WebSocketConsumer _consumer;
 
   int _outCloseCode;
   String _outCloseReason;
 
   static final HttpClient _httpClient = new HttpClient();
 
-  static Future<WebSocket> connect(String url, [protocols]) {
+  static Future<WebSocket> connect(String url, List<String> protocols) {
     Uri uri = Uri.parse(url);
     if (uri.scheme != "ws" && uri.scheme != "wss") {
       throw new WebSocketException("Unsupported URL scheme '${uri.scheme}'");
     }
     if (uri.userInfo != "") {
       throw new WebSocketException("Unsupported user info '${uri.userInfo}'");
     }
 
     Random random = new Random();
     // Generate 16 random bytes.
@@ skipping 10 matching lines @@
                   path: uri.path,
                   query: uri.query,
                   fragment: uri.fragment);
     return _httpClient.openUrl("GET", uri)
       .then((request) {
         // Setup the initial handshake.
         request.headers.add(HttpHeaders.CONNECTION, "upgrade");
         request.headers.set(HttpHeaders.UPGRADE, "websocket");
         request.headers.set("Sec-WebSocket-Key", nonce);
         request.headers.set("Sec-WebSocket-Version", "13");
+        if (protocols.isNotEmpty) {
+          request.headers.add("Sec-WebSocket-Protocol", protocols);
+        }
         return request.close();
       })
       .then((response) {
         void error(String message) {
           // Flush data.
           response.detachSocket().then((socket) {
             socket.destroy();
           });
           throw new WebSocketException(message);
         }
@@ skipping 14 matching lines @@
         List<int> expectedAccept = sha1.close();
         List<int> receivedAccept = _CryptoUtils.base64StringToBytes(accept);
         if (expectedAccept.length != receivedAccept.length) {
           error("Reasponse header 'Sec-WebSocket-Accept' is the wrong length");
         }
         for (int i = 0; i < expectedAccept.length; i++) {
           if (expectedAccept[i] != receivedAccept[i]) {
             error("Bad response 'Sec-WebSocket-Accept' header");
           }
         }
+        var protocol = response.headers.value('Sec-WebSocket-Protocol');
         return response.detachSocket()
-            .then((socket) => new _WebSocketImpl._fromSocket(socket));
+            .then((socket) => new _WebSocketImpl._fromSocket(socket, protocol));
       });
   }
 
   _WebSocketImpl._fromSocket(Socket this._socket,
+                             String this.protocol,
                              [bool this._serverSide = false]) {
     _consumer = new _WebSocketConsumer(this, _socket);
     _sink = new _StreamSinkImpl(_consumer);
     _readyState = WebSocket.OPEN;
 
     var transformer = new _WebSocketProtocolTransformer(_serverSide);
     _subscription = _socket.transform(transformer).listen(
         (data) {
           if (data is _WebSocketPing) {
             if (!_writeClosed) _consumer.add(new _WebSocketPong(data.payload));
@@ skipping 59 matching lines @@
       _pingTimer = new Timer(_pingInterval, () {
         // No pong received.
         _close(WebSocketStatus.GOING_AWAY);
       });
     });
   }
 
   int get readyState => _readyState;
 
   String get extensions => null;
-  String get protocol => null;
   int get closeCode => _closeCode;
   String get closeReason => _closeReason;
 
   void add(data) => _sink.add(data);
   void addError(error, [StackTrace stackTrace]) =>
       _sink.addError(error, stackTrace);
   Future addStream(Stream stream) => _sink.addStream(stream);
   Future get done => _sink.done;
 
   Future close([int code, String reason]) {
@@ skipping 22 matching lines @@
            (code < WebSocketStatus.NORMAL_CLOSURE ||
             code == WebSocketStatus.RESERVED_1004 ||
             code == WebSocketStatus.NO_STATUS_RECEIVED ||
             code == WebSocketStatus.ABNORMAL_CLOSURE ||
             (code > WebSocketStatus.INTERNAL_SERVER_ERROR &&
              code < WebSocketStatus.RESERVED_1015) ||
             (code >= WebSocketStatus.RESERVED_1015 &&
              code < 3000));
   }
 }