Use new PreparedMessage interface in Precompress

Author: uNetworkingAB

examples/Precompress.cpp

@@ -6,28 +6,22 @@ int main() {
         /* Fill with user data */
     };
 
-	/* Keeping track of last precompressed message both in original and compressed format */
-	std::string originalMessage;
-	std::string compressedMessage;
+	/* Keeping track of last prepared message */
+    uWS::PreparedMessage preparedMessage;
 	std::mutex m;
 
 	/* For demo, we create a thread that will update the precompressed message every second */
-	std::thread t2([&originalMessage, &compressedMessage, &m]() {
-		uWS::ZlibContext zlibContext;
-		uWS::DeflationStream compressor(uWS::DEDICATED_COMPRESSOR);
-		int counter = 0;
-
-		while (true) {
-			counter++;
-
+	std::thread t2([&m, &preparedMessage]() {
+		for (int counter = 1; true; counter++) {
 			m.lock();
-			originalMessage = "Hello you are looking at message number " + std::to_string(counter) + " and this text should be precompressed";		
-			compressedMessage = compressor.deflate(&zlibContext, {originalMessage.data(), originalMessage.length()}, true);
-			m.unlock();
-
+			std::string newMessage = "Hello you are looking at message number " + std::to_string(counter) + " and this text should be precompressed";		
+            
+            /* Here the current preparedMessage is updated */
+            preparedMessage = uWS::Loop::get()->prepareMessage(newMessage, uWS::OpCode::TEXT, true);
+            
+            m.unlock();
 			std::this_thread::sleep_for(std::chrono::milliseconds(500));
 		}
-
 	});
 
     uWS::App().ws<PerSocketData>("/*", {
@@ -39,19 +33,14 @@ int main() {
             /* Open event here, you may access ws->getUserData() which points to a PerSocketData struct */
 
         },
-        .message = [&originalMessage, &compressedMessage, &m](auto *ws, std::string_view message, uWS::OpCode opCode) {
+        .message = [&m, &preparedMessage](auto *ws, std::string_view message, uWS::OpCode opCode) {
 
 			/* First respond by echoing what they send us, without compression */
 			ws->send(message, opCode, false);
 
-			/* This should be wrapped up into ws->sendPrepared(PreparedMessage) in the future, experimental for now */
+            /* Send last prepared message */
 			m.lock();
-			if (ws->hasNegotiatedCompression() && compressedMessage.length() < originalMessage.length()) {
-				std::cout << "Responding with precompressed message saving " << (originalMessage.length() - compressedMessage.length()) << " bytes" << std::endl;
-				ws->send({compressedMessage.data(), compressedMessage.length()}, uWS::OpCode::TEXT, uWS::CompressFlags::ALREADY_COMPRESSED);
-			} else {
-				ws->send({originalMessage.data(), originalMessage.length()}, uWS::OpCode::TEXT);
-			}
+            ws->sendPrepared(preparedMessage);
 			m.unlock();
         },
         .dropped = [](auto */*ws*/, std::string_view /*message*/, uWS::OpCode /*opCode*/) {

src/Loop.h

@@ -25,6 +25,15 @@
 #include <iostream>
 
 namespace uWS {
+
+/* A prepared message is dependent on the Loop, so it belongs here */
+struct PreparedMessage {
+    /* These should be a single alloation along with the PreparedMessage itself (they are static) */
+    std::string originalMessage, compressedMessage;
+    bool compressed;
+    int opCode;
+};
+
 struct Loop {
 private:
     static void wakeupCb(us_loop_t *loop) {
@@ -106,6 +115,29 @@ struct Loop {
     }
 
 public:
+
+    /* Preformatted messages need the Loop */
+    PreparedMessage prepareMessage(std::string_view message, int opCode, bool compress = true) {
+        /* The message could be formatted right here, but this optimization is not done yet */
+        PreparedMessage preparedMessage;
+        preparedMessage.compressed = compress;
+        preparedMessage.opCode = opCode;
+        preparedMessage.originalMessage = message;
+
+        LoopData *loopData = (LoopData *) us_loop_ext((us_loop_t *) this);
+
+        /* Initialize loop's deflate inflate streams */
+        if (!loopData->zlibContext) {
+            loopData->zlibContext = new ZlibContext;
+            loopData->inflationStream = new InflationStream(CompressOptions::DEDICATED_DECOMPRESSOR);
+            loopData->deflationStream = new DeflationStream(CompressOptions::DEDICATED_COMPRESSOR);
+        }
+
+        preparedMessage.compressedMessage = loopData->deflationStream->deflate(loopData->zlibContext, {preparedMessage.originalMessage.data(), preparedMessage.originalMessage.length()}, true);
+        
+        return preparedMessage;
+    }
+
     /* Lazily initializes a per-thread loop and returns it.
      * Will automatically free all initialized loops at exit. */
     static Loop *get(void *existingNativeLoop = nullptr) {

src/WebSocket.h

@@ -100,6 +100,15 @@ struct WebSocket : AsyncSocket<SSL> {
         return webSocketData->compressionStatus == WebSocketData::ENABLED;
     }
 
+    /* Experimental */
+    void sendPrepared(PreparedMessage &preparedMessage) {
+        if (hasNegotiatedCompression() && preparedMessage.compressedMessage.length() < preparedMessage.originalMessage.length()) {
+            send({preparedMessage.compressedMessage.data(), preparedMessage.compressedMessage.length()}, (OpCode) preparedMessage.opCode, uWS::CompressFlags::ALREADY_COMPRESSED);
+        } else {
+            send({preparedMessage.originalMessage.data(), preparedMessage.originalMessage.length()}, (OpCode) preparedMessage.opCode);
+        }
+    }
+
     /* Send or buffer a WebSocket frame, compressed or not. Returns BACKPRESSURE on increased user space backpressure,
      * DROPPED on dropped message (due to backpressure) or SUCCCESS if you are free to send even more now. */
     SendStatus send(std::string_view message, OpCode opCode = OpCode::BINARY, int compress = false, bool fin = true) {