Initial Implementation of 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;
@@ skipping 11 matching lines @@
   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);
 
+  _WebSocketPerMessageDeflateHelper _deflateHelper;
+
   _WebSocketProtocolTransformer([this._serverSide = false]);
 
   Stream bind(Stream stream) {
     return new Stream.eventTransformed(
         stream,
         (EventSink eventSink) {
           if (_eventSink != null) {
             throw new StateError("WebSocket transformer already used.");
           }
           _eventSink = eventSink;
@@ skipping 192 matching lines @@
         _prepareForNextFrame();
       } else {
         _messageFrameEnd();
       }
     } else {
       _state = PAYLOAD;
     }
   }
 
   void _messageFrameEnd() {
     if (_fin) {

[Søren Gjesse, 2015/05/26 09:11:33]

Move getting bytes and optionally decompression up here.

You also need to check the RSV1 bit to check whether to decompress.

       switch (_currentMessageType) {
         case _WebSocketMessageType.TEXT:
-          _eventSink.add(UTF8.decode(_payload.takeBytes()));
+          var bytes = _payload.takeBytes();
+          if (_deflateHelper != null) {
+            bytes = _deflateHelper.processIncomingMessage(bytes);
+          }
+          _eventSink.add(UTF8.decode(bytes));
           break;
         case _WebSocketMessageType.BINARY:
-          _eventSink.add(_payload.takeBytes());
+          var bytes = _payload.takeBytes();
+          if (_deflateHelper != null) {
+            bytes = _deflateHelper.processIncomingMessage(bytes);
+          }
+          _eventSink.add(bytes);
           break;
       }
       _currentMessageType = _WebSocketMessageType.NONE;
     }
     _prepareForNextFrame();
   }
 
   void _controlFrameEnd() {
     switch (_opcode) {
       case _WebSocketOpcode.CLOSE:
@@ skipping 94 matching lines @@
           ..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);
       }
+
+      var extensionHeader = request.headers.value("Sec-WebSocket-Extensions");
+
+      if (extensionHeader == null) {
+        extensionHeader = "";
+      }
+
+      Iterable<List<String>> extensions = extensionHeader.split(",").map((it) => it.split("; "));

[Søren Gjesse, 2015/05/26 09:11:33]

I think it will be helpful to create a class to represent the
Sec-WebSocket-Extensions header, e.g. like the ContentType for the HTTP
Content-Type
(https://api.dartlang.org/apidocs/channels/stable/dartdoc-viewer/dart:io.Conte...).

[Søren Gjesse, 2015/05/26 09:11:33]

Long line - more below.

+
+      _WebSocketPerMessageDeflateHelper deflateHelper;
+
+      if (extensions.any((x) => x[0] == "permessage-deflate")) {
+        response.headers.add("Sec-WebSocket-Extensions", "permessage-deflate");
+        deflateHelper = new _WebSocketPerMessageDeflateHelper();
+      }
+
       response.headers.contentLength = 0;
       return response.detachSocket()
           .then((socket) => new _WebSocketImpl._fromSocket(
-                socket, protocol, true));
+                socket, protocol, true).._deflateHelper = deflateHelper);

[Søren Gjesse, 2015/05/26 09:11:33]

Please pass the arguments used for handling compression to the constructor, and
let the constructor create the deflater.

     }
 
     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 _WebSocketPerMessageDeflateHelper {

[Søren Gjesse, 2015/05/26 09:11:33]

I would just call this _WebSocketPerMessageDeflater.

+  int windowBits;
+
+  ZLibDecoder decoder;
+  ZLibEncoder encoder;
+
+  _WebSocketPerMessageDeflateHelper({this.windowBits}) {
+    decoder = windowBits != null ? new ZLibDecoder(windowBits: windowBits) : new ZLibDecoder();
+    encoder = windowBits != null ? new ZLibEncoder(windowBits: windowBits) : new ZLibEncoder();
+  }
+
+  List<int> processIncomingMessage(List<int> msg) {
+    var builder = new BytesBuilder();
+    builder.add(msg);
+    builder.add(const [0x00, 0x00, 0xff, 0xff]);
+    return decoder.convert(builder.takeBytes());
+  }
+
+  List<int> processOutgoingMessage(List<int> msg) {
+    var c = encoder.convert(msg);
+    c = c.sublist(0, c.length - 4);
+    return c;
+  }
+}
 
 // TODO(ajohnsen): Make this transformer reusable.
 class _WebSocketOutgoingTransformer implements StreamTransformer, EventSink {
   final _WebSocketImpl webSocket;
   EventSink _eventSink;
 
+  _WebSocketPerMessageDeflateHelper _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);
+        }
       }
     } else {
       opcode = _WebSocketOpcode.TEXT;
     }

[Søren Gjesse, 2015/05/26 09:11:33]

Move the compression down here to avoid the duplication (checking if data !=
null).

     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;
@@ skipping 237 matching lines @@
   bool _writeClosed = false;
   int _closeCode;
   String _closeReason;
   Duration _pingInterval;
   Timer _pingTimer;
   _WebSocketConsumer _consumer;
 
   int _outCloseCode;
   String _outCloseReason;
   Timer _closeTimer;
+  _WebSocketPerMessageDeflateHelper _deflateHelper;
 
   static final HttpClient _httpClient = new HttpClient();
 
   static Future<WebSocket> connect(
       String url, Iterable<String> protocols, Map<String, dynamic> headers) {
     Uri uri = Uri.parse(url);
     if (uri.scheme != "ws" && uri.scheme != "wss") {
       throw new WebSocketException("Unsupported URL scheme '${uri.scheme}'");
     }
 
@@ skipping 23 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/05/26 09:11:33]

There should be some way of controlling aspects of the per message compression.

E.g. for the context takeover and window bits. Not sure how to best do this,
maybe a CompressionOptions object that can be passed with some pre-defined
instances, e.g.

class CompressionOptions {
  CompressionOptions DEFAULT = const CompressionOptions(...);
  CompressionOptions SAVE_CLIENT_MEMORY = const CompressionOptions(...);
  ...

  CompressionOptions(...);
}

         if (protocols != null) {
           request.headers.add("Sec-WebSocket-Protocol", protocols.toList());
         }
         return request.close();
       })
       .then((response) {
         void error(String message) {
           // Flush data.
           response.detachSocket().then((socket) {
             socket.destroy();
@@ skipping 18 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');
+
+        String extensionHeader = response.headers.value('Sec-WebSocket-Extensions');
+
+        if (extensionHeader == null) {
+          extensionHeader = "";
+        }
+
+        Iterable<List<String>> extensions = extensionHeader.split(", ").map((it) => it.split("; "));
+
+        _WebSocketPerMessageDeflateHelper deflateHelper;
+
+        if (extensions.any((x) => x[0] == "permessage-deflate")) {
+          response.headers.add("Sec-WebSocket-Extensions", "permessage-deflate");

[Søren Gjesse, 2015/05/26 09:11:33]

Even for the simplest implementation you need to support
'client_no_context_takeover'. The spec says:

A client MUST support the "client_no_context_takeover" extension parameter in an
extension negotiation response

in section 8.1.1.2.

+          deflateHelper = new _WebSocketPerMessageDeflateHelper();
+        }
+
         return response.detachSocket()
-            .then((socket) => new _WebSocketImpl._fromSocket(socket, protocol));
+            .then((socket) => new _WebSocketImpl._fromSocket(socket, protocol)
+              .._deflateHelper = deflateHelper);
       });
   }
 
   _WebSocketImpl._fromSocket(this._socket, this.protocol,
                              [this._serverSide = false]) {
     _consumer = new _WebSocketConsumer(this, _socket);
     _sink = new _StreamSinkImpl(_consumer);
     _readyState = WebSocket.OPEN;
 
     var transformer = new _WebSocketProtocolTransformer(_serverSide);
+    transformer._deflateHelper = _deflateHelper;
     _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));
   }
 }