diff --git a/examples/EchoServer.cpp b/examples/EchoServer.cpp index ff9c625..7df9b7a 100644 --- a/examples/EchoServer.cpp +++ b/examples/EchoServer.cpp @@ -19,7 +19,7 @@ int main() { .passphrase = "1234" }).ws("/*", { /* Settings */ - .compression = uWS::DEDICATED_COMPRESSOR_4KB, + .compression = uWS::CompressOptions(uWS::DEDICATED_COMPRESSOR_4KB | uWS::DEDICATED_DECOMPRESSOR), .maxPayloadLength = 100 * 1024 * 1024, .idleTimeout = 16, .maxBackpressure = 100 * 1024 * 1024, @@ -30,6 +30,7 @@ int main() { .upgrade = nullptr, .open = [](auto */*ws*/) { /* Open event here, you may access ws->getUserData() which points to a PerSocketData struct */ + }, .message = [](auto *ws, std::string_view message, uWS::OpCode opCode) { ws->send(message, opCode, true); diff --git a/src/App.h b/src/App.h index 1cb5b5d..284ece5 100644 --- a/src/App.h +++ b/src/App.h @@ -123,7 +123,7 @@ public: unsigned int numSubscribers(std::string_view topic) { Topic *t = topicTree->lookupTopic(topic); if (t) { - return t->size(); + return (unsigned int) t->size(); } return 0; diff --git a/src/HttpResponse.h b/src/HttpResponse.h index 148db07..fd3e74c 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -236,8 +236,11 @@ public: CompressOptions compressOptions = CompressOptions::DISABLED; if (secWebSocketExtensions.length() && webSocketContextData->compression != DISABLED) { - /* We always want shared inflation */ + /* We always want shared inflation, (or the full 15) */ int wantedInflationWindow = 0; + if (webSocketContextData->compression & DEDICATED_DECOMPRESSOR) { + wantedInflationWindow = 15; + } /* Map from selected compressor */ int wantedCompressionWindow = (webSocketContextData->compression & 0xFF00) >> 8; diff --git a/src/PerMessageDeflate.h b/src/PerMessageDeflate.h index 2cc5367..8f338e4 100644 --- a/src/PerMessageDeflate.h +++ b/src/PerMessageDeflate.h @@ -24,10 +24,14 @@ /* We always define these options no matter if ZLIB is enabled or not */ namespace uWS { - /* Compressor mode is HIGH8(windowBits), LOW8(memLevel) */ + /* Compressor mode is 16 low bit where HIGH8(windowBits), LOW8(memLevel) */ enum CompressOptions : uint32_t { DISABLED = 0, - SHARED_COMPRESSOR = 1, + /* Highest bit is shared compressor */ + SHARED_COMPRESSOR = (uint32_t)1 << (uint32_t)31, + /* Second highest bit is DEDICATED_DECOMPRESSOR */ + DEDICATED_DECOMPRESSOR = (uint32_t)1 << (uint32_t)30, + /* Lowest 16 bit describe compressor */ DEDICATED_COMPRESSOR_3KB = 9 << 8 | 1, DEDICATED_COMPRESSOR_4KB = 9 << 8 | 2, DEDICATED_COMPRESSOR_8KB = 10 << 8 | 3, @@ -59,15 +63,15 @@ namespace uWS { #if defined(UWS_NO_ZLIB) || defined(UWS_MOCK_ZLIB) struct ZlibContext {}; struct InflationStream { - std::optional inflate(ZlibContext *zlibContext, std::string_view compressed, size_t maxPayloadLength) { + std::optional inflate(ZlibContext * /*zlibContext*/, std::string_view compressed, size_t maxPayloadLength, bool /*reset*/) { return compressed.substr(0, std::min(maxPayloadLength, compressed.length())); } }; struct DeflationStream { - std::string_view deflate(ZlibContext *zlibContext, std::string_view raw, bool reset) { + std::string_view deflate(ZlibContext * /*zlibContext*/, std::string_view raw, bool /*reset*/) { return raw; } - DeflationStream(int compressOptions) { + DeflationStream(CompressOptions /*compressOptions*/) { } }; #else @@ -201,7 +205,7 @@ struct InflationStream { } /* Zero length inflates are possible and valid */ - std::optional inflate(ZlibContext *zlibContext, std::string_view compressed, size_t maxPayloadLength) { + std::optional inflate(ZlibContext *zlibContext, std::string_view compressed, size_t maxPayloadLength, bool reset) { #ifdef UWS_USE_LIBDEFLATE /* Try fast path first */ @@ -242,7 +246,9 @@ struct InflationStream { } while (inflationStream.avail_out == 0 && zlibContext->dynamicInflationBuffer.length() <= maxPayloadLength); - inflateReset(&inflationStream); + if (reset) { + inflateReset(&inflationStream); + } if ((err != Z_BUF_ERROR && err != Z_OK) || zlibContext->dynamicInflationBuffer.length() > maxPayloadLength) { return std::nullopt; diff --git a/src/WebSocketContext.h b/src/WebSocketContext.h index 6fd7ba2..cd4acbf 100644 --- a/src/WebSocketContext.h +++ b/src/WebSocketContext.h @@ -72,7 +72,14 @@ private: webSocketData->compressionStatus = WebSocketData::CompressionStatus::ENABLED; LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(SSL, us_socket_context(SSL, (us_socket_t *) s))); - auto inflatedFrame = loopData->inflationStream->inflate(loopData->zlibContext, {data, length}, webSocketContextData->maxPayloadLength); + /* Decompress using shared or dedicated decompressor */ + std::optional inflatedFrame; + if (webSocketData->inflationStream) { + inflatedFrame = webSocketData->inflationStream->inflate(loopData->zlibContext, {data, length}, webSocketContextData->maxPayloadLength, false); + } else { + inflatedFrame = loopData->inflationStream->inflate(loopData->zlibContext, {data, length}, webSocketContextData->maxPayloadLength, true); + } + if (!inflatedFrame.has_value()) { forceClose(webSocketState, s, ERR_TOO_BIG_MESSAGE_INFLATION); return true; @@ -124,7 +131,14 @@ private: ) ); - auto inflatedFrame = loopData->inflationStream->inflate(loopData->zlibContext, {webSocketData->fragmentBuffer.data(), webSocketData->fragmentBuffer.length() - 9}, webSocketContextData->maxPayloadLength); + /* Decompress using shared or dedicated decompressor */ + std::optional inflatedFrame; + if (webSocketData->inflationStream) { + inflatedFrame = webSocketData->inflationStream->inflate(loopData->zlibContext, {webSocketData->fragmentBuffer.data(), webSocketData->fragmentBuffer.length() - 9}, webSocketContextData->maxPayloadLength, false); + } else { + inflatedFrame = loopData->inflationStream->inflate(loopData->zlibContext, {webSocketData->fragmentBuffer.data(), webSocketData->fragmentBuffer.length() - 9}, webSocketContextData->maxPayloadLength, true); + } + if (!inflatedFrame.has_value()) { forceClose(webSocketState, s, ERR_TOO_BIG_MESSAGE_INFLATION); return true; diff --git a/src/WebSocketData.h b/src/WebSocketData.h index 6a4c52e..1506a51 100644 --- a/src/WebSocketData.h +++ b/src/WebSocketData.h @@ -46,6 +46,8 @@ private: /* We might have a dedicated compressor */ DeflationStream *deflationStream = nullptr; + /* And / or a dedicated decompressor */ + InflationStream *inflationStream = nullptr; /* We could be a subscriber */ Subscriber *subscriber = nullptr; @@ -53,10 +55,14 @@ public: WebSocketData(bool perMessageDeflate, CompressOptions compressOptions, BackPressure &&backpressure) : AsyncSocketData(std::move(backpressure)), WebSocketState() { compressionStatus = perMessageDeflate ? ENABLED : DISABLED; - /* Initialize the dedicated sliding window */ - if (perMessageDeflate && (compressOptions != CompressOptions::SHARED_COMPRESSOR)) { + /* Initialize the dedicated sliding window(s) */ + if (perMessageDeflate && (0 == (compressOptions & CompressOptions::SHARED_COMPRESSOR))) { deflationStream = new DeflationStream(compressOptions); } + + if (perMessageDeflate && (compressOptions & CompressOptions::DEDICATED_DECOMPRESSOR)) { + inflationStream = new InflationStream(); + } } ~WebSocketData() { @@ -64,6 +70,10 @@ public: delete deflationStream; } + if (inflationStream) { + delete inflationStream; + } + if (subscriber) { delete subscriber; }