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 20 matching lines @@
   static const int RESERVED_D = 13;
   static const int RESERVED_E = 14;
   static const int RESERVED_F = 15;
 }
 
 /**
  * The web socket protocol transformer handles the protocol byte stream
  * which is supplied through the [:handleData:]. As the protocol is processed,
  * it'll output frame data as either a List<int> or String.
  *
- * Important infomation about usage: Be sure you use cancelOnError, so the
- * socket will be closed when the processer encounter an error. Not using it
+ * Important information about usage: Be sure you use cancelOnError, so the
+ * socket will be closed when the processor encounter an error. Not using it
  * will lead to undefined behaviour.
  */
 // 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;
 
   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;
   int _unmaskingIndex = 0;
   int _currentMessageType = _WebSocketMessageType.NONE;
   int closeCode = WebSocketStatus.NO_STATUS_RECEIVED;
   String closeReason = "";
 
   EventSink _eventSink;
 
   final bool _serverSide;
   final List _maskingBytes = new List(4);
   final BytesBuilder _payload = new BytesBuilder(copy: false);
 
-  _WebSocketProtocolTransformer([this._serverSide = false]);
+  _WebSocketPerMessageDeflate _deflate;
+  _WebSocketProtocolTransformer([this._serverSide = false, this._deflate]);
 
   Stream bind(Stream stream) {
     return new Stream.eventTransformed(
         stream,
         (EventSink eventSink) {
           if (_eventSink != null) {
             throw new StateError("WebSocket transformer already used.");
           }
           _eventSink = eventSink;
           return this;
@@ skipping 15 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;

[Søren Gjesse, 2015/06/25 15:41:27]

Please add constants for FIN, RSV1, RSV2 and RSV3 bit-patterns.

-          if ((byte & 0x70) != 0) {
-            // The RSV1, RSV2 bits RSV3 must be all zero.
+
+          if ((byte & 0x40) != 0) {
+            _compressed = true;
+          }
+
+          if ((byte & 0x20) != 0 || (byte & 0x10) != 0) {
+            // The RSV2 and RSV3 bits must be all zero.
             throw new WebSocketException("Protocol error");
           }
+
           _opcode = (byte & 0xF);

[Søren Gjesse, 2015/06/25 15:41:27]

Please add a constant for the opcode mask.

+
           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) {
                 throw new WebSocketException("Protocol error");
@@ skipping 151 matching lines @@
       } else {
         _messageFrameEnd();
       }
     } else {
       _state = PAYLOAD;
     }
   }
 
   void _messageFrameEnd() {
     if (_fin) {
+      var bytes = _payload.takeBytes();
+      if (_deflate != null && _compressed) {
+        bytes = _deflate.processIncomingMessage(bytes);
+      }
+
       switch (_currentMessageType) {
         case _WebSocketMessageType.TEXT:
-          _eventSink.add(UTF8.decode(_payload.takeBytes()));
+          _eventSink.add(UTF8.decode(bytes));
           break;
         case _WebSocketMessageType.BINARY:
-          _eventSink.add(_payload.takeBytes());
+          _eventSink.add(bytes);
           break;
       }
       _currentMessageType = _WebSocketMessageType.NONE;
     }
     _prepareForNextFrame();
   }
 
   void _controlFrameEnd() {
     switch (_opcode) {
       case _WebSocketOpcode.CLOSE:
@@ skipping 54 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;
+  final CompressionOptions _compression;
 
-  _WebSocketTransformerImpl(this._protocolSelector);
+  _WebSocketTransformerImpl(this._protocolSelector, this._compression);
 
   Stream<WebSocket> bind(Stream<HttpRequest> stream) {
     stream.listen((request) {
-        _upgrade(request, _protocolSelector)
+        _upgrade(request, _protocolSelector, _compression)
             .then((WebSocket webSocket) => _controller.add(webSocket))
             .catchError(_controller.addError);
     }, onDone: () {
       _controller.close();
     });
 
     return _controller.stream;
   }
 
-  static Future<WebSocket> _upgrade(HttpRequest request, _protocolSelector) {
+  static Future<WebSocket> _upgrade(HttpRequest request, _protocolSelector,
+                                    CompressionOptions compression) {
     var response = request.response;
     if (!_isUpgradeRequest(request)) {
       // Send error response.
       response
           ..statusCode = HttpStatus.BAD_REQUEST
           ..close();
       return new Future.error(
           new WebSocketException("Invalid WebSocket upgrade request"));
     }
 
     Future upgrade(String protocol) {
       // Send the upgrade response.
       response
           ..statusCode = HttpStatus.SWITCHING_PROTOCOLS
           ..headers.add(HttpHeaders.CONNECTION, "Upgrade")
           ..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);
       }
+

[Søren Gjesse, 2015/06/25 15:41:27]

Extract server side negotiation into a separate method.

+      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));
+      }
+
       response.headers.contentLength = 0;
       return response.detachSocket()
           .then((socket) => new _WebSocketImpl._fromSocket(

[Søren Gjesse, 2015/06/25 15:41:27]

Compression not supported for server (no additional optional argument passed)?

-                socket, protocol, true));
+                socket, protocol, compression, true));
     }
 
     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) {
@@ skipping 36 matching lines @@
       return false;
     }
     String key = request.headers.value("Sec-WebSocket-Key");
     if (key == null) {
       return false;
     }
     return true;
   }
 }
 
+class _WebSocketPerMessageDeflate {
+  bool noContextTakeover;
+  int clientMaxWindowBits;
+  int serverMaxWindowBits;
+  bool serverSide;
+
+  ZLibDecoder decoder;
+  ZLibEncoder encoder;
+
+  _WebSocketPerMessageDeflate({this.clientMaxWindowBits,

[Søren Gjesse, 2015/06/25 15:41:27]

No need to use optional named argument for internal private classes.

+                              this.serverMaxWindowBits, this.noContextTakeover,
+                              this.serverSide: false}) {
+    if (clientMaxWindowBits == null) {

[Søren Gjesse, 2015/06/25 15:41:27]

Make the caller always pass a non null value - it must know from the negoation.

+      clientMaxWindowBits = 15;

[Søren Gjesse, 2015/06/25 15:41:27]

Create a constant for this (maybe in CompressionOptions).

+    }
+
+    if (serverMaxWindowBits == null) {
+      serverMaxWindowBits = 15;
+    }
+  }
+
+  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]);

[Søren Gjesse, 2015/06/25 15:41:27]

For the noContextTakeover case set decoder to null, or figure out if there is
some way to reset it to not consume space.

+    return decoder.convert(builder.takeBytes());
+  }
+
+  List<int> processOutgoingMessage(List<int> msg) {
+    _ensureEncoder();
+    var c = encoder.convert(msg);
+    c = c.sublist(0, c.length - 4);

[Søren Gjesse, 2015/06/25 15:41:27]

Ditto.

+    return c;
+  }
+}
 
 // TODO(ajohnsen): Make this transformer reusable.
 class _WebSocketOutgoingTransformer implements StreamTransformer, EventSink {
   final _WebSocketImpl webSocket;
   EventSink _eventSink;
 
+  _WebSocketPerMessageDeflate _deflateHelper;
+
   _WebSocketOutgoingTransformer(this.webSocket);
 
   Stream bind(Stream stream) {
     return new Stream.eventTransformed(
         stream,
         (EventSink eventSink) {
           if (_eventSink != null) {
             throw new StateError("WebSocket transformer already used");
           }
           _eventSink = eventSink;
           return this;
         });
   }
 
   void add(message) {
     if (message is _WebSocketPong) {
       addFrame(_WebSocketOpcode.PONG, message.payload);
       return;
     }
     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);
   }
 
   void addError(Object error, [StackTrace stackTrace]) =>
       _eventSink.addError(error, stackTrace);
 
   void close() {
     int code = webSocket._outCloseCode;
     String reason = webSocket._outCloseReason;
     List<int> data;
     if (code != null) {
       data = new List<int>();
       data.add((code >> 8) & 0xFF);
       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).forEach(_eventSink.add);
+    createFrame(opcode, data, webSocket._serverSide, webSocket._deflate != null)
+      .forEach((e) {
+        _eventSink.add(e);
+      });
 
-  static Iterable createFrame(int opcode, List<int> data, bool serverSide) {
+  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) {
       headerSize += 8;
     } else if (dataLength > 125) {
       headerSize += 2;
     }
     Uint8List header = new Uint8List(headerSize);
     int index = 0;
     // Set FIN and opcode.
-    header[index++] = 0x80 | opcode;
+    var hoc = 0;
+
+    hoc |= 0x80;
+
+    if (compressed) {
+      hoc |= 0x40;
+    }
+
+    hoc |= opcode & 0xF;
+
+    header[index++] = hoc;
     // Determine size and position of length field.
     int lengthBytes = 1;
-    int firstLengthByte = 1;
     if (dataLength > 65535) {
       header[index++] = 127;
       lengthBytes = 8;
     } else if (dataLength > 125) {
       header[index++] = 126;
       lengthBytes = 2;
     }
     // Write the length in network byte order into the header.
     for (int i = 0; i < lengthBytes; i++) {
       header[index++] = dataLength >> (((lengthBytes - 1) - i) * 8) & 0xFF;
@@ skipping 195 matching lines @@
   bool _writeClosed = false;
   int _closeCode;
   String _closeReason;
   Duration _pingInterval;
   Timer _pingTimer;
   _WebSocketConsumer _consumer;
 
   int _outCloseCode;
   String _outCloseReason;
   Timer _closeTimer;
+  _WebSocketPerMessageDeflate _deflate;
 
   static final HttpClient _httpClient = new HttpClient();
 
   static Future<WebSocket> connect(
-      String url, Iterable<String> protocols, Map<String, dynamic> headers) {
+      String url, Iterable<String> protocols, Map<String, dynamic> headers,
+      {CompressionOptions compression: CompressionOptions.DEFAULT}) {
     Uri uri = Uri.parse(url);
     if (uri.scheme != "ws" && uri.scheme != "wss") {
       throw new WebSocketException("Unsupported URL scheme '${uri.scheme}'");
     }
 
     Random random = new Random();
     // Generate 16 random bytes.
     Uint8List nonceData = new Uint8List(16);
     for (int i = 0; i < 16; i++) {
       nonceData[i] = random.nextInt(256);
@@ skipping 18 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-Version", "13")
+            ..set("Sec-WebSocket-Extensions", "permessage-deflate");

[Søren Gjesse, 2015/06/25 15:41:27]

Isn't this already handled by adding Sec-WebSocket-Extensions below?

         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) {
           // Flush data.
           response.detachSocket().then((socket) {
             socket.destroy();
           });
           throw new WebSocketException(message);
         }
@@ skipping 15 matching lines @@
         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');
+

[Søren Gjesse, 2015/06/25 15:41:27]

Please extract this into a separate method called something like
negotiateClientCompression.

+        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);

[Søren Gjesse, 2015/06/25 15:41:27]

client -> ${type} here as well?

Use

var parameter = "${type}_max_window_bits="

to only have this once.

+              o = int.parse(o);
+            } catch (e) {
+              return 15;
+            }
+
+            return o;
+          }
+
+          deflate = new _WebSocketPerMessageDeflate(
+              clientMaxWindowBits: getWindowBits("client"),
+              serverMaxWindowBits: getWindowBits("server"),
+              noContextTakeover: noContextTakeover);
+        }
+
         return response.detachSocket()
-            .then((socket) => new _WebSocketImpl._fromSocket(socket, protocol));
+          .then((socket) => new _WebSocketImpl._fromSocket(socket, protocol,
+            compression, false, deflate));
       });
   }
 
   _WebSocketImpl._fromSocket(this._socket, this.protocol,
-                             [this._serverSide = false]) {
+      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);
+    var transformer = new _WebSocketProtocolTransformer(_serverSide, _deflate);
     _subscription = _socket.transform(transformer).listen(
         (data) {
           if (data is _WebSocketPing) {
             if (!_writeClosed) _consumer.add(new _WebSocketPong(data.payload));
           } else if (data is _WebSocketPong) {
             // Simply set pingInterval, as it'll cancel any timers.
             pingInterval = _pingInterval;
           } else {
             _controller.add(data);
           }
@@ skipping 153 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));
   }
 }