Web Socket compression - take two

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 {
   static const int NONE = 0;
   static const int TEXT = 1;
   static const int BINARY = 2;
 }
 
-
 class _WebSocketOpcode {
   static const int CONTINUATION = 0;
   static const int TEXT = 1;
   static const int BINARY = 2;
   static const int RESERVED_3 = 3;
   static const int RESERVED_4 = 4;
   static const int RESERVED_5 = 5;
   static const int RESERVED_6 = 6;
   static const int RESERVED_7 = 7;
   static const int CLOSE = 8;
   static const int PING = 9;
   static const int PONG = 10;
   static const int RESERVED_B = 11;
   static const int RESERVED_C = 12;
   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;
+  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;
   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;
-        });
+    return new Stream.eventTransformed(stream, (EventSink eventSink) {
+      if (_eventSink != null) {
+        throw new StateError("WebSocket transformer already used.");
+      }
+      _eventSink = eventSink;
+      return this;
+    });
   }
 
   void addError(Object error, [StackTrace stackTrace]) =>
       _eventSink.addError(error, stackTrace);
 
   void close() => _eventSink.close();
 
   /**
    * Process data received from the underlying communication channel.
    */
   void add(Uint8List buffer) {
-    int count = buffer.length;
     int index = 0;
-    int lastIndex = count;
+    int lastIndex = buffer.length;
     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;
-          if ((byte & 0x70) != 0) {
-            // The RSV1, RSV2 bits RSV3 must be all zero.
+          _fin = (byte & FIN) != 0;
+          var rsv = (byte & (RSV1 | RSV2 | RSV3)) >> 4;

[Søren Gjesse, 2015/10/19 16:53:20]

I don't think this temp is needed (see below).

[butlermatt, 2015/10/22 19:36:08]

Done.

+
+          if (rsv != 0 && rsv != 4) {

[Søren Gjesse, 2015/10/19 16:53:20]

Just check ((byte & (RSV2 | RSV3)) != 0)

[butlermatt, 2015/10/22 19:36:08]

Done.

+            // The RSV1, RSV2, RSV3 bits must be all zero.

[Søren Gjesse, 2015/10/19 16:53:20]

This comment is no longer true.

[butlermatt, 2015/10/22 19:36:08]

Done.

             throw new WebSocketException("Protocol error");
           }
-          _opcode = (byte & 0xF);
+
+          if (rsv == 4) {

[Søren Gjesse, 2015/10/19 16:53:20]

4 -> RSV1 with no shift above.

[butlermatt, 2015/10/22 19:36:08]

Done.

+            _compressed = true;
+          } else {
+            _compressed = false;
+          }
+          _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);
+                  _opcode == _WebSocketOpcode.BINARY);
               if (_currentMessageType != _WebSocketMessageType.NONE) {
                 throw new WebSocketException("Protocol error");
               }
               _currentMessageType = _opcode;
             }
           } else if (_opcode >= _WebSocketOpcode.CLOSE &&
-                     _opcode <= _WebSocketOpcode.PONG) {
+              _opcode <= _WebSocketOpcode.PONG) {
             // Control frames cannot be fragmented.
             if (!_fin) throw new WebSocketException("Protocol error");
           } else {
             throw new WebSocketException("Protocol error");
           }
           _state = LEN_FIRST;
         } else if (_state == LEN_FIRST) {
           _masked = (byte & 0x80) != 0;
           _len = byte & 0x7F;
           if (_isControlFrame() && _len > 125) {
@@ skipping 28 matching lines @@
         } else {
           assert(_state == PAYLOAD);
           // The payload is not handled one byte at a time but in blocks.
           int payloadLength = min(lastIndex - index, _remainingPayloadBytes);
           _remainingPayloadBytes -= payloadLength;
           // Unmask payload if masked.
           if (_masked) {
             _unmask(index, payloadLength, buffer);
           }
           // Control frame and data frame share _payloads.
-          _payload.add(
-              new Uint8List.view(buffer.buffer, index, payloadLength));
+          _payload.add(new Uint8List.view(buffer.buffer, index, payloadLength));
           index += payloadLength;
           if (_isControlFrame()) {
             if (_remainingPayloadBytes == 0) _controlFrameEnd();
           } else {
             if (_currentMessageType != _WebSocketMessageType.TEXT &&
                 _currentMessageType != _WebSocketMessageType.BINARY) {
-                throw new WebSocketException("Protocol error");
+              throw new WebSocketException("Protocol error");
             }
             if (_remainingPayloadBytes == 0) _messageFrameEnd();
           }
 
           // Hack - as we always do index++ below.
           index--;
         }
       }
 
       // Move to the next byte.
@@ skipping 14 matching lines @@
       index += startOffset;
       length -= startOffset;
       final int blockCount = length ~/ BLOCK_SIZE;
       if (blockCount > 0) {
         // Create mask block.
         int mask = 0;
         for (int i = 3; i >= 0; i--) {
           mask = (mask << 8) | _maskingBytes[(_unmaskingIndex + i) & 3];
         }
         Int32x4 blockMask = new Int32x4(mask, mask, mask, mask);
-        Int32x4List blockBuffer = new Int32x4List.view(
-            buffer.buffer, index, blockCount);
+        Int32x4List blockBuffer =
+            new Int32x4List.view(buffer.buffer, index, blockCount);
         for (int i = 0; i < blockBuffer.length; i++) {
           blockBuffer[i] ^= blockMask;
         }
         final int bytes = blockCount * BLOCK_SIZE;
         index += bytes;
         length -= bytes;
       }
     }
     // Handle end.
     final int end = index + length;
@@ skipping 41 matching lines @@
         }
         _prepareForNextFrame();
       } else {
         _messageFrameEnd();
       }
     } else {
       _state = PAYLOAD;
     }
   }
 
+  static Utf8Decoder _utfDecoder;
+
   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()));
+          if (_utfDecoder == null) {
+            _utfDecoder = new Utf8Decoder();

[Søren Gjesse, 2015/10/19 16:53:20]

Why not just use UTF8.decode?

[butlermatt, 2015/10/22 19:36:09]

Done.

[alexandre.ardhuin, 2015/10/22 20:06:04]

Unlike `UTF8.encode` that uses const, `UTF8.decode` creates a new instance of
`Utf8Decoder` everytime `UTF8.decode` is call (see
https://github.com/dart-lang/sdk/blob/master/sdk/lib/convert/utf.dart#L69-L72).
That's why I think `const Utf8Decoder().convert(bytes)` is better.

+          }

[alexandre.ardhuin, 2015/10/13 20:22:57]

You can remove this `if` by either :
- `_eventSink.add(const Utf8Decoder().convert(bytes));`
- `static final Utf8Decoder _utfDecoder = new Utf8Decoder();` as "class variable
that’s declared as final is initialized the first time it’s used"
(https://www.dartlang.org/docs/dart-up-and-running/ch02.html#final-and-const)

[butlermatt, 2015/10/22 19:36:08]

Done.

+          _eventSink.add(_utfDecoder.convert(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 21 matching lines @@
 
       case _WebSocketOpcode.PONG:
         _eventSink.add(new _WebSocketPong(_payload.takeBytes()));
         break;
     }
     _prepareForNextFrame();
   }
 
   bool _isControlFrame() {
     return _opcode == _WebSocketOpcode.CLOSE ||
-           _opcode == _WebSocketOpcode.PING ||
-           _opcode == _WebSocketOpcode.PONG;
+        _opcode == _WebSocketOpcode.PING ||
+        _opcode == _WebSocketOpcode.PONG;
   }
 
   void _prepareForNextFrame() {
     if (_state != CLOSED && _state != FAILURE) _state = START;
     _fin = false;
     _opcode = -1;
     _len = -1;
     _remainingLenBytes = -1;
     _remainingMaskingKeyBytes = 4;
     _remainingPayloadBytes = -1;
     _unmaskingIndex = 0;
   }
 }
 
-
 class _WebSocketPing {
   final List<int> payload;
   _WebSocketPing([this.payload = null]);
 }
 
-
 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)
-            .then((WebSocket webSocket) => _controller.add(webSocket))
-            .catchError(_controller.addError);
+      _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();
+        ..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");
+        ..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);
       }
+
+      var deflate = _negotiateCompression(request, response, compression);
+
       response.headers.contentLength = 0;
-      return response.detachSocket()
-          .then((socket) => new _WebSocketImpl._fromSocket(
-                socket, protocol, true));
+      return response.detachSocket().then((socket) =>
+          new _WebSocketImpl._fromSocket(
+              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);
+      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) {
+    var extensionHeader = request.headers.value("Sec-WebSocket-Extensions");
+
+    if (extensionHeader == null) {
+      extensionHeader = "";
+    }
+
+    var hv = HeaderValue.parse(extensionHeader);
+    if (compression.enabled && hv.value == _WebSocketImpl.PER_MESSAGE_DEFLATE) {
+      var info = compression._createHeader(hv);
+
+      response.headers.add("Sec-WebSocket-Extensions", info[0]);
+      var serverNoContextTakeover =
+          hv.parameters.containsKey("server_no_context_takeover");
+      var clientNoContextTakeover =
+          hv.parameters.containsKey("client_no_context_takeover");
+      var deflate = new _WebSocketPerMessageDeflate(
+          serverNoContextTakeover: serverNoContextTakeover,
+          clientNoContextTakeover: clientNoContextTakeover,
+          serverMaxWindowBits: info[1],
+          clientMaxWindowBits: info[1],
+          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;
     });
     if (!isUpgrade) return false;
     String upgrade = request.headers.value(HttpHeaders.UPGRADE);
     if (upgrade == null || upgrade.toLowerCase() != "websocket") {
       return false;
     }
     String version = request.headers.value("Sec-WebSocket-Version");
     if (version == null || version != "13") {
       return false;
     }
     String key = request.headers.value("Sec-WebSocket-Key");
     if (key == null) {
       return false;
     }
     return true;
   }
 }
 
+class _WebSocketPerMessageDeflate {
+  bool serverNoContextTakeover;
+  bool clientNoContextTakeover;
+  int clientMaxWindowBits;
+  int serverMaxWindowBits;
+  bool serverSide;
+
+  _Filter decoder;
+  _Filter encoder;
+
+  _WebSocketPerMessageDeflate(
+      {this.clientMaxWindowBits,
+       this.serverMaxWindowBits,
+       this.serverNoContextTakeover: false,
+       this.clientNoContextTakeover: false,
+       this.serverSide: false}) {
+    if (clientMaxWindowBits == null) {

[Søren Gjesse, 2015/10/19 16:53:20]

Just add default values in the argument list?

[butlermatt, 2015/10/22 19:36:08]

Done.

+      clientMaxWindowBits = _WebSocketImpl.DEFAULT_WINDOW_BITS;
+    }
+
+    if (serverMaxWindowBits == null) {
+      serverMaxWindowBits = _WebSocketImpl.DEFAULT_WINDOW_BITS;
+    }
+  }
+
+  void _ensureDecoder() {
+    if (decoder == null) {
+      decoder = _Filter._newZLibInflateFilter(
+          serverSide ? clientMaxWindowBits : serverMaxWindowBits, null, true);
+    }
+  }
+
+  void _ensureEncoder() {
+    if (encoder == null) {
+      encoder = _Filter._newZLibDeflateFilter(
+          false,
+          ZLibOption.DEFAULT_LEVEL,
+          serverSide ? serverMaxWindowBits : clientMaxWindowBits,
+          ZLibOption.DEFAULT_MEM_LEVEL,
+          ZLibOption.STRATEGY_DEFAULT,
+          null,
+          true);
+    }
+  }
+
+  List<int> processIncomingMessage(List<int> msg) {
+    _ensureDecoder();
+
+    var data = [];
+    data.addAll(msg);
+    data.addAll(const [0x00, 0x00, 0xff, 0xff]);
+
+    decoder.process(data, 0, data.length);
+    var reuse =
+        !(serverSide ? clientNoContextTakeover : serverNoContextTakeover);
+    var result = [];
+    var out;
+
+    while ((out = decoder.processed(flush: reuse)) != null) {
+      result.addAll(out);
+    }
+
+    decoder.processed(flush: reuse);
+
+    if (!reuse) {
+      decoder.end();
+      decoder = null;
+    }
+    return result;
+  }
+
+  List<int> processOutgoingMessage(List<int> msg) {
+    _ensureEncoder();
+    var reuse =
+        !(serverSide ? serverNoContextTakeover : clientNoContextTakeover);
+    var result = [];
+    var buffer = new Uint8List(msg.length);

[Søren Gjesse, 2015/10/19 16:53:20]

Avoid new buffer if msg is Uint8List.

[butlermatt, 2015/10/22 19:36:08]

Done.

+    var out;
+
+    for(var i = 0; i < msg.length; i++) {

[Søren Gjesse, 2015/10/19 16:53:20]

Space after "for".

[butlermatt, 2015/10/22 19:36:08]

Done.

+      if (msg[i] < 0 || 255 < msg[i]) {
+        throw new ArgumentError("List element is not a byte value "
+            "(value ${msg[i]} at index $i)");
+      }
+      buffer[i] = msg[i];
+    }
+
+    encoder.process(buffer, 0, buffer.length);
+
+    while ((out = encoder.processed(flush: reuse)) != null) {
+      result.addAll(out);
+    }
+
+    if (serverSide ? serverNoContextTakeover : clientNoContextTakeover) {
+      encoder.end();
+      encoder = null;
+    }
+
+    if (result.length > 4) {
+      result = result.sublist(0, result.length - 4);
+    }
+
+    return result;
+  }
+}
 
 // TODO(ajohnsen): Make this transformer reusable.
 class _WebSocketOutgoingTransformer implements StreamTransformer, EventSink {
   final _WebSocketImpl webSocket;
   EventSink _eventSink;
 
-  _WebSocketOutgoingTransformer(this.webSocket);
+  _WebSocketPerMessageDeflate _deflateHelper;
+
+  _WebSocketOutgoingTransformer(this.webSocket) {
+    _deflateHelper = webSocket._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;
-        });
+    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 (_utfEncoder == null) {
+          _utfEncoder = new Utf8Encoder();
+        }

[alexandre.ardhuin, 2015/10/13 20:22:57]

You can remove this `if` with `static final Utf8Encoder _utfEncoder = new
Utf8Encoder();` as "class variable
that’s declared as final is initialized the first time it’s used"
(https://www.dartlang.org/docs/dart-up-and-running/ch02.html#final-and-const)

[butlermatt, 2015/10/22 19:36:08]

Converted to UTF8.encode to match previous comment regarding UTF8.decode

+        data = _utfEncoder.convert(message);
       } else {
-        if (message is !List<int>) {
+        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;
     }
     addFrame(opcode, data);
   }
 
+  static Utf8Encoder _utfEncoder;
+
   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);
+  void addFrame(int opcode, List<int> data) => createFrame(
+          opcode,
+          data,
+          webSocket._serverSide,
+          _deflateHelper != null &&
+              (opcode == _WebSocketOpcode.TEXT ||
+                  opcode == _WebSocketOpcode.BINARY)).forEach((e) {
+        _eventSink.add(e);
+      });
 
-  static Iterable createFrame(int opcode, List<int> data, bool serverSide) {
-    bool mask = !serverSide;  // Masking not implemented for server.
+  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;
+
+    var rsv = 0;

[Søren Gjesse, 2015/10/19 16:53:20]

Use the RSVx constants.

[butlermatt, 2015/10/22 19:36:08]

Acknowledged.

+    if (compressed) {
+      rsv = 4;
+    }
+
     // Set FIN and opcode.
-    header[index++] = 0x80 | opcode;
+    var hoc = (1 << 7) | (rsv % 8) << 4 | (opcode % 128);

[Søren Gjesse, 2015/10/19 16:53:20]

How about

var hoc = FIN | (compressed ? RSV1 : 0) | (opcode & OPCODE);

and no rsv temp?

[butlermatt, 2015/10/22 19:36:08]

Done.

+
+    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;
     }
     if (mask) {
       header[1] |= 1 << 7;
       var maskBytes = _IOCrypto.getRandomBytes(4);
       header.setRange(index, index + 4, maskBytes);
       index += 4;
       if (data != null) {
         Uint8List list;
         // If this is a text message just do the masking inside the
         // encoded data.
         if (opcode == _WebSocketOpcode.TEXT && data is Uint8List) {
           list = data;
         } else {
           if (data is Uint8List) {
             list = new Uint8List.fromList(data);
           } else {
             list = new Uint8List(data.length);
             for (int i = 0; i < data.length; i++) {
               if (data[i] < 0 || 255 < data[i]) {
-                throw new ArgumentError(
-                    "List element is not a byte value "
+                throw new ArgumentError("List element is not a byte value "
                     "(value ${data[i]} at index $i)");
               }
               list[i] = data[i];
             }
           }
         }
         const int BLOCK_SIZE = 16;
         int blockCount = list.length ~/ BLOCK_SIZE;
         if (blockCount > 0) {
           // Create mask block.
           int mask = 0;
           for (int i = 3; i >= 0; i--) {
             mask = (mask << 8) | maskBytes[i];
           }
           Int32x4 blockMask = new Int32x4(mask, mask, mask, mask);
-          Int32x4List blockBuffer = new Int32x4List.view(
-              list.buffer, 0, blockCount);
+          Int32x4List blockBuffer =
+              new Int32x4List.view(list.buffer, 0, blockCount);
           for (int i = 0; i < blockBuffer.length; i++) {
             blockBuffer[i] ^= blockMask;
           }
         }
         // Handle end.
         for (int i = blockCount * BLOCK_SIZE; i < list.length; i++) {
           list[i] ^= maskBytes[i & 3];
         }
         data = list;
       }
     }
     assert(index == headerSize);
     if (data == null) {
       return [header];
     } else {
       return [header, data];
     }
   }
 }
 
-
 class _WebSocketConsumer implements StreamConsumer {
   final _WebSocketImpl webSocket;
   final Socket socket;
   StreamController _controller;
   StreamSubscription _subscription;
   bool _issuedPause = false;
   bool _closed = false;
   Completer _closeCompleter = new Completer();
   Completer _completer;
 
@@ skipping 24 matching lines @@
   void _cancel() {
     if (_subscription != null) {
       var subscription = _subscription;
       _subscription = null;
       subscription.cancel();
     }
   }
 
   _ensureController() {
     if (_controller != null) return;
-    _controller = new StreamController(sync: true,
-                                       onPause: _onPause,
-                                       onResume: _onResume,
-                                       onCancel: _onListen);
-    var stream = _controller.stream.transform(
-        new _WebSocketOutgoingTransformer(webSocket));
-    socket.addStream(stream)
-        .then((_) {
-          _done();
-          _closeCompleter.complete(webSocket);
-        }, onError: (error, StackTrace stackTrace) {
-          _closed = true;
-          _cancel();
-          if (error is ArgumentError) {
-            if (!_done(error, stackTrace)) {
-              _closeCompleter.completeError(error, stackTrace);
-            }
-          } else {
-            _done();
-            _closeCompleter.complete(webSocket);
-          }
-        });
+    _controller = new StreamController(
+        sync: true,
+        onPause: _onPause,
+        onResume: _onResume,
+        onCancel: _onListen);
+    var stream = _controller.stream
+        .transform(new _WebSocketOutgoingTransformer(webSocket));
+    socket.addStream(stream).then((_) {
+      _done();
+      _closeCompleter.complete(webSocket);
+    }, onError: (error, StackTrace stackTrace) {
+      _closed = true;
+      _cancel();
+      if (error is ArgumentError) {
+        if (!_done(error, stackTrace)) {
+          _closeCompleter.completeError(error, stackTrace);
+        }
+      } else {
+        _done();
+        _closeCompleter.complete(webSocket);
+      }
+    });
   }
 
   bool _done([error, StackTrace stackTrace]) {
     if (_completer == null) return false;
     if (error != null) {
       _completer.completeError(error, stackTrace);
     } else {
       _completer.complete(webSocket);
     }
     _completer = null;
     return true;
   }
 
   Future addStream(var stream) {
     if (_closed) {
       stream.listen(null).cancel();
       return new Future.value(webSocket);
     }
     _ensureController();
     _completer = new Completer();
-    _subscription = stream.listen(
-        (data) {
-          _controller.add(data);
-        },
-        onDone: _done,
-        onError: _done,
-        cancelOnError: true);
+    _subscription = stream.listen((data) {
+      _controller.add(data);
+    }, onDone: _done, onError: _done, cancelOnError: true);
     if (_issuedPause) {
       _subscription.pause();
       _issuedPause = false;
     }
     return _completer.future;
   }
 
   Future close() {
     _ensureController();
     Future closeSocket() {
       return socket.close().catchError((_) {}).then((_) => webSocket);
     }
     _controller.close();
     return _closeCompleter.future.then((_) => closeSocket());
   }
 
   void add(data) {
     if (_closed) return;
     _ensureController();
     _controller.add(data);
   }
 
   void closeSocket() {
     _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;
   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);
     }
     String nonce = _CryptoUtils.bytesToBase64(nonceData);
 
-    uri = new Uri(scheme: uri.scheme == "wss" ? "https" : "http",
-                  userInfo: uri.userInfo,
-                  host: uri.host,
-                  port: uri.port,
-                  path: uri.path,
-                  query: uri.query,
-                  fragment: uri.fragment);
-    return _httpClient.openUrl("GET", uri)
-      .then((request) {
-        if (uri.userInfo != null && !uri.userInfo.isEmpty) {
-          // If the URL contains user information use that for basic
-          // authorization.
-          String auth =
-          _CryptoUtils.bytesToBase64(UTF8.encode(uri.userInfo));
-          request.headers.set(HttpHeaders.AUTHORIZATION, "Basic $auth");
+    uri = new Uri(
+        scheme: uri.scheme == "wss" ? "https" : "http",
+        userInfo: uri.userInfo,
+        host: uri.host,
+        port: uri.port,
+        path: uri.path,
+        query: uri.query,
+        fragment: uri.fragment);
+    return _httpClient.openUrl("GET", uri).then((request) {
+      if (uri.userInfo != null && !uri.userInfo.isEmpty) {
+        // If the URL contains user information use that for basic
+        // authorization.
+        String auth = _CryptoUtils.bytesToBase64(UTF8.encode(uri.userInfo));
+        request.headers.set(HttpHeaders.AUTHORIZATION, "Basic $auth");
+      }
+      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");
+      if (protocols != null) {
+        request.headers.add("Sec-WebSocket-Protocol", protocols.toList());
+      }
+
+      if(compression.enabled) {

[Søren Gjesse, 2015/10/19 16:53:20]

Space after "if"

[butlermatt, 2015/10/22 19:36:08]

Done.

+        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);
+      }
+      if (response.statusCode != HttpStatus.SWITCHING_PROTOCOLS ||
+          response.headers[HttpHeaders.CONNECTION] == null ||
+          !response.headers[HttpHeaders.CONNECTION]
+              .any((value) => value.toLowerCase() == "upgrade") ||
+          response.headers.value(HttpHeaders.UPGRADE).toLowerCase() !=
+              "websocket") {
+        error("Connection to '$uri' was not upgraded to websocket");
+      }
+      String accept = response.headers.value("Sec-WebSocket-Accept");
+      if (accept == null) {
+        error("Response did not contain a 'Sec-WebSocket-Accept' header");
+      }
+      _SHA1 sha1 = new _SHA1();
+      sha1.add("$nonce$_webSocketGUID".codeUnits);
+      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");
         }
-        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");
-        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();
-          });
-          throw new WebSocketException(message);
-        }
-        if (response.statusCode != HttpStatus.SWITCHING_PROTOCOLS ||
-            response.headers[HttpHeaders.CONNECTION] == null ||
-            !response.headers[HttpHeaders.CONNECTION].any(
-                (value) => value.toLowerCase() == "upgrade") ||
-            response.headers.value(HttpHeaders.UPGRADE).toLowerCase() !=
-                "websocket") {
-          error("Connection to '$uri' was not upgraded to websocket");
-        }
-        String accept = response.headers.value("Sec-WebSocket-Accept");
-        if (accept == null) {
-          error("Response did not contain a 'Sec-WebSocket-Accept' header");
-        }
-        _SHA1 sha1 = new _SHA1();
-        sha1.add("$nonce$_webSocketGUID".codeUnits);
-        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, protocol));
-      });
+      }
+      var protocol = response.headers.value('Sec-WebSocket-Protocol');
+
+      _WebSocketPerMessageDeflate deflate =
+          negotiateClientCompression(response, compression);
+
+      return response.detachSocket().then((socket) =>
+          new _WebSocketImpl._fromSocket(
+              socket, protocol, compression, false, deflate));
+    });
   }
 
-  _WebSocketImpl._fromSocket(this._socket, this.protocol,
-                             [this._serverSide = false]) {
+  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("; "));

[Søren Gjesse, 2015/10/19 16:53:20]

This seems fragile. I don't think the specification requires space after , or ;.
Also how about quoted values containing , or ; inside quotes?

E.g.:

Sec-WebSocket-Extensions: my_ext; x="per_message_deflate;
server_no_context_takeover; server_max_window_bits=7", per_message_deflate;
server_max_window_bits=15

I think a real parser is required here.

See my comment on this in the previous CL. You already did some of the suggested
changes to http_header.dart.

[butlermatt, 2015/10/22 19:36:08]

Done.

+
+    if (compression.enabled &&
+        extensions.any((x) => x[0] == PER_MESSAGE_DEFLATE)) {
+      var opts = extensions.firstWhere((x) => x[0] == PER_MESSAGE_DEFLATE);
+      var serverNoContextTakeover = opts.contains("server_no_context_takeover");
+      var clientNoContextTakeover = opts.contains("client_no_context_takeover");
+
+      int getWindowBits(String type) {
+        var o = opts.firstWhere((x) => x.startsWith("${type}_max_window_bits="),

[Søren Gjesse, 2015/10/19 16:53:20]

This is also fragile. Doesn't the spec allow for whitespace on both sides of the
'='?

[butlermatt, 2015/10/22 19:36:08]

Acknowledged.

+            orElse: () => null);
+
+        if (o == null) {
+          return DEFAULT_WINDOW_BITS;
+        }
+
+        try {
+          o = o.substring("${type}_max_window_bits=".length);
+          o = int.parse(o);

[Søren Gjesse, 2015/10/19 16:53:20]

According to the spec the value can be quoted.

[butlermatt, 2015/10/22 19:36:08]

Acknowledged.

+        } catch (e) {
+          return DEFAULT_WINDOW_BITS;
+        }
+
+        return o;
+      }
+
+      return new _WebSocketPerMessageDeflate(
+          clientMaxWindowBits: getWindowBits("client"),
+          serverMaxWindowBits: getWindowBits("server"),
+          clientNoContextTakeover: clientNoContextTakeover,
+          serverNoContextTakeover: serverNoContextTakeover);
+    }
+
+    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);
-    _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);
-          }
-        },
-        onError: (error) {
-          if (_closeTimer != null) _closeTimer.cancel();
-          if (error is FormatException) {
-            _close(WebSocketStatus.INVALID_FRAME_PAYLOAD_DATA);
-          } else {
-            _close(WebSocketStatus.PROTOCOL_ERROR);
-          }
-          // An error happened, set the close code set above.
-          _closeCode = _outCloseCode;
-          _closeReason = _outCloseReason;
-          _controller.close();
-        },
-        onDone: () {
-          if (_closeTimer != null) _closeTimer.cancel();
-          if (_readyState == WebSocket.OPEN) {
-            _readyState = WebSocket.CLOSING;
-            if (!_isReservedStatusCode(transformer.closeCode)) {
-              _close(transformer.closeCode, transformer.closeReason);
-            } else {
-              _close();
-            }
-            _readyState = WebSocket.CLOSED;
-          }
-          // Protocol close, use close code from transformer.
-          _closeCode = transformer.closeCode;
-          _closeReason = transformer.closeReason;
-          _controller.close();
-        },
-        cancelOnError: true);
+    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);
+      }
+    }, onError: (error, stackTrace) {
+      if (_closeTimer != null) _closeTimer.cancel();
+      if (error is FormatException) {
+        _close(WebSocketStatus.INVALID_FRAME_PAYLOAD_DATA);
+      } else {
+        _close(WebSocketStatus.PROTOCOL_ERROR);
+      }
+      // An error happened, set the close code set above.
+      _closeCode = _outCloseCode;
+      _closeReason = _outCloseReason;
+      _controller.close();
+    }, onDone: () {
+      if (_closeTimer != null) _closeTimer.cancel();
+      if (_readyState == WebSocket.OPEN) {
+        _readyState = WebSocket.CLOSING;
+        if (!_isReservedStatusCode(transformer.closeCode)) {
+          _close(transformer.closeCode, transformer.closeReason);
+        } else {
+          _close();
+        }
+        _readyState = WebSocket.CLOSED;
+      }
+      // Protocol close, use close code from transformer.
+      _closeCode = transformer.closeCode;
+      _closeReason = transformer.closeReason;
+      _controller.close();
+    }, cancelOnError: true);
     _subscription.pause();
-    _controller = new StreamController(sync: true,
-                                       onListen: _subscription.resume,
-                                       onCancel: () {
-                                         _subscription.cancel();
-                                         _subscription = null;
-                                       },
-                                       onPause: _subscription.pause,
-                                       onResume: _subscription.resume);
+    _controller = new StreamController(
+        sync: true, onListen: _subscription.resume, onCancel: () {
+      _subscription.cancel();
+      _subscription = null;
+    }, onPause: _subscription.pause, onResume: _subscription.resume);
 
     _webSockets[_serviceId] = this;
-    try { _socket._owner = this; } catch (_) {}
+    try {
+      _socket._owner = this;
+    } catch (_) {}
   }
 
   StreamSubscription listen(void onData(message),
-                            {Function onError,
-                             void onDone(),
-                             bool cancelOnError}) {
+      {Function onError, void onDone(), bool cancelOnError}) {
     return _controller.stream.listen(onData,
-                                     onError: onError,
-                                     onDone: onDone,
-                                     cancelOnError: cancelOnError);
+        onError: onError, onDone: onDone, cancelOnError: cancelOnError);
   }
 
   Duration get pingInterval => _pingInterval;
 
   void set pingInterval(Duration interval) {
     if (_writeClosed) return;
     if (_pingTimer != null) _pingTimer.cancel();
     _pingInterval = interval;
 
     if (_pingInterval == null) return;
@@ skipping 85 matching lines @@
         'type': '@Socket',
         'name': 'UserSocket',
         'user_name': 'UserSocket',
       };
     }
     return r;
   }
 
   static bool _isReservedStatusCode(int code) {
     return code != null &&
-           (code < WebSocketStatus.NORMAL_CLOSURE ||
+        (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));
+                code < WebSocketStatus.RESERVED_1015) ||
+            (code >= WebSocketStatus.RESERVED_1015 && code < 3000));
   }
 }