diff --git a/misc/main.cpp b/misc/main.cpp index 2b9a85c..2d4482b 100644 --- a/misc/main.cpp +++ b/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; diff --git a/src/App.h b/src/App.h index a6fcf8d..21b74f4 100644 --- a/src/App.h +++ b/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 *webSocket = (WebSocket *) StaticDispatch::static_dispatch(us_ssl_socket_context_adopt_socket, us_socket_context_adopt_socket)( - (typename StaticDispatch::SOCKET_CONTEXT_TYPE *) webSocketContext, (typename StaticDispatch::SOCKET_TYPE *) res, /*sizeof(WebSocketData)*/ 150); + (typename StaticDispatch::SOCKET_CONTEXT_TYPE *) webSocketContext, (typename StaticDispatch::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(); diff --git a/src/LoopData.h b/src/LoopData.h index 36ab301..d1988cb 100644 --- a/src/LoopData.h +++ b/src/LoopData.h @@ -48,6 +48,7 @@ public: /* Compression data */ InflationStream *inflationStream = nullptr; + DeflationStream *deflationStream = nullptr; }; } diff --git a/src/PerMessageDeflate.h b/src/PerMessageDeflate.h index ccb929f..1f70737 100644 --- a/src/PerMessageDeflate.h +++ b/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; diff --git a/src/WebSocket.h b/src/WebSocket.h index ebcb9b5..b7709de 100644 --- a/src/WebSocket.h +++ b/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>::messageFrameSize(message.length()); auto[sendBuffer, requiresWrite] = Super::getSendBuffer(messageFrameSize); - WebSocketProtocol>::formatMessage(sendBuffer, message.data(), message.length(), opCode, message.length(), false); + WebSocketProtocol>::formatMessage(sendBuffer, message.data(), message.length(), opCode, message.length(), compress); if (requiresWrite) { auto[written, failed] = Super::write(sendBuffer, messageFrameSize); diff --git a/src/WebSocketData.h b/src/WebSocketData.h index 4fb8c20..b64feb7 100644 --- a/src/WebSocketData.h +++ b/src/WebSocketData.h @@ -39,7 +39,7 @@ private: } compressionStatus; public: WebSocketData(bool perMessageDeflate) : WebSocketState() { - std::cout << "perMessageDeflate: " << perMessageDeflate << std::endl; + //std::cout << "perMessageDeflate: " << perMessageDeflate << std::endl; compressionStatus = perMessageDeflate ? ENABLED : DISABLED; } }; diff --git a/src/WebSocketProtocol.h b/src/WebSocketProtocol.h index 993f52c..94e6c92 100644 --- a/src/WebSocketProtocol.h +++ b/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