Per-message deflate as described in RFC 7692

Github issue #5

For my work-load (somewhat redundant JSON) it reduces the transmitted
message sizes to nearly 1/10th their uncompressed sizes so it will be
a huge bandwidth saver.

It basically does the bare minimum to be RFC 7692 compatible. I have
tested the compression with chromium and websocketpp 0.7.0 and seasocks
seems to compress/decompress messages properly in both directions.

Current limitations:
  - Can't configure the *_no_context_takeover or *_max_window_bits
    extensions. It will only negotiate the default (context takeover
    and max window bits in both directions)
  - All transmitted messages will be compressed. Ideally you should
    be able to opt-out of compressing particular messages (but it
    does support clients that compress particular messages)
  - Only ever issues one DEFLATE block per message since the
    seasocks API doesn't support incremental message construction
    (but I believe it does support receiving multiple blocks from
    clients that do)
  - Various tuning parameters such as buffer size and memLevel
    are hard-coded. Some apps might like to customise them
  - Requires linking to zlib even if no compression will be used
  - No tests :(

Author: hoytech

CMakeLists.txt

@@ -34,6 +34,9 @@ if( NOT PYTHON_BIN )
 endif()
 
 
+find_package(ZLIB REQUIRED)
+
+
 add_subdirectory("src/main/c")
 add_subdirectory("src/main/web")
 add_subdirectory("src/app/c")
@@ -43,4 +46,3 @@ if (UNITTESTS)
     enable_testing()
     add_subdirectory("src/test/c")
 endif ()
-

src/app/c/CMakeLists.txt

@@ -1,6 +1,6 @@
 macro(add_app _NAME)
     add_executable(${_NAME} ${_NAME}.cpp)
-    target_link_libraries(${_NAME} seasocks)
+    target_link_libraries(${_NAME} seasocks "${ZLIB_LIBRARIES}")
 endmacro()
 
 add_app(ph_test)

src/main/c/CMakeLists.txt

@@ -28,8 +28,8 @@ target_include_directories(seasocks PUBLIC .)
 target_link_libraries(seasocks PRIVATE ${CMAKE_THREAD_LIBS_INIT} embedded)
 
 add_library(seasocks_so SHARED $<TARGET_OBJECTS:seasocks_obj>)
-target_include_directories(seasocks_so PUBLIC .)
-target_link_libraries(seasocks_so PRIVATE ${CMAKE_THREAD_LIBS_INIT} embedded)
+target_include_directories(seasocks_so PUBLIC ${ZLIB_INCLUDE_DIRS} .)
+target_link_libraries(seasocks_so PRIVATE ${CMAKE_THREAD_LIBS_INIT} embedded "${ZLIB_LIBRARIES}")
 set_target_properties(seasocks_so PROPERTIES OUTPUT_NAME seasocks)
 
 install(TARGETS seasocks seasocks_so

src/main/c/Connection.cpp

@@ -59,6 +59,7 @@
 #include <unistd.h>
 #include <byteswap.h>
 #include <unordered_map>
+#include <memory>
 
 namespace {
 
@@ -570,7 +571,22 @@ void Connection::send(const uint8_t* data, size_t length) {
 
 void Connection::sendHybi(uint8_t opcode, const uint8_t* webSocketResponse, size_t messageLength) {
     uint8_t firstByte = 0x80 | opcode;
+    if (_perMessageDeflate) firstByte |= 0x40;
     if (!write(&firstByte, 1, false)) return;
+
+    if (_perMessageDeflate) {
+        std::vector<uint8_t> compressed;
+
+        zlibContext->deflate(webSocketResponse, messageLength, compressed);
+
+        LS_DEBUG(_logger, "Compression result: " << messageLength << " bytes -> " << compressed.size() << " bytes");
+        sendHybiData(compressed.data(), compressed.size());
+    } else {
+        sendHybiData(webSocketResponse, messageLength);
+    }
+}
+
+void Connection::sendHybiData(const uint8_t* webSocketResponse, size_t messageLength) {
     if (messageLength < 126) {
         uint8_t nextByte = messageLength; // No MASK bit set.
         if (!write(&nextByte, 1, false)) return;
@@ -637,7 +653,38 @@ void Connection::handleHybiWebSocket() {
     bool done = false;
     while (!done) {
         std::vector<uint8_t> decodedMessage;
-        switch (decoder.decodeNextMessage(decodedMessage)) {
+        bool deflateNeeded = false;
+
+        auto messageState = decoder.decodeNextMessage(decodedMessage, deflateNeeded);
+
+        if (deflateNeeded) {
+            if (!_perMessageDeflate) {
+                LS_WARNING(_logger, "Received deflated hybi frame but deflate wasn't negotiated");
+                closeInternal();
+                return;
+            }
+
+            size_t compressed_size = decodedMessage.size();
+
+            std::vector<uint8_t> decompressed;
+            int zlibError;
+
+            // Note: inflate() alters decodedMessage
+            bool success = zlibContext->inflate(decodedMessage, decompressed, zlibError);
+
+            if (!success) {
+                LS_WARNING(_logger, "Decompression error from zlib: " << zlibError);
+                closeInternal();
+                return;
+            }
+
+            LS_DEBUG(_logger, "Decompression result: " << compressed_size << " bytes -> " << decodedMessage.size() << " bytes");
+
+            decodedMessage.swap(decompressed);
+        }
+
+
+        switch (messageState) {
         default:
             closeInternal();
             LS_WARNING(_logger, "Unknown HybiPacketDecoder state");
@@ -809,6 +856,10 @@ bool Connection::processHeaders(uint8_t* first, uint8_t* last) {
             return send404();
         }
         verb = Request::Verb::WebSocket;
+
+        if (_server.server().getPerMessageDeflateEnabled() && headers.count("Sec-WebSocket-Extensions")) {
+            parsePerMessageDeflateHeader(headers["Sec-WebSocket-Extensions"]);
+        }
     }
 
     _request.reset(new PageRequest(_address, requestUri, _server.server(),
@@ -975,6 +1026,7 @@ bool Connection::handleHybiHandshake(
     bufferLine("Upgrade: websocket");
     bufferLine("Connection: Upgrade");
     bufferLine("Sec-WebSocket-Accept: " + getAcceptKey(webSocketKey));
+    if (_perMessageDeflate) bufferLine("Sec-WebSocket-Extensions: permessage-deflate");
     bufferLine("");
     flush();
 
@@ -985,6 +1037,19 @@ bool Connection::handleHybiHandshake(
     return true;
 }
 
+void Connection::parsePerMessageDeflateHeader(const std::string& header) {
+    for (auto &extField : seasocks::split(header, ';')) {
+        while (!extField.empty() && isspace(extField[0]))
+            extField = extField.substr(1);
+
+        if (seasocks::caseInsensitiveSame(extField, "permessage-deflate")) {
+            LS_INFO(_logger, "Enabling per-message deflate");
+            _perMessageDeflate = true;
+            zlibContext = std::unique_ptr<ZlibContext>(new ZlibContext());
+        }
+    }
+}
+
 bool Connection::parseRange(const std::string& rangeStr, Range& range) const {
     size_t minusPos = rangeStr.find('-');
     if (minusPos == std::string::npos) {

src/main/c/HybiPacketDecoder.cpp

@@ -39,7 +39,7 @@ HybiPacketDecoder::HybiPacketDecoder(Logger& logger,
 }
 
 HybiPacketDecoder::MessageState HybiPacketDecoder::decodeNextMessage(
-        std::vector<uint8_t>& messageOut) {
+        std::vector<uint8_t>& messageOut, bool& deflateNeeded) {
     if (_messageStart + 1 >= _buffer.size()) {
         return MessageState::NoMessage;
     }
@@ -49,10 +49,15 @@ HybiPacketDecoder::MessageState HybiPacketDecoder::decodeNextMessage(
         LS_WARNING(&_logger, "Received hybi frame without FIN bit set - unsupported");
         return MessageState::Error;
     }
-    if ((_buffer[_messageStart] & (7<<4)) != 0) {
+
+    auto reservedBits = _buffer[_messageStart] & (7<<4);
+    if ((reservedBits & 0x30) != 0) {
         LS_WARNING(&_logger, "Received hybi frame with reserved bits set - error");
         return MessageState::Error;
     }
+
+    deflateNeeded = !!(reservedBits & 0x40);
+
     auto opcode = static_cast<Opcode>(_buffer[_messageStart] & 0xf);
     size_t payloadLength = _buffer[_messageStart + 1] & 0x7fu;
     auto maskBit = _buffer[_messageStart + 1] & 0x80;

src/main/c/Server.cpp

@@ -592,6 +592,11 @@ void Server::setMaxKeepAliveDrops(int maxKeepAliveDrops) {
     _maxKeepAliveDrops = maxKeepAliveDrops;
 }
 
+void Server::setPerMessageDeflateEnabled(bool enabled) {
+    LS_INFO(_logger, "Setting per-message deflate to " << (enabled ? "enabled" : "disabled"));
+    _perMessageDeflateEnabled = enabled;
+}
+
 void Server::checkThread() const {
     auto thisTid = gettid();
     if (thisTid != _threadId) {

src/main/c/internal/HybiPacketDecoder.h

@@ -59,7 +59,11 @@ class HybiPacketDecoder {
         Pong,
         Close
     };
-    MessageState decodeNextMessage(std::vector<uint8_t>& messageOut);
+    MessageState decodeNextMessage(std::vector<uint8_t>& messageOut, bool& deflateNeeded);
+    MessageState decodeNextMessage(std::vector<uint8_t>& messageOut) {
+        bool ignore;
+        return decodeNextMessage(messageOut, ignore);
+    }
 
     size_t numBytesDecoded() const;
 };

src/main/c/internal/ZlibContext.h

@@ -0,0 +1,111 @@
+#pragma once
+
+#include <zlib.h>
+
+#include <vector>
+#include <stdexcept>
+
+
+namespace seasocks {
+
+class ZlibContext {
+public:
+    ZlibContext(const ZlibContext&) = delete;
+    ZlibContext& operator=(const ZlibContext&) = delete;
+
+    ZlibContext(int deflateBits=15, int inflateBits=15, int memLevel=6) {
+        int ret;
+
+        deflateStream.zalloc = Z_NULL;
+        deflateStream.zfree = Z_NULL;
+        deflateStream.opaque = Z_NULL;
+
+        ret = ::deflateInit2(
+            &deflateStream,
+            Z_DEFAULT_COMPRESSION,
+            Z_DEFLATED,
+            deflateBits * -1,
+            memLevel,
+            Z_DEFAULT_STRATEGY
+        );
+
+        if (ret != Z_OK) {
+            throw std::runtime_error("error initialising zlib deflater");
+        }
+
+        inflateStream.zalloc = Z_NULL;
+        inflateStream.zfree = Z_NULL;
+        inflateStream.opaque = Z_NULL;
+        inflateStream.avail_in = 0;
+        inflateStream.next_in = Z_NULL;
+
+        ret = ::inflateInit2(
+            &inflateStream,
+            inflateBits * -1
+        );
+
+        if (ret != Z_OK) {
+            ::deflateEnd(&deflateStream);
+            throw std::runtime_error("error initialising zlib inflater");
+        }
+
+        streamsInitialised = true;
+    }
+
+    ~ZlibContext() {
+        if (!streamsInitialised) return;
+        ::deflateEnd(&deflateStream);
+        ::inflateEnd(&inflateStream);
+    }
+
+    void deflate(const uint8_t* input, size_t inputLen, std::vector<uint8_t>& output) {
+        deflateStream.next_in = (unsigned char *)input;
+        deflateStream.avail_in = inputLen;
+
+        do {
+            deflateStream.next_out = buffer;
+            deflateStream.avail_out = sizeof(buffer);
+
+            (void) ::deflate(&deflateStream, Z_SYNC_FLUSH);
+
+            output.insert(output.end(), buffer, buffer + sizeof(buffer) - deflateStream.avail_out);
+        } while (deflateStream.avail_out == 0);
+
+        // Remove 4-byte tail end prior to transmission (see RFC 7692, section 7.2.1)
+        output.resize(output.size() - 4);
+    }
+
+    // WARNING: inflate() alters input
+    bool inflate(std::vector<uint8_t>& input, std::vector<uint8_t>& output, int& zlibError) {
+        // Append 4 octets prior to decompression (see RFC 7692, section 7.2.2)
+        uint8_t tail_end[4] = {0x00, 0x00, 0xff, 0xff};
+        input.insert(input.end(), tail_end, tail_end + 4);
+
+        inflateStream.next_in = input.data();
+        inflateStream.avail_in = input.size();
+
+        do {
+            inflateStream.next_out = buffer;
+            inflateStream.avail_out = sizeof(buffer);
+
+            int ret = ::inflate(&inflateStream, Z_SYNC_FLUSH);
+
+            if (ret != Z_OK && ret != Z_STREAM_END) {
+                zlibError = ret;
+                return false;
+            }
+
+            output.insert(output.end(), buffer, buffer + sizeof(buffer) - inflateStream.avail_out);
+        } while (inflateStream.avail_out == 0);
+
+        return true;
+    }
+
+private:
+    z_stream deflateStream;
+    z_stream inflateStream;
+    bool streamsInitialised = false;
+    uint8_t buffer[16384];
+};
+
+}  // namespace seasocks

src/main/c/seasocks/Connection.h

@@ -30,6 +30,8 @@
 #include "seasocks/ResponseWriter.h"
 #include "seasocks/TransferEncoding.h"
 
+#include "internal/ZlibContext.h"
+
 #include <netinet/in.h>
 
 #include <sys/socket.h>
@@ -126,6 +128,8 @@ class Connection : public WebSocket {
 
     void sendHybi(uint8_t opcode, const uint8_t* webSocketResponse,
                   size_t messageLength);
+    void sendHybiData(const uint8_t* webSocketResponse, size_t messageLength);
+
 
     bool sendResponse(std::shared_ptr<Response> response);
 
@@ -177,6 +181,10 @@ class Connection : public WebSocket {
     unsigned _chunk;
     std::shared_ptr<Writer> _writer;
 
+    void parsePerMessageDeflateHeader(const std::string& header);
+    bool _perMessageDeflate = false;
+    std::unique_ptr<ZlibContext> zlibContext;
+
     enum class State {
         INVALID,
         READING_HEADERS,

src/main/c/seasocks/Server.h

@@ -119,6 +119,9 @@ class Server : private ServerImpl {
     void setClientBufferSize(size_t bytesToBuffer);
     size_t clientBufferSize() const override { return _clientBufferSize; }
 
+    void setPerMessageDeflateEnabled(bool enabled);
+    bool getPerMessageDeflateEnabled() { return _perMessageDeflateEnabled;}
+
     class Runnable {
     public:
         virtual ~Runnable() {}
@@ -166,6 +169,9 @@ class Server : private ServerImpl {
     size_t _clientBufferSize;
     time_t _nextDeadConnectionCheck;
 
+    // Compression settings
+    bool _perMessageDeflateEnabled = false;
+
     struct WebSocketHandlerEntry {
         std::shared_ptr<WebSocket::Handler> handler;
         bool allowCrossOrigin;