Add initial send compression

misc/main.cpp

@@ -28,11 +28,11 @@ int main(int argc, char **argv) {
         .open = [](auto *ws, auto *req) {
             std::cout << "WebSocket connected" << std::endl;
             /* Access per socket data */
-            /*PerSocketData *perSocketData = */ws->getUserData();
+            /*PerSocketData *perSocketData = *///ws->getUserData();
             /*perSocketData->hello = 13;*/
         },
         .message = [](auto *ws, std::string_view message, uWS::OpCode opCode) {
-            ws->send(message, opCode);
+            ws->send(message, opCode, true);
         },
         .drain = [](auto *ws) {
             std::cout << "Drainage: " << ws->getBufferedAmount() << std::endl;

src/App.h

@@ -76,6 +76,10 @@ public:
             if (!loopData->inflationStream) {
                 loopData->inflationStream = new InflationStream;
             }
+
+            if (!loopData->deflationStream) {
+                loopData->deflationStream = new DeflationStream;
+            }
         }
 
         /* Copy all handlers */
@@ -122,16 +126,23 @@ public:
 
                 /* 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);
+                            (typename StaticDispatch<SSL>::SOCKET_CONTEXT_TYPE *) webSocketContext, (typename StaticDispatch<SSL>::SOCKET_TYPE *) res, sizeof(WebSocketData));
+
+                /* Update corked socket in case we got a new one (assuming we always are corked in handlers). */
+                webSocket->cork();
 
                 httpContext->upgradeToWebSocket(
                             webSocket->init(perMessageDeflate)
                             );
 
+                /* Emit open event */
                 if (behavior.open) {
                     behavior.open(webSocket, req);
                 }
 
+                // todo: perform all the checks such as shutdown, closed, etc!
+                // bug? or does this happen automatically? no!
+
             } else {
                 /* For now we do not support having HTTP and websocket routes on the same URL */
                 res->close();

src/LoopData.h

@@ -48,6 +48,7 @@ public:
 
     /* Compression data */
     InflationStream *inflationStream = nullptr;
+    DeflationStream *deflationStream = nullptr;
 };
 
 }

src/PerMessageDeflate.h

@@ -30,9 +30,82 @@
 
 // we also need DeflationStream
 
+struct DeflationStream {
+
+    // share this under the Loop
+    std::string dynamicZlibBuffer;
+    z_stream deflationStream = {};
+    char *zlibBuffer;
+
+    DeflationStream() {
+        std::cout << "Constructing DeflationStream" << std::endl;
+        zlibBuffer = (char *) malloc(LARGE_BUFFER_SIZE);
+
+        deflateInit2(&deflationStream, 1, Z_DEFLATED, -15, 8, Z_DEFAULT_STRATEGY);
+    }
+
+    std::string_view deflate(std::string_view raw) {
+
+        // slidingDeflateWindow är input, length är in/ut
+
+        z_stream *slidingDeflateWindow = nullptr;
+
+        dynamicZlibBuffer.clear();
+
+        z_stream *compressor = slidingDeflateWindow ? slidingDeflateWindow : &deflationStream;
+
+        compressor->next_in = (Bytef *) raw.data();
+        compressor->avail_in = (unsigned int) raw.length();
+
+        // note: zlib requires more than 6 bytes with Z_SYNC_FLUSH
+        const int DEFLATE_OUTPUT_CHUNK = LARGE_BUFFER_SIZE;
+
+        int err;
+        do {
+            compressor->next_out = (Bytef *) zlibBuffer;
+            compressor->avail_out = DEFLATE_OUTPUT_CHUNK;
+
+            err = ::deflate(compressor, Z_SYNC_FLUSH);
+            if (Z_OK == err && compressor->avail_out == 0) {
+                dynamicZlibBuffer.append(zlibBuffer, DEFLATE_OUTPUT_CHUNK - compressor->avail_out);
+                continue;
+            } else {
+                break;
+            }
+        } while (true);
+
+        // note: should not change avail_out
+        if (!slidingDeflateWindow) {
+            deflateReset(compressor);
+        }
+
+        if (dynamicZlibBuffer.length()) {
+            dynamicZlibBuffer.append(zlibBuffer, DEFLATE_OUTPUT_CHUNK - compressor->avail_out);
+
+            return {(char *) dynamicZlibBuffer.data(), dynamicZlibBuffer.length() - 4};
+
+            //length = dynamicZlibBuffer.length() - 4;
+            //return (char *) dynamicZlibBuffer.data();
+        }
+
+        return {
+            zlibBuffer,
+            DEFLATE_OUTPUT_CHUNK - compressor->avail_out - 4
+        };
+
+        //length = DEFLATE_OUTPUT_CHUNK - compressor->avail_out - 4;
+        //return zlibBuffer;
+    }
+
+    ~DeflationStream() {
+        std::cout << "Destructing DeflationStream" << std::endl;
+    }
+};
+
 // the loop holds one of these
 struct InflationStream {
 
+    // share this under the Loop
     std::string dynamicZlibBuffer;
     z_stream inflationStream = {};
     char *zlibBuffer;

src/WebSocket.h

@@ -54,14 +54,22 @@ public:
     bool send(std::string_view message, uWS::OpCode opCode = uWS::OpCode::BINARY, bool compress = false) {
         /* Transform the message to compressed domain if requested */
         if (compress) {
-            //message = ;
-            std::cout << "send compression ignored!" << std::endl;
+            WebSocketData *webSocketData = (WebSocketData *) Super::getExt();
+
+            /* Check and correct the compress hint */
+            if (opCode < 3 && webSocketData->compressionStatus == WebSocketData::ENABLED) {
+                // todo: shared deflate window
+                LoopData *loopData = Super::getLoopData();
+                message = loopData->deflationStream->deflate(message);
+            } else {
+                compress = false;
+            }
         }
 
         /* Get size, alloate size, write if needed */
         size_t messageFrameSize = WebSocketProtocol<isServer, WebSocket<SSL, isServer>>::messageFrameSize(message.length());
         auto[sendBuffer, requiresWrite] = Super::getSendBuffer(messageFrameSize);
-        WebSocketProtocol<isServer, WebSocket<SSL, isServer>>::formatMessage(sendBuffer, message.data(), message.length(), opCode, message.length(), false);
+        WebSocketProtocol<isServer, WebSocket<SSL, isServer>>::formatMessage(sendBuffer, message.data(), message.length(), opCode, message.length(), compress);
         if (requiresWrite) {
             auto[written, failed] = Super::write(sendBuffer, messageFrameSize);
 

src/WebSocketData.h

@@ -39,7 +39,7 @@ private:
     } compressionStatus;
 public:
     WebSocketData(bool perMessageDeflate) : WebSocketState<true>() {
-        std::cout << "perMessageDeflate: " << perMessageDeflate << std::endl;
+        //std::cout << "perMessageDeflate: " << perMessageDeflate << std::endl;
         compressionStatus = perMessageDeflate ? ENABLED : DISABLED;
     }
 };

src/WebSocketProtocol.h

@@ -17,6 +17,8 @@
 #ifndef WEBSOCKETPROTOCOL_UWS_H
 #define WEBSOCKETPROTOCOL_UWS_H
 
+// this module depends on windows ws2_lib being linked which is nonsense, fix!
+
 /* This segment is not cross-platform! Fix! */
 /* PortableEndianConversion.h */
 #ifdef __linux