Add basic permessage-deflate, pass entire Autobahn

README.md

@@ -8,7 +8,7 @@
 </div>
 
 #### Tread lightly.
-This is the **development branch** of v0.15. Track progress & issues [here](https://github.com/uNetworking/issues/issues). Checkout [v0.14](https://github.com/uNetworking/uWebSockets/tree/v0.14) for something stable.
+This is the **completely broken development branch** of v0.15. Track progress & issues [here](https://github.com/uNetworking/issues/issues). Checkout [v0.14](https://github.com/uNetworking/uWebSockets/tree/v0.14) for something stable.
 
 #### Express yourself briefly.
 ```c++

misc/15.pro

@@ -31,8 +31,10 @@ HEADERS += \
     ../src/WebSocket.h \
     ../src/WebSocketData.h \
     ../src/WebSocketContext.h \
-    ../src/WebSocketContextData.h
+    ../src/WebSocketContextData.h \
+    ../src/WebSocketExtensions.h \
+    ../src/PerMessageDeflate.h
 
 INCLUDEPATH += ../uSockets/src ../src
 QMAKE_CXXFLAGS += -fsanitize=address
-LIBS += -lasan -pthread -lssl -lcrypto -lstdc++fs
+LIBS += -lasan -pthread -lssl -lcrypto -lz -lstdc++fs

misc/main.cpp

@@ -12,6 +12,9 @@ int main(int argc, char **argv) {
     uWS::App().get("/hello", [](auto *res, auto *req) {
         res->end("Hello HTTP!");
     }).ws<void>("/*", {
+        /*.compression = */true,
+        /*.maxPayloadLength*/
+
         /*.open = */[](auto *ws, auto *req) {
 
         },

src/App.h

@@ -24,6 +24,7 @@
 #include "HttpResponse.h"
 #include "WebSocketContext.h"
 #include "WebSocket.h"
+#include "WebSocketExtensions.h"
 
 #include "libwshandshake.hpp"
 
@@ -49,6 +50,7 @@ public:
     }
 
     struct WebSocketBehavior {
+        bool compression = false;
         std::function<void(uWS::WebSocket<SSL, true> *, HttpRequest *)> open = nullptr;
         std::function<void(uWS::WebSocket<SSL, true> *, std::string_view, uWS::OpCode)> message = nullptr;
     };
@@ -58,6 +60,15 @@ public:
         /* Every route has its own websocket context with its own behavior and user data type */
         auto *webSocketContext = WebSocketContext<SSL, true>::create(Loop::defaultLoop(), (typename StaticDispatch<SSL>::SOCKET_CONTEXT_TYPE *) httpContext);
 
+        /* If we are the first one to use compression, initialize it */
+        if (behavior.compression) {
+            LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(webSocketContext->getSocketContext()));
+
+            if (!loopData->inflationStream) {
+                loopData->inflationStream = new InflationStream;
+            }
+        }
+
         /* Copy all handlers */
         webSocketContext->getExt()->messageHandler = behavior.message;
 
@@ -65,9 +76,6 @@ public:
             /* If we have this header set, it's a websocket */
             std::string_view secWebSocketKey = req->getHeader("sec-websocket-key");
             if (secWebSocketKey.length()) {
-
-                // todo: negotiate extensions such as compression here and pass autobahn fully
-
                 // note: OpenSSL can be used here to speed this up somewhat
                 char secWebSocketAccept[29] = {};
                 WebSocketHandshake::generate(secWebSocketKey.data(), secWebSocketAccept);
@@ -75,15 +83,38 @@ public:
                 res->writeStatus("101 Switching Protocols")
                     ->writeHeader("Upgrade", "websocket")
                     ->writeHeader("Connection", "Upgrade")
-                    ->writeHeader("Sec-WebSocket-Accept", secWebSocketAccept)
-                    ->end();
+                    ->writeHeader("Sec-WebSocket-Accept", secWebSocketAccept);
+
+                /* Negotiate compression */
+                bool perMessageDeflate = false;
+                if (behavior.compression) {
+                    std::string_view extensions = req->getHeader("sec-websocket-extensions");
+                    if (extensions.length()) {
+                        // basically: parse<isServer>(options, extensions)
+                        ExtensionsNegotiator<true> extensionsNegotiator(PERMESSAGE_DEFLATE | CLIENT_NO_CONTEXT_TAKEOVER); // take options
+                        extensionsNegotiator.readOffer(extensions);
+
+                        //std::cout << extensions << " => " << extensionsNegotiator.generateOffer() << std::endl;
+
+                        /* Todo: remove these mid string copies */
+                        res->writeHeader("Sec-WebSocket-Extensions", extensionsNegotiator.generateOffer());
+
+                        /* Did we negotiate permessage-deflate? */
+                        if (extensionsNegotiator.getNegotiatedOptions() & PERMESSAGE_DEFLATE) {
+                            perMessageDeflate = true;
+                        }
+                    }
+                }
+
+                /* Add mark, we don't want to end anything */
+                res->writeHeader("WebSocket-Server", "uWebSockets")->end();
 
                 /* Adopting a socket invalidates it, do not rely on it directly to carry any data */
                 WebSocket<SSL, true> *webSocket = (WebSocket<SSL, true> *) StaticDispatch<SSL>::static_dispatch(us_ssl_socket_context_adopt_socket, us_socket_context_adopt_socket)(
                             (typename StaticDispatch<SSL>::SOCKET_CONTEXT_TYPE *) webSocketContext, (typename StaticDispatch<SSL>::SOCKET_TYPE *) res, /*sizeof(WebSocketData)*/ 150);
 
                 httpContext->upgradeToWebSocket(
-                            webSocket->init()
+                            webSocket->init(perMessageDeflate)
                             );
 
                 if (behavior.open) {

src/LoopData.h

@@ -22,6 +22,8 @@
 #include <vector>
 #include <mutex>
 
+#include "PerMessageDeflate.h"
+
 namespace uWS {
 
 struct Loop;
@@ -43,6 +45,9 @@ public:
     char *corkBuffer = new char[CORK_BUFFER_SIZE];
     int corkOffset = 0;
     bool corked = false;
+
+    /* Compression data */
+    InflationStream *inflationStream = nullptr;
 };
 
 }

src/PerMessageDeflate.h

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2018 Alex Hultman and contributors.
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ *     http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// inflationStream? Ciompression
+
+#ifndef PERMESSAGEDEFLATE_H
+#define PERMESSAGEDEFLATE_H
+/* Do not compile this module if we don't want it */
+#ifndef UWS_NO_ZLIB
+
+#include <zlib.h>
+
+#include <string>
+#include <iostream>
+
+#define LARGE_BUFFER_SIZE 16000 // fix this
+
+// we also need DeflationStream
+
+// the loop holds one of these
+struct InflationStream {
+
+    std::string dynamicZlibBuffer;
+    z_stream inflationStream = {};
+    char *zlibBuffer;
+
+    InflationStream() {
+        std::cout << "Initliazing shared compression" << std::endl;
+        zlibBuffer = (char *) malloc(LARGE_BUFFER_SIZE);
+
+        inflateInit2(&inflationStream, -15);
+    }
+
+    std::string_view inflate(std::string_view compressed) {
+
+        int maxPayload = 160000; // todo: fix this
+
+        dynamicZlibBuffer.clear();
+
+        inflationStream.next_in = (Bytef *) compressed.data();
+        inflationStream.avail_in = (unsigned int) compressed.length();
+
+        int err;
+        do {
+            inflationStream.next_out = (Bytef *) zlibBuffer;
+            inflationStream.avail_out = LARGE_BUFFER_SIZE;
+            err = ::inflate(&inflationStream, Z_FINISH);
+            if (!inflationStream.avail_in) {
+                break;
+            }
+
+            dynamicZlibBuffer.append(zlibBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out);
+        } while (err == Z_BUF_ERROR && dynamicZlibBuffer.length() <= maxPayload);
+
+        inflateReset(&inflationStream);
+
+        if ((err != Z_BUF_ERROR && err != Z_OK) || dynamicZlibBuffer.length() > maxPayload) {
+            return {nullptr, 0};
+        }
+
+        if (dynamicZlibBuffer.length()) {
+            dynamicZlibBuffer.append(zlibBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out);
+            return {dynamicZlibBuffer.data(), dynamicZlibBuffer.length()};
+        }
+
+        return {zlibBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out};
+    }
+
+};
+
+#endif
+#endif // PERMESSAGEDEFLATE_H

src/WebSocket.h

@@ -31,8 +31,8 @@ struct WebSocket : AsyncSocket<SSL> {
 private:
     typedef AsyncSocket<SSL> Super;
 
-    void *init() {
-        new (us_socket_ext((us_socket *) this)) WebSocketData;
+    void *init(bool perMessageDeflate) {
+        new (us_socket_ext((us_socket *) this)) WebSocketData(perMessageDeflate);
         return this;
     }
 public:

src/WebSocketContext.h

@@ -46,8 +46,16 @@ private:
         return (WebSocketContextData<SSL> *) us_socket_context_ext((SOCKET_CONTEXT_TYPE *) this);
     }
 
-    static bool setCompressed(uWS::WebSocketState<isServer> *wState) {
-        return false; // do not support it
+    /* If we have negotiated compression, set this frame compressed */
+    static bool setCompressed(uWS::WebSocketState<isServer> *wState, void *s) {
+        WebSocketData *webSocketData = (WebSocketData *) us_socket_ext((us_socket *) s);
+
+        if (webSocketData->compressionStatus == WebSocketData::CompressionStatus::ENABLED) {
+            webSocketData->compressionStatus = WebSocketData::CompressionStatus::COMPRESSED_FRAME;
+            return true;
+        } else {
+            return false;
+        }
     }
 
     static void forceClose(uWS::WebSocketState<isServer> *wState, void *s) {
@@ -65,6 +73,22 @@ private:
             /* Did we get everything in one go? */
             if (!remainingBytes && fin && !webSocketData->fragmentBuffer.length()) {
 
+                /* Handle compressed frame */
+                if (webSocketData->compressionStatus == WebSocketData::CompressionStatus::COMPRESSED_FRAME) {
+                        webSocketData->compressionStatus = WebSocketData::CompressionStatus::ENABLED;
+
+                        LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) s)));
+
+                        std::string_view inflatedFrame = loopData->inflationStream->inflate({data, length});
+                        if (!inflatedFrame.length()) {
+                            forceClose(webSocketState, s);
+                            return true;
+                        } else {
+                            data = (char *) inflatedFrame.data();
+                            length = inflatedFrame.length();
+                        }
+                }
+
                 /* Check text messages for Utf-8 validity */
                 if (opCode == 1 && !WebSocketProtocol<isServer, WebSocketContext<SSL, isServer>>::isValidUtf8((unsigned char *) data, length)) {
                     forceClose(webSocketState, s);
@@ -87,9 +111,30 @@ private:
                 // what if we don't have any remaining bytes yet we are not fin? forceclose!
                 if (!remainingBytes && fin) {
 
-                    // reset length and data ptrs
-                    length = webSocketData->fragmentBuffer.length();
-                    data = webSocketData->fragmentBuffer.data();
+                    /* Handle compression */
+                    if (webSocketData->compressionStatus == WebSocketData::CompressionStatus::COMPRESSED_FRAME) {
+                            webSocketData->compressionStatus = WebSocketData::CompressionStatus::ENABLED;
+
+                            // what's really the story here?
+                            webSocketData->fragmentBuffer.append("....");
+
+                            LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) s)));
+
+                            std::string_view inflatedFrame = loopData->inflationStream->inflate({webSocketData->fragmentBuffer.data(), webSocketData->fragmentBuffer.length() - 4});
+                            if (!inflatedFrame.length()) {
+                                forceClose(webSocketState, s);
+                                return true;
+                            } else {
+                                data = (char *) inflatedFrame.data();
+                                length = inflatedFrame.length();
+                            }
+
+
+                    } else {
+                        // reset length and data ptrs
+                        length = webSocketData->fragmentBuffer.length();
+                        data = webSocketData->fragmentBuffer.data();
+                    }
 
                     /* Check text messages for Utf-8 validity */
                     if (opCode == 1 && !WebSocketProtocol<isServer, WebSocketContext<SSL, isServer>>::isValidUtf8((unsigned char *) data, length)) {

src/WebSocketData.h

@@ -32,9 +32,15 @@ private:
     std::string fragmentBuffer;
     int controlTipLength = 0;
     bool isShuttingDown = 0;
+    enum CompressionStatus : char {
+        DISABLED,
+        ENABLED,
+        COMPRESSED_FRAME
+    } compressionStatus;
 public:
-    WebSocketData() : WebSocketState<true>() {
-        //std::cout << "init websocket data!" << std::endl;
+    WebSocketData(bool perMessageDeflate) : WebSocketState<true>() {
+        std::cout << "perMessageDeflate: " << perMessageDeflate << std::endl;
+        compressionStatus = perMessageDeflate ? ENABLED : DISABLED;
     }
 };
 

src/WebSocketExtensions.h

@@ -0,0 +1,176 @@
+/*
+ * Copyright 2018 Alex Hultman and contributors.
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ *     http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef WEBSOCKETEXTENSIONS_H
+#define WEBSOCKETEXTENSIONS_H
+
+#include <string_view>
+
+namespace uWS {
+
+enum Options : unsigned int {
+    NO_OPTIONS = 0,
+    PERMESSAGE_DEFLATE = 1,
+    SERVER_NO_CONTEXT_TAKEOVER = 2, // remove this
+    CLIENT_NO_CONTEXT_TAKEOVER = 4, // remove this
+    NO_DELAY = 8,
+    SLIDING_DEFLATE_WINDOW = 16
+};
+
+template <bool isServer>
+class ExtensionsNegotiator {
+protected:
+    int options;
+public:
+    ExtensionsNegotiator(int wantedOptions);
+    std::string generateOffer();
+    void readOffer(std::string_view offer);
+    int getNegotiatedOptions();
+};
+
+enum ExtensionTokens {
+    TOK_PERMESSAGE_DEFLATE = 1838,
+    TOK_SERVER_NO_CONTEXT_TAKEOVER = 2807,
+    TOK_CLIENT_NO_CONTEXT_TAKEOVER = 2783,
+    TOK_SERVER_MAX_WINDOW_BITS = 2372,
+    TOK_CLIENT_MAX_WINDOW_BITS = 2348
+};
+
+class ExtensionsParser {
+private:
+    int *lastInteger = nullptr;
+
+public:
+    bool perMessageDeflate = false;
+    bool serverNoContextTakeover = false;
+    bool clientNoContextTakeover = false;
+    int serverMaxWindowBits = 0;
+    int clientMaxWindowBits = 0;
+
+    int getToken(const char *&in, const char *stop);
+    ExtensionsParser(const char *data, size_t length);
+};
+
+int ExtensionsParser::getToken(const char *&in, const char *stop) {
+    while (!isalnum(*in) && in != stop) {
+        in++;
+    }
+
+    int hashedToken = 0;
+    while (isalnum(*in) || *in == '-' || *in == '_') {
+        if (isdigit(*in)) {
+            hashedToken = hashedToken * 10 - (*in - '0');
+        } else {
+            hashedToken += *in;
+        }
+        in++;
+    }
+    return hashedToken;
+}
+
+ExtensionsParser::ExtensionsParser(const char *data, size_t length) {
+    const char *stop = data + length;
+    int token = 1;
+    for (; token && token != TOK_PERMESSAGE_DEFLATE; token = getToken(data, stop));
+
+    perMessageDeflate = (token == TOK_PERMESSAGE_DEFLATE);
+    while ((token = getToken(data, stop))) {
+        switch (token) {
+        case TOK_PERMESSAGE_DEFLATE:
+            return;
+        case TOK_SERVER_NO_CONTEXT_TAKEOVER:
+            serverNoContextTakeover = true;
+            break;
+        case TOK_CLIENT_NO_CONTEXT_TAKEOVER:
+            clientNoContextTakeover = true;
+            break;
+        case TOK_SERVER_MAX_WINDOW_BITS:
+            serverMaxWindowBits = 1;
+            lastInteger = &serverMaxWindowBits;
+            break;
+        case TOK_CLIENT_MAX_WINDOW_BITS:
+            clientMaxWindowBits = 1;
+            lastInteger = &clientMaxWindowBits;
+            break;
+        default:
+            if (token < 0 && lastInteger) {
+                *lastInteger = -token;
+            }
+            break;
+        }
+    }
+}
+
+template <bool isServer>
+ExtensionsNegotiator<isServer>::ExtensionsNegotiator(int wantedOptions) {
+    options = wantedOptions;
+}
+
+template <bool isServer>
+std::string ExtensionsNegotiator<isServer>::generateOffer() {
+    std::string extensionsOffer;
+    if (options & Options::PERMESSAGE_DEFLATE) {
+        extensionsOffer += "permessage-deflate";
+
+        if (options & Options::CLIENT_NO_CONTEXT_TAKEOVER) {
+            extensionsOffer += "; client_no_context_takeover";
+        }
+
+        // we do not support accepting this yet
+        // todo: if we agree on this, do not allocate a compressor
+        // per socket!
+
+        // It is RECOMMENDED that a server supports the
+        // "server_no_context_takeover" extension parameter in an extension
+        // negotiation offer.
+        if (options & Options::SERVER_NO_CONTEXT_TAKEOVER) {
+            //extensionsOffer += "; server_no_context_takeover";
+        }
+    }
+
+    return extensionsOffer;
+}
+
+template <bool isServer>
+void ExtensionsNegotiator<isServer>::readOffer(std::string_view offer) {
+    if (isServer) {
+        ExtensionsParser extensionsParser(offer.data(), offer.length());
+        if ((options & PERMESSAGE_DEFLATE) && extensionsParser.perMessageDeflate) {
+            if (extensionsParser.clientNoContextTakeover || (options & CLIENT_NO_CONTEXT_TAKEOVER)) {
+                options |= CLIENT_NO_CONTEXT_TAKEOVER;
+            }
+
+            if (extensionsParser.serverNoContextTakeover) {
+                options |= SERVER_NO_CONTEXT_TAKEOVER;
+            } else {
+                options &= ~SERVER_NO_CONTEXT_TAKEOVER;
+            }
+        } else {
+            options &= ~PERMESSAGE_DEFLATE;
+        }
+    } else {
+        // todo!
+    }
+}
+
+template <bool isServer>
+int ExtensionsNegotiator<isServer>::getNegotiatedOptions() {
+    return options;
+}
+
+}
+
+#endif // WEBSOCKETEXTENSIONS_H

src/WebSocketProtocol.h

@@ -370,7 +370,7 @@ public:
             while (length >= SHORT_MESSAGE_HEADER) {
 
                 // invalid reserved bits / invalid opcodes / invalid control frames / set compressed frame
-                if ((rsv1(src) && !Impl::setCompressed(wState)) || rsv23(src) || (getOpCode(src) > 2 && getOpCode(src) < 8) ||
+                if ((rsv1(src) && !Impl::setCompressed(wState, user)) || rsv23(src) || (getOpCode(src) > 2 && getOpCode(src) < 8) ||
                     getOpCode(src) > 10 || (getOpCode(src) > 2 && (!isFin(src) || payloadLength(src) > 125))) {
                     Impl::forceClose(wState, user);
                     return;