diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index 120e0eb..6081d39 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -33,6 +33,15 @@ template struct WebSocket; template struct WebSocketContextData { +private: + /* Used for prepending unframed messages when using dedicated compressors */ + struct MessageMetadata { + unsigned int length; + OpCode opCode; + bool compress; + }; + +public: /* The callbacks for this context */ fu2::unique_function *)> openHandler = nullptr; fu2::unique_function *, std::string_view, uWS::OpCode)> messageHandler = nullptr; @@ -73,7 +82,39 @@ struct WebSocketContextData { /* Are we using compression? Fine, pick the compressed data track */ WebSocketData *webSocketData = (WebSocketData *) asyncSocket->getAsyncSocketData(); if (webSocketData->compressionStatus != WebSocketData::CompressionStatus::DISABLED) { + + /* This is used for both shared and dedicated paths */ selectedData = data.second; + + /* However, dedicated compression has its own path */ + if (compression != SHARED_COMPRESSOR) { + + WebSocket *ws = (WebSocket *) asyncSocket; + + /* We need to handle being corked, and corking here */ + + + while (selectedData.length()) { + /* Interpret the data like so */ + MessageMetadata mm; + memcpy((char *) &mm, selectedData.data(), sizeof(MessageMetadata)); + std::string_view unframedMessage(selectedData.data() + sizeof(MessageMetadata), mm.length); + + //std::cout << "<" << unframedMessage << ">" << std::endl; + + /* Here we perform the actual compression and framing */ + ws->send(unframedMessage, mm.opCode, mm.compress); + + /* Advance until empty */ + selectedData.remove_prefix(sizeof(MessageMetadata) + mm.length); + } + + /* Here we need to uncork or keep it as was */ + + + /* See below */ + return 0; + } } /* Note: this assumes we are not corked, as corking will swallow things and fail later on */ @@ -110,25 +151,54 @@ struct WebSocketContextData { char *dst = (char *) malloc(protocol::messageFrameSize(message.size())); size_t dst_length = protocol::formatMessage(dst, message.data(), message.length(), opCode, message.length(), false); - if (compress) { - /* Loop data holds shared compressor */ - LoopData *loopData = (LoopData *) us_loop_ext((us_loop_t *) Loop::get()); - - /* Compress it */ - std::string_view compressedMessage = loopData->deflationStream->deflate(loopData->zlibContext, message, true); - - /* Frame it */ - char *dst_compressed = (char *) malloc(protocol::messageFrameSize(compressedMessage.size())); - size_t dst_compressed_length = protocol::formatMessage(dst_compressed, compressedMessage.data(), compressedMessage.length(), opCode, compressedMessage.length(), true); - - /* Always publish the shortest one in any case */ - topicTree.publish(topic, {std::string_view(dst, dst_length), dst_compressed_length >= dst_length ? std::string_view(dst, dst_length) : std::string_view(dst_compressed, dst_compressed_length)}); - - /* We don't care for allocation here */ - ::free(dst_compressed); + /* If compression is disabled */ + if (compression == DISABLED) { + /* Leave second field empty as nobody will ever read it */ + topicTree.publish(topic, {std::string_view(dst, dst_length), {}}); } else { - /* If not compressing, put same message on both tracks */ - topicTree.publish(topic, {std::string_view(dst, dst_length), std::string_view(dst, dst_length)}); + if (compress) { + /* Shared compression mode publishes compressed, framed data */ + if (compression == SHARED_COMPRESSOR) { + /* Loop data holds shared compressor */ + LoopData *loopData = (LoopData *) us_loop_ext((us_loop_t *) Loop::get()); + + /* Compress it */ + std::string_view compressedMessage = loopData->deflationStream->deflate(loopData->zlibContext, message, true); + + /* Frame it */ + char *dst_compressed = (char *) malloc(protocol::messageFrameSize(compressedMessage.size())); + size_t dst_compressed_length = protocol::formatMessage(dst_compressed, compressedMessage.data(), compressedMessage.length(), opCode, compressedMessage.length(), true); + + /* Always publish the shortest one in any case */ + topicTree.publish(topic, {std::string_view(dst, dst_length), dst_compressed_length >= dst_length ? std::string_view(dst, dst_length) : std::string_view(dst_compressed, dst_compressed_length)}); + + /* We don't care for allocation here */ + ::free(dst_compressed); + } else { + /* Dedicated compression mode publishes metadata + unframed uncompressed data */ + char *dst_compressed = (char *) malloc(message.length() + sizeof(MessageMetadata)); + + MessageMetadata mm = { + (unsigned int) message.length(), + opCode, + compress + }; + + memcpy(dst_compressed, (char *) &mm, sizeof(MessageMetadata)); + memcpy(dst_compressed + sizeof(MessageMetadata), message.data(), message.length()); + + /* Interpretation of compressed data depends on what compressor we use */ + topicTree.publish(topic, { + std::string_view(dst, dst_length), + std::string_view(dst_compressed, message.length() + sizeof(MessageMetadata)) + }); + + ::free(dst_compressed); + } + } else { + /* If not compressing, put same message on both tracks */ + topicTree.publish(topic, {std::string_view(dst, dst_length), std::string_view(dst, dst_length)}); + } } ::free(dst);