From 91620c6c42b51b42c2e31fe6cf9fd0ae20b86061 Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Wed, 21 Nov 2018 22:46:17 +0100 Subject: [PATCH] (Just barely) pass Autobahn tests chapter 1 - 10 --- src/App.h | 2 +- src/AsyncSocket.h | 2 +- src/WebSocket.h | 24 +++++ src/WebSocketContext.h | 193 ++++++++++++++++++++++++++++++++++------ src/WebSocketData.h | 8 +- src/WebSocketProtocol.h | 8 +- 6 files changed, 202 insertions(+), 35 deletions(-) diff --git a/src/App.h b/src/App.h index 614fbce..8f3263d 100644 --- a/src/App.h +++ b/src/App.h @@ -54,7 +54,7 @@ public: // this method creates a new websocket context and attaches it to a path TemplatedApp &ws(std::string pattern, std::function connectHandler, std::function *, std::string_view, uWS::OpCode)> messageHandler) { // init the websocket context here! - uWS::WebSocketContext *webSocketContext = uWS::WebSocketContext::create(uWS::Loop::defaultLoop(), (typename StaticDispatch::SOCKET_CONTEXT_TYPE *) httpContext); + uWS::WebSocketContext *webSocketContext = uWS::WebSocketContext::create(uWS::Loop::defaultLoop(), (typename StaticDispatch::SOCKET_CONTEXT_TYPE *) httpContext); webSocketContext->getExt()->messageHandler = messageHandler; diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index 64b05e3..02bc77f 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -28,7 +28,7 @@ namespace uWS { template struct AsyncSocket : StaticDispatch { template friend struct HttpContext; - template friend struct WebSocketContext; + template friend struct WebSocketContext; protected: using SOCKET_TYPE = typename StaticDispatch::SOCKET_TYPE; using StaticDispatch::static_dispatch; diff --git a/src/WebSocket.h b/src/WebSocket.h index 1dfd1e3..7fe3d49 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -31,6 +31,30 @@ public: } + /* Emit close event, stat passive timeout */ + void close(int code, std::string_view message = {}) { + // closing should trigger close event! + + std::cout << "Closing websocket: " << code << " = " << message << std::endl; + + static const int MAX_CLOSE_PAYLOAD = 123; + int length = std::min(MAX_CLOSE_PAYLOAD, message.length()); + + // here we start a timeout and handle it accordingly in the timeout handler + + WebSocketData *webSocketData = (WebSocketData *) us_socket_ext((us_socket *) this); + + webSocketData->isShuttingDown = true; + + /* Format and send the close frame */ + char closePayload[MAX_CLOSE_PAYLOAD + 2]; + int closePayloadLength = (int) WebSocketProtocol>::formatClosePayload(closePayload, code, message.data(), length); + send(std::string_view(closePayload, closePayloadLength), OpCode::CLOSE); + + // why should we fin here? + //us_socket_shutdown((us_socket *) this); + } + // absolutely not public! void init() { // construct us diff --git a/src/WebSocketContext.h b/src/WebSocketContext.h index 76ae041..de96123 100644 --- a/src/WebSocketContext.h +++ b/src/WebSocketContext.h @@ -13,9 +13,10 @@ namespace uWS { -template +template struct WebSocketContext : StaticDispatch { template friend struct TemplatedApp; + template friend struct WebSocketProtocol; private: using SOCKET_CONTEXT_TYPE = typename StaticDispatch::SOCKET_CONTEXT_TYPE; using SOCKET_TYPE = typename StaticDispatch::SOCKET_TYPE; @@ -30,39 +31,164 @@ private: return (WebSocketContextData *) us_socket_context_ext((SOCKET_CONTEXT_TYPE *) this); } - // I don't even.. merge this with the context itself! - template - struct WebSocketProtcolImplementation { - static bool setCompressed(uWS::WebSocketState *wState) { - std::cout << "set compressed" << std::endl; - return true; + // could still lie in its own struct! + static bool setCompressed(uWS::WebSocketState *wState) { + std::cout << "set compressed" << std::endl; + return false; // do not support it + } + + // todo: pass along user! + static void forceClose(uWS::WebSocketState *wState, void *s) { + std::cout << "force close" << std::endl; + + us_socket_close((us_socket *) s); + + } + + static bool handleFragment(char *data, size_t length, unsigned int remainingBytes, int opCode, bool fin, uWS::WebSocketState *webSocketState, void *s) { + + // this is maybe not the most elegant but who cares + WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(us_socket_get_context((us_socket *) s)); + + // we need to get the WebSocket data also! + WebSocketData *webSocketData = (WebSocketData *) us_socket_ext((us_socket *) s); + + if (opCode < 3) { + + if (!remainingBytes && fin && !webSocketData->fragmentBuffer.length()) { + + /* Check text messages for Utf-8 validity */ + if (opCode == 1 && !WebSocketProtocol>::isValidUtf8((unsigned char *) data, length)) { + forceClose(webSocketState, s); + return true; + } + + webSocketContextData->messageHandler((WebSocket *) s, std::string_view(data, length), (uWS::OpCode) opCode); + + // todo: check if shut down or shutting down (shut down from websocket perspective) + // if so, then return true + + } else { + /* Allocate fragment buffer up front first time */ + if (!webSocketData->fragmentBuffer.length()) { + std::cout << "Resizing buffers to " << (length + remainingBytes) << " bytes" << std::endl; + webSocketData->fragmentBuffer.reserve(length + remainingBytes); + } + + webSocketData->fragmentBuffer.append(data, length); + + std::cout << "buffering incomplete fragment: " << webSocketData->fragmentBuffer.length() << " added " << length << std::endl; + + /* Are we done now? */ + // what if we don't have any remaining bytes yet we are not fin? forceclose! + if (!remainingBytes && fin) { + + std::cout << "GOT FINAL FRAGMENT!" << std::endl; + + // reset length and data ptrs + length = webSocketData->fragmentBuffer.length(); + data = webSocketData->fragmentBuffer.data(); + + + /* Check text messages for Utf-8 validity */ + if (opCode == 1 && !WebSocketProtocol>::isValidUtf8((unsigned char *) data, length)) { + forceClose(webSocketState, s); + return true; + } + + webSocketContextData->messageHandler((WebSocket *) s, std::string_view(data, length), (uWS::OpCode) opCode); + + // todo: check if shut down or shutting down (shut down from websocket perspective) + // if so, then return true + + + webSocketData->fragmentBuffer.clear(); + } + + } + + + } else { + WebSocket *webSocket = (WebSocket *) s; + + if (!remainingBytes && fin && !webSocketData->controlTipLength) { + if (opCode == CLOSE) { + typename WebSocketProtocol>::CloseFrame closeFrame = WebSocketProtocol>::parseClosePayload(data, length); + webSocket->close(closeFrame.code, std::string_view(closeFrame.message, closeFrame.length)); + return true; + } else { + if (opCode == PING) { + webSocket->send(std::string_view(data, length), (OpCode) OpCode::PONG); + /*group->pingHandler(webSocket, data, length); + if (webSocket->isClosed() || webSocket->isShuttingDown()) { + return true; + }*/ + } else if (opCode == PONG) { + /*group->pongHandler(webSocket, data, length); + if (webSocket->isClosed() || webSocket->isShuttingDown()) { + return true; + }*/ + } + } + } else { + // todo, buffer control frames + + std::cout << "control frames! BUFFFERRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRR" << std::endl; + + + // unsure how big this is? + webSocketData->fragmentBuffer.append(data, length); + webSocketData->controlTipLength += length; + + + if (!remainingBytes && fin) { + char *controlBuffer = (char *) webSocketData->fragmentBuffer.data() + webSocketData->fragmentBuffer.length() - webSocketData->controlTipLength; + if (opCode == CLOSE) { + typename WebSocketProtocol>::CloseFrame closeFrame = WebSocketProtocol>::parseClosePayload(controlBuffer, webSocketData->controlTipLength); + webSocket->close(closeFrame.code, std::string_view(closeFrame.message, closeFrame.length)); + return true; + } else { + if (opCode == PING) { + webSocket->send(std::string_view(controlBuffer, webSocketData->controlTipLength), (OpCode) OpCode::PONG); + /*group->pingHandler(webSocket, controlBuffer, webSocket->controlTipLength); + if (webSocket->isClosed() || webSocket->isShuttingDown()) { + return true; + }*/ + } else if (opCode == PONG) { + /*group->pongHandler(webSocket, controlBuffer, webSocket->controlTipLength); + if (webSocket->isClosed() || webSocket->isShuttingDown()) { + return true; + }*/ + } + } + + // not optimal but slow path + webSocketData->fragmentBuffer.resize(webSocketData->fragmentBuffer.length() - webSocketData->controlTipLength); + webSocketData->controlTipLength = 0; + } + + + + + } + + } - static void forceClose(uWS::WebSocketState *wState) { - std::cout << "force close" << std::endl; - } - static bool handleFragment(char *data, size_t length, unsigned int remainingBytes, int opCode, bool fin, uWS::WebSocketState *webSocketState, void *s) { - // this is maybe not the most elegant but who cares - WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(us_socket_get_context((us_socket *) s)); + // the only thing here to check is probably closed - // emit message - webSocketContextData->messageHandler((WebSocket *) s, std::string_view(data, length), (uWS::OpCode) opCode); + // why does it not do anything immediately on true? + return false; + } - // the only thing here to check is probably closed + static bool refusePayloadLength(uint64_t length, uWS::WebSocketState *wState) { + //std::cout << "refusepayloadlength" << std::endl; + return false; + } - // why does it not do anything immediately on true? - return false; - } - - static bool refusePayloadLength(uint64_t length, uWS::WebSocketState *wState) { - //std::cout << "refusepayloadlength" << std::endl; - return false; - } - }; - - WebSocketContext *init() { + WebSocketContext *init() { /* I guess open is never called */ @@ -86,11 +212,16 @@ private: WebSocketData *wsState = (WebSocketData *) us_socket_ext(s); // this parser requires almost no time -> 215k req/sec of 215k possible - uWS::WebSocketProtocol>::consume(data, length, wsState, s); + uWS::WebSocketProtocol>::consume(data, length, wsState, s); webSocket->uncork(); + // are we shutdown? + if (wsState->isShuttingDown) { + webSocket->shutdown(); + } + return s; }); @@ -99,6 +230,12 @@ private: std::cout << "websocket writable" << std::endl; + // we need to drain here! + + AsyncSocket *webSocket = (AsyncSocket *) s; + + webSocket->write(nullptr, 0); // drainage - also check for shutdown! + return s; }); diff --git a/src/WebSocketData.h b/src/WebSocketData.h index 316e0e5..985edde 100644 --- a/src/WebSocketData.h +++ b/src/WebSocketData.h @@ -4,12 +4,18 @@ #include "WebSocketProtocol.h" #include "AsyncSocketData.h" +#include + namespace uWS { // take care with get_ext here ! struct WebSocketData : AsyncSocketData, WebSocketState { + template friend struct WebSocketContext; + template friend struct WebSocket; private: - + std::string fragmentBuffer; + int controlTipLength = 0; + bool isShuttingDown = 0; public: WebSocketData() : WebSocketState() { std::cout << "init websocket data!" << std::endl; diff --git a/src/WebSocketProtocol.h b/src/WebSocketProtocol.h index 2bd914d..3ce3e9b 100644 --- a/src/WebSocketProtocol.h +++ b/src/WebSocketProtocol.h @@ -141,18 +141,18 @@ protected: static inline bool consumeMessage(T payLength, char *&src, unsigned int &length, WebSocketState *wState, void *user) { if (getOpCode(src)) { if (wState->state.opStack == 1 || (!wState->state.lastFin && getOpCode(src) < 2)) { - Impl::forceClose(wState); + Impl::forceClose(wState, user); return true; } wState->state.opCode[++wState->state.opStack] = (OpCode) getOpCode(src); } else if (wState->state.opStack == -1) { - Impl::forceClose(wState); + Impl::forceClose(wState, user); return true; } wState->state.lastFin = isFin(src); if (Impl::refusePayloadLength(payLength, wState)) { - Impl::forceClose(wState); + Impl::forceClose(wState, user); return true; } @@ -370,7 +370,7 @@ public: // invalid reserved bits / invalid opcodes / invalid control frames / set compressed frame if ((rsv1(src) && !Impl::setCompressed(wState)) || rsv23(src) || (getOpCode(src) > 2 && getOpCode(src) < 8) || getOpCode(src) > 10 || (getOpCode(src) > 2 && (!isFin(src) || payloadLength(src) > 125))) { - Impl::forceClose(wState); + Impl::forceClose(wState, user); return; }