Initial Implementation of WebSocket Compression

sdk/lib/io/websocket_impl.dart

Index: sdk/lib/io/websocket_impl.dart
diff --git a/sdk/lib/io/websocket_impl.dart b/sdk/lib/io/websocket_impl.dart
index 265760690923f502512435c8e6d8df3d1fada22a..5b4168bdc86e6beaffe3f639bb6ff37ebf2cae64 100644
--- a/sdk/lib/io/websocket_impl.dart
+++ b/sdk/lib/io/websocket_impl.dart
@@ -38,8 +38,8 @@ class _WebSocketOpcode {
  * 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?
@@ -71,6 +71,8 @@ class _WebSocketProtocolTransformer implements StreamTransformer, EventSink {
   final List _maskingBytes = new List(4);
   final BytesBuilder _payload = new BytesBuilder(copy: false);
 
+  _WebSocketPerMessageDeflateHelper _deflateHelper;
+
   _WebSocketProtocolTransformer([this._serverSide = false]);
 
   Stream bind(Stream stream) {
@@ -286,10 +288,18 @@ class _WebSocketProtocolTransformer implements StreamTransformer, EventSink {
     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;
@@ -404,10 +414,26 @@ class _WebSocketTransformerImpl implements WebSocketTransformer {
       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'];
@@ -464,12 +490,38 @@ class _WebSocketTransformerImpl implements WebSocketTransformer {
   }
 }
 
+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) {
@@ -499,12 +551,20 @@ class _WebSocketOutgoingTransformer implements StreamTransformer, EventSink {
       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;
@@ -766,6 +826,7 @@ class _WebSocketImpl extends Stream with _ServiceObject implements WebSocket {
   int _outCloseCode;
   String _outCloseReason;
   Timer _closeTimer;
+  _WebSocketPerMessageDeflateHelper _deflateHelper;
 
   static final HttpClient _httpClient = new HttpClient();
 
@@ -809,7 +870,8 @@ class _WebSocketImpl extends Stream with _ServiceObject implements WebSocket {
             ..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());
         }
@@ -848,8 +910,25 @@ class _WebSocketImpl extends Stream with _ServiceObject implements WebSocket {
           }
         }
         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);
       });
   }
 
@@ -860,6 +939,7 @@ class _WebSocketImpl extends Stream with _ServiceObject implements WebSocket {
     _readyState = WebSocket.OPEN;
 
     var transformer = new _WebSocketProtocolTransformer(_serverSide);
+    transformer._deflateHelper = _deflateHelper;
     _subscription = _socket.transform(transformer).listen(
         (data) {
           if (data is _WebSocketPing) {