WebSocket Compression

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";
 
 // Matches _WebSocketOpcode.
 class _WebSocketMessageType {
@@ skipping 33 matching lines @@
  */
 // TODO(ajohnsen): make this transformer reusable?
 class _WebSocketProtocolTransformer implements StreamTransformer, EventSink {
   static const int START = 0;
   static const int LEN_FIRST = 1;
   static const int LEN_REST = 2;
   static const int MASK = 3;
   static const int PAYLOAD = 4;
   static const int CLOSED = 5;
   static const int FAILURE = 6;
+  static const int FIN = 0x80;
+  static const int RSV1 = 0x40;
+  static const int RSV2 = 0x20;
+  static const int RSV3 = 0x10;
+  static const int OPCODE = 0xF;
 
   int _state = START;
   bool _fin = false;
   bool _compressed = false;
   int _opcode = -1;
   int _len = -1;
   bool _masked = false;
   int _remainingLenBytes = -1;
   int _remainingMaskingKeyBytes = 4;
   int _remainingPayloadBytes = -1;
@@ skipping 38 matching lines @@
     if (_state == CLOSED) {
       throw new WebSocketException("Data on closed connection");
     }
     if (_state == FAILURE) {
       throw new WebSocketException("Data on failed connection");
     }
     while ((index < lastIndex) && _state != CLOSED && _state != FAILURE) {
       int byte = buffer[index];
       if (_state <= LEN_REST) {
         if (_state == START) {
-          _fin = (byte & 0x80) != 0;
+          _fin = (byte & FIN) != 0;
 
-          if ((byte & 0x40) != 0) {
+          if ((byte & RSV1) != 0) {
             _compressed = true;
           }
 
-          if ((byte & 0x20) != 0 || (byte & 0x10) != 0) {
+          if ((byte & RSV2) != 0 || (byte & RSV3) != 0) {
             // The RSV2 and RSV3 bits must be all zero.
             throw new WebSocketException("Protocol error");
           }
 
-          _opcode = (byte & 0xF);
+          _opcode = (byte & OPCODE);
 
           if (_opcode <= _WebSocketOpcode.BINARY) {
             if (_opcode == _WebSocketOpcode.CONTINUATION) {
               if (_currentMessageType == _WebSocketMessageType.NONE) {
                 throw new WebSocketException("Protocol error");
               }
             } else {
               assert(_opcode == _WebSocketOpcode.TEXT ||
                      _opcode == _WebSocketOpcode.BINARY);
               if (_currentMessageType != _WebSocketMessageType.NONE) {
@@ skipping 280 matching lines @@
           ..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) {
         response.headers.add("Sec-WebSocket-Protocol", protocol);
       }
 
-      var extensionHeader = request.headers.value("Sec-WebSocket-Extensions");
-
-      if (extensionHeader == null) {
-        extensionHeader = "";
-      }
-
-      Iterable<List<String>> extensions = extensionHeader.split(",").map((it) => it.split("; "));
-
-      if (compression.enabled && extensions.any((x) => x[0] == "permessage-deflate")) {
-        var opts = extensions.firstWhere((x) => x[0] == "permessage-deflate");
-        response.headers.add("Sec-WebSocket-Extensions", compression._createHeader(opts));
-      }
+      var deflate = _negotiateCompression(request, response, compression);
 
       response.headers.contentLength = 0;
       return response.detachSocket()
           .then((socket) => new _WebSocketImpl._fromSocket(
-                socket, protocol, compression, true));
+                socket, protocol, compression, true, deflate));
     }
 
     var protocols = request.headers['Sec-WebSocket-Protocol'];
     if (protocols != null && _protocolSelector != null) {
       // The suggested protocols can be spread over multiple lines, each
       // consisting of multiple protocols. To unify all of them, first join
       // the lists with ', ' and then tokenize.
       protocols = _HttpParser._tokenizeFieldValue(protocols.join(', '));
       return new Future(() => _protocolSelector(protocols))
         .then((protocol) {
           if (protocols.indexOf(protocol) < 0) {
             throw new WebSocketException(
                 "Selected protocol is not in the list of available protocols");
           }
           return protocol;
         })
         .catchError((error) {
           response
               ..statusCode = HttpStatus.INTERNAL_SERVER_ERROR
               ..close();
           throw error;
         })
         .then(upgrade);
     } else {
       return upgrade(null);
     }
   }
 
+  static _WebSocketPerMessageDeflate _negotiateCompression(HttpRequest request,
+                                    HttpResponse response,
+                                    CompressionOptions compression) {

[kustermann, 2015/07/20 09:34:12]

Consider using dart_style for automatic indentation.

+    var extensionHeader = request.headers.value("Sec-WebSocket-Extensions");
+
+    if (extensionHeader == null) {
+      extensionHeader = "";
+    }
+
+    Iterable<List<String>> extensions = extensionHeader
+        .split(",")
+        .map((it) => it.split("; "));
+
+    if (compression.enabled && extensions.any((x) => x[0] == _WebSocketImpl.PER_MESSAGE_DEFLATE)) {
+      var opts = extensions.firstWhere((x) => x[0] == _WebSocketImpl.PER_MESSAGE_DEFLATE);

[kustermann, 2015/07/20 09:34:12]

Seems redundant to do two traversals of [extensions]. Maybe just do one
'extensions.firstWhere(...)' and make `null` the default. 

[kustermann, 2015/07/20 09:34:12]

Normally we don't abbreviate, meaning opts -> options.

+      response.headers.add("Sec-WebSocket-Extensions", compression._createHeader(opts));
+      var noContextTakeover = opts.contains("server_no_context_takeover");
+      var deflate = new _WebSocketPerMessageDeflate(
+          noContextTakeover: noContextTakeover,
+          serverSide: true);
+
+      return deflate;
+    }
+
+    return null;
+  }
+
   static bool _isUpgradeRequest(HttpRequest request) {
     if (request.method != "GET") {
       return false;
     }
     if (request.headers[HttpHeaders.CONNECTION] == null) {
       return false;
     }
     bool isUpgrade = false;
     request.headers[HttpHeaders.CONNECTION].forEach((String value) {
       if (value.toLowerCase() == "upgrade") isUpgrade = true;
@@ skipping 18 matching lines @@
 class _WebSocketPerMessageDeflate {
   bool noContextTakeover;
   int clientMaxWindowBits;
   int serverMaxWindowBits;
   bool serverSide;
 
   ZLibDecoder decoder;
   ZLibEncoder encoder;
 
   _WebSocketPerMessageDeflate({this.clientMaxWindowBits,
-                              this.serverMaxWindowBits, this.noContextTakeover,
+                              this.serverMaxWindowBits,
+                              this.noContextTakeover,
                               this.serverSide: false}) {
     if (clientMaxWindowBits == null) {
-      clientMaxWindowBits = 15;
+      clientMaxWindowBits = _WebSocketImpl.DEFAULT_WINDOW_BITS;
     }
 
     if (serverMaxWindowBits == null) {
-      serverMaxWindowBits = 15;
+      serverMaxWindowBits = _WebSocketImpl.DEFAULT_WINDOW_BITS;
     }
   }
 
   void _ensureDecoder() {
     if (noContextTakeover || decoder == null) {
       decoder = new ZLibDecoder(windowBits: serverSide ?
         clientMaxWindowBits : serverMaxWindowBits);
     }
   }
 
   void _ensureEncoder() {
     if (noContextTakeover || encoder == null) {
       encoder = new ZLibEncoder(windowBits: serverSide ?
         serverMaxWindowBits : clientMaxWindowBits);
     }
   }
 
   List<int> processIncomingMessage(List<int> msg) {
     _ensureDecoder();
     var builder = new BytesBuilder();
     builder.add(msg);
     builder.add(const [0x00, 0x00, 0xff, 0xff]);
-    return decoder.convert(builder.takeBytes());
+    var result = decoder.convert(builder.takeBytes());
+    if (noContextTakeover) {
+      decoder = null;
+    }
+    return result;
   }
 
   List<int> processOutgoingMessage(List<int> msg) {
     _ensureEncoder();
     var c = encoder.convert(msg);
     c = c.sublist(0, c.length - 4);
+    if (noContextTakeover) {
+      encoder = null;
+    }
     return c;
   }
 }
 
 // TODO(ajohnsen): Make this transformer reusable.
 class _WebSocketOutgoingTransformer implements StreamTransformer, EventSink {
   final _WebSocketImpl webSocket;
   EventSink _eventSink;
 
   _WebSocketPerMessageDeflate _deflateHelper;
@@ skipping 20 matching lines @@
     if (message is _WebSocketPing) {
       addFrame(_WebSocketOpcode.PING, message.payload);
       return;
     }
     List<int> data;
     int opcode;
     if (message != null) {
       if (message is String) {
         opcode = _WebSocketOpcode.TEXT;
         data = UTF8.encode(message);
-
-        if (_deflateHelper != null) {
-          data = _deflateHelper.processOutgoingMessage(data);
-        }
       } else {
         if (message is !List<int>) {
           throw new ArgumentError(message);
         }
         opcode = _WebSocketOpcode.BINARY;
         data = message;
-
-        if (_deflateHelper != null) {
-          data = _deflateHelper.processOutgoingMessage(data);
-        }
       }
 
       if (_deflateHelper != null) {
         data = _deflateHelper.processOutgoingMessage(data);
       }
     } else {
       opcode = _WebSocketOpcode.TEXT;
     }
     addFrame(opcode, data);
   }
@@ skipping 11 matching lines @@
       data.add(code & 0xFF);
       if (reason != null) {
         data.addAll(UTF8.encode(reason));
       }
     }
     addFrame(_WebSocketOpcode.CLOSE, data);
     _eventSink.close();
   }
 
   void addFrame(int opcode, List<int> data) =>
-    createFrame(opcode, data, webSocket._serverSide, webSocket._deflate != null)
+    createFrame(opcode, data, webSocket._serverSide, _deflateHelper != null)
       .forEach((e) {
         _eventSink.add(e);
       });
 
   static Iterable createFrame(int opcode, List<int> data, bool serverSide, bool compressed) {
     bool mask = !serverSide;  // Masking not implemented for server.
     int dataLength = data == null ? 0 : data.length;
     // Determine the header size.
     int headerSize = (mask) ? 6 : 2;
     if (dataLength > 65535) {
@@ skipping 205 matching lines @@
     _closed = true;
     _cancel();
     close();
   }
 }
 
 
 class _WebSocketImpl extends Stream with _ServiceObject implements WebSocket {
   // Use default Map so we keep order.
   static Map<int, _WebSocketImpl> _webSockets = new Map<int, _WebSocketImpl>();
+  static const int DEFAULT_WINDOW_BITS = 15;
+  static const String PER_MESSAGE_DEFLATE = "permessage-deflate";
 
   final String protocol;
 
   StreamController _controller;
   StreamSubscription _subscription;
   StreamSink _sink;
 
   final _socket;
   final bool _serverSide;
   int _readyState = WebSocket.CONNECTING;
@@ skipping 45 matching lines @@
         }
         if (headers != null) {
           headers.forEach((field, value) => request.headers.add(field, value));
         }
         // Setup the initial handshake.
         request.headers
             ..set(HttpHeaders.CONNECTION, "Upgrade")
             ..set(HttpHeaders.UPGRADE, "websocket")
             ..set("Sec-WebSocket-Key", nonce)
             ..set("Cache-Control", "no-cache")
-            ..set("Sec-WebSocket-Version", "13")
-            ..set("Sec-WebSocket-Extensions", "permessage-deflate");
+            ..set("Sec-WebSocket-Version", "13");
         if (protocols != null) {
           request.headers.add("Sec-WebSocket-Protocol", protocols.toList());
         }
 
         request.headers.add("Sec-WebSocket-Extensions", compression._createHeader());
 
         return request.close();
       })
       .then((response) {
         void error(String message) {
@@ skipping 22 matching lines @@
         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');
 
-        String extensionHeader = response.headers.value('Sec-WebSocket-Extensions');
-
-        if (extensionHeader == null) {
-          extensionHeader = "";
-        }
-
-        Iterable<List<String>> extensions = extensionHeader
-          .split(", ")
-          .map((it) => it.split("; "));
-
-        _WebSocketPerMessageDeflate deflate;
-
-        if (compression.enabled && extensions.any((x) => x[0] == "permessage-deflate")) {
-          var opts = extensions.firstWhere((x) => x[0] == "permessage-deflate");
-          var noContextTakeover = opts.contains("client_no_context_takeover");
-
-          int getWindowBits(String type) {
-            var o = opts.firstWhere((x) =>
-              x.startsWith("${type}_max_window_bits="), orElse: () => null);
-
-            if (o == null) {
-              return 15;
-            }
-
-            try {
-              o = o.substring("client_max_window_bits=".length);
-              o = int.parse(o);
-            } catch (e) {
-              return 15;
-            }
-
-            return o;
-          }
-
-          deflate = new _WebSocketPerMessageDeflate(
-              clientMaxWindowBits: getWindowBits("client"),
-              serverMaxWindowBits: getWindowBits("server"),
-              noContextTakeover: noContextTakeover);
-        }
+        _WebSocketPerMessageDeflate deflate =
+          negotiateClientCompression(response, compression);
 
         return response.detachSocket()
           .then((socket) => new _WebSocketImpl._fromSocket(socket, protocol,
             compression, false, deflate));
       });
   }
 
+  static _WebSocketPerMessageDeflate negotiateClientCompression(
+        HttpClientResponse response,
+        CompressionOptions compression) {
+    String extensionHeader = response.headers.value('Sec-WebSocket-Extensions');
+
+    if (extensionHeader == null) {
+      extensionHeader = "";
+    }
+
+    Iterable<List<String>> extensions = extensionHeader
+      .split(", ")
+      .map((it) => it.split("; "));
+
+    if (compression.enabled && extensions.any((x) => x[0] == PER_MESSAGE_DEFLATE)) {
+      var opts = extensions.firstWhere((x) => x[0] == PER_MESSAGE_DEFLATE);
+      var noContextTakeover = opts.contains("client_no_context_takeover");
+
+      int getWindowBits(String type) {
+        var o = opts.firstWhere((x) =>
+          x.startsWith("${type}_max_window_bits="), orElse: () => null);
+
+        if (o == null) {
+          return DEFAULT_WINDOW_BITS;
+        }
+
+        try {
+          o = o.substring("${type}_max_window_bits=".length);
+          o = int.parse(o);
+        } catch (e) {
+          return DEFAULT_WINDOW_BITS;
+        }
+
+        return o;
+      }
+
+      return new _WebSocketPerMessageDeflate(
+          clientMaxWindowBits: getWindowBits("client"),
+          serverMaxWindowBits: getWindowBits("server"),
+          noContextTakeover: noContextTakeover);
+    }
+
+    return null;
+  }
+
   _WebSocketImpl._fromSocket(this._socket, this.protocol,
       CompressionOptions compression, [this._serverSide = false,
                              _WebSocketPerMessageDeflate deflate]) {
     _consumer = new _WebSocketConsumer(this, _socket);
     _sink = new _StreamSinkImpl(_consumer);
     _readyState = WebSocket.OPEN;
     _deflate = deflate;
 
     var transformer = new _WebSocketProtocolTransformer(_serverSide, _deflate);
     _subscription = _socket.transform(transformer).listen(
@@ skipping 162 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));
   }
 }