talksik/uWebSockets
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add DEDICATED_DECOMPRESSOR CompressOption

Browse Source
This commit is contained in:
Alex Hultman
2021-09-28 04:06:49 +02:00
parent 7a3bc47f75
commit 415fec3830
6 changed files with 48 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/EchoServer.cpp
+2 -1
View File
@@ -19,7 +19,7 @@ int main() {
.passphrase = "1234"
}).ws<PerSocketData>("/*", {
/* 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);
src/App.h
+1 -1
View File
@@ -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;
src/HttpResponse.h
+4 -1
View File
@@ -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;
src/PerMessageDeflate.h
+13 -7
View File
@@ -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<std::string_view> inflate(ZlibContext *zlibContext, std::string_view compressed, size_t maxPayloadLength) {
std::optional<std::string_view> 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<std::string_view> inflate(ZlibContext *zlibContext, std::string_view compressed, size_t maxPayloadLength) {
std::optional<std::string_view> 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;
src/WebSocketContext.h
+16 -2
View File
@@ -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<std::string_view> 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<std::string_view> 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;
src/WebSocketData.h
+12 -2
View File
@@ -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<false>(std::move(backpressure)), WebSocketState<true>() {
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;
}