From cb48712b3d3faf4bcee199674f7ac208eb3f411d Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Tue, 2 Jun 2020 12:40:51 +0200 Subject: [PATCH] Make it stick together just barely --- examples/EchoServer.cpp | 61 +++++++++++++++++++++++++++++++++++--- src/App.h | 47 ++++++++--------------------- src/HttpContext.h | 3 +- src/HttpResponse.h | 58 ++++++++++++++++++++++++++++++++++-- src/WebSocket.h | 1 + src/WebSocketContextData.h | 8 +++-- src/WebSocketData.h | 1 + 7 files changed, 134 insertions(+), 45 deletions(-) diff --git a/examples/EchoServer.cpp b/examples/EchoServer.cpp index 5d5403f..d591efe 100644 --- a/examples/EchoServer.cpp +++ b/examples/EchoServer.cpp @@ -25,11 +25,64 @@ int main() { .idleTimeout = 10, .maxBackpressure = 1 * 1024 * 1204, /* Handlers */ - .upgrade = [](auto *res, auto *req) { - std::cout << "Upgrade now!" << std::endl; - /* Pass user data from upgrade to open handler here */ - //res.upgrade({.something = 15}, req); + //.syncUpgrade() + + .upgrade = [](auto *res, auto *req, auto *context) { + + /* Immediate path */ + res->template upgrade(req->getHeader("sec-websocket-key"), + req->getHeader("sec-websocket-protocol"), + req->getHeader("sec-websocket-extensions"), + context); + + return; + + std::cout << "Upgrade request!" << std::endl; + + /* Async path, you have to COPY headers */ + std::string secWebSocketKey(req->getHeader("sec-websocket-key")); + std::string secWebSocketProtocol(req->getHeader("sec-websocket-protocol")); + std::string secWebSocketExtensions(req->getHeader("sec-websocket-extensions")); + + /* If the client disconnects inbetween, you MUST avoid upgrading */ + bool *aborted = new bool(false); + res->onAborted([=]() { + std::cout << "WebSocket upgrade aborted!" << std::endl; + *aborted = true; + }); + + /* Simulate checking auth for 10 seconds */ + struct us_loop_t *loop = (struct us_loop_t *) uWS::Loop::get(); + struct us_timer_t *delayTimer = us_create_timer(loop, 0, 0); + us_timer_set(delayTimer, [](struct us_timer_t *t) { + std::cout << "Timer triggered!" << std::endl; + + us_timer_close(t); + }, 5000, 0); + + /* Simulate doing async work by deferring upgrade to next event loop iteration */ + uWS::Loop::get()->defer([=](){ + + std::cout << "Upgrading now!" << std::endl; + + if (!*aborted) { + + // this should get some kind of ticket + res->template upgrade(secWebSocketKey, + secWebSocketProtocol, + secWebSocketExtensions, + context); + } + + delete aborted; + }); + + /* Immediate path is in all seriousness a simple return bool */ + + // not exactly, you need to pass UserData - return a moved UserData and bool? + + }, .open = [](auto *ws, auto *req) { /* Open event here, you may access ws->getUserData() which points to a PerSocketData struct */ diff --git a/src/App.h b/src/App.h index b349eb3..72d48f7 100644 --- a/src/App.h +++ b/src/App.h @@ -86,7 +86,7 @@ public: int maxPayloadLength = 16 * 1024; int idleTimeout = 120; int maxBackpressure = 1 * 1024 * 1024; - fu2::unique_function *, HttpRequest *)> upgrade = nullptr; + fu2::unique_function *, HttpRequest *, struct us_socket_context_t *)> upgrade = nullptr; fu2::unique_function *, HttpRequest *)> open = nullptr; fu2::unique_function *, std::string_view, uWS::OpCode)> message = nullptr; fu2::unique_function *)> drain = nullptr; @@ -129,6 +129,7 @@ public: } /* Copy all handlers */ + webSocketContext->getExt()->openHandler = std::move(behavior.open); webSocketContext->getExt()->messageHandler = std::move(behavior.message); webSocketContext->getExt()->drainHandler = std::move(behavior.drain); webSocketContext->getExt()->closeHandler = std::move([closeHandler = std::move(behavior.close)](WebSocket *ws, int code, std::string_view message) mutable { @@ -146,6 +147,7 @@ public: webSocketContext->getExt()->maxPayloadLength = behavior.maxPayloadLength; webSocketContext->getExt()->idleTimeout = behavior.idleTimeout; webSocketContext->getExt()->maxBackpressure = behavior.maxBackpressure; + webSocketContext->getExt()->compression = behavior.compression; httpContext->onHttp("get", pattern, [webSocketContext, httpContext = this->httpContext, behavior = std::move(behavior)](auto *res, auto *req) mutable { @@ -159,7 +161,13 @@ public: // a regular HttpResponse does not know about UserData or any of the Websocket upgrade procedure - behavior.upgrade(res, req); + // behavior.compression, (struct us_socket_context_t *) webSocketContext, httpContext, behavior.idleTimeout, behavior.open + + // webSocketContext håller behavior - håller den även httpContext? + + // int, void *, void *, int, + + behavior.upgrade(res, req, (struct us_socket_context_t *) webSocketContext); // if upgrade handler does not upgrade or end within the callback, lift a token? @@ -171,43 +179,12 @@ public: std::string_view secWebSocketProtocol = req->getHeader("sec-websocket-protocol"); std::string_view secWebSocketExtensions = req->getHeader("sec-websocket-extensions"); - res->upgrade(secWebSocketKey, secWebSocketProtocol, secWebSocketExtensions, behavior.compression); + res->template upgrade(secWebSocketKey, secWebSocketProtocol, secWebSocketExtensions, (struct us_socket_context_t *) webSocketContext); } - - - - - /* Move any backpressure */ - std::string backpressure(std::move(((AsyncSocketData *) res->getHttpResponseData())->buffer)); - - /* Keep any fallback buffer alive until we returned from open event, keeping req valid */ - std::string fallback(std::move(res->getHttpResponseData()->salvageFallbackBuffer())); - - /* Destroy HttpResponseData */ - res->getHttpResponseData()->~HttpResponseData(); - - /* Adopting a socket invalidates it, do not rely on it directly to carry any data */ - WebSocket *webSocket = (WebSocket *) us_socket_context_adopt_socket(SSL, - (us_socket_context_t *) webSocketContext, (us_socket_t *) res, sizeof(WebSocketData) + sizeof(UserData)); - - /* Update corked socket in case we got a new one (assuming we always are corked in handlers). */ - webSocket->AsyncSocket::cork(); - - /* Initialize websocket with any moved backpressure intact */ - httpContext->upgradeToWebSocket( - webSocket->init(/*perMessageDeflate*/ false, /*compressOptions*/ 0, std::move(backpressure)) - ); - - /* Arm idleTimeout */ - us_socket_timeout(SSL, (us_socket_t *) webSocket, behavior.idleTimeout); - - /* Default construct the UserData right before calling open handler */ - new (webSocket->getUserData()) UserData; - /* Emit open event and start the timeout */ if (behavior.open) { - behavior.open(webSocket, req); + //behavior.open(webSocket, req); } /* We are going to get uncorked by the Http get return */ diff --git a/src/HttpContext.h b/src/HttpContext.h index c7571e7..ced6f80 100644 --- a/src/HttpContext.h +++ b/src/HttpContext.h @@ -35,6 +35,7 @@ template struct HttpResponse; template struct HttpContext { template friend struct TemplatedApp; + template friend struct HttpResponse; private: HttpContext() = delete; @@ -205,7 +206,7 @@ private: if (us_socket_is_shut_down(SSL, (us_socket_t *) user)) { return nullptr; } - + /* If we were given the last data chunk, reset data handler to ensure following * requests on the same socket won't trigger any previously registered behavior */ if (fin) { diff --git a/src/HttpResponse.h b/src/HttpResponse.h index f164213..e6d7aba 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -27,6 +27,10 @@ #include "WebSocketExtensions.h" #include "WebSocketHandshake.h" +#include "WebSocket.h" +#include "WebSocketContextData.h" + +#include "HttpContext.h" #include "f2/function2.hpp" @@ -164,8 +168,17 @@ private: } } +public: /* This call is identical to end, but will never write content-length and is thus suitable for upgrades */ - void upgrade(std::string_view secWebSocketKey, std::string_view secWebSocketProtocol, std::string_view secWebSocketExtensions, int compression) { + template + void upgrade(std::string_view secWebSocketKey, std::string_view secWebSocketProtocol, + std::string_view secWebSocketExtensions, + struct us_socket_context_t *webSocketContext) { + + /* Extract needed parameters from WebSocketContextData */ + WebSocketContextData *webSocketContextData = (WebSocketContextData *) us_socket_context_ext(SSL, webSocketContext); + int compression = webSocketContextData->compression; + int idleTimeout = webSocketContextData->idleTimeout; /* Note: OpenSSL can be used here to speed this up somewhat */ char secWebSocketAccept[29] = {}; @@ -177,7 +190,6 @@ private: ->writeHeader("Sec-WebSocket-Accept", secWebSocketAccept); /* Select first subprotocol if present */ - //std::string_view secWebSocketProtocol = req->getHeader("sec-websocket-protocol"); if (secWebSocketProtocol.length()) { writeHeader("Sec-WebSocket-Protocol", secWebSocketProtocol.substr(0, secWebSocketProtocol.find(','))); } @@ -221,9 +233,49 @@ private: } internalEnd({nullptr, 0}, 0, false, false); + + /* Grab the httpContext from res */ + HttpContext *httpContext = (HttpContext *) us_socket_context(SSL, (struct us_socket_t *) this); + + /* Move any backpressure */ + std::string backpressure(std::move(((AsyncSocketData *) getHttpResponseData())->buffer)); + + /* Keep any fallback buffer alive until we returned from open event, keeping req valid */ + std::string fallback(std::move(getHttpResponseData()->salvageFallbackBuffer())); + + /* Destroy HttpResponseData */ + getHttpResponseData()->~HttpResponseData(); + + /* Adopting a socket invalidates it, do not rely on it directly to carry any data */ + WebSocket *webSocket = (WebSocket *) us_socket_context_adopt_socket(SSL, + (us_socket_context_t *) webSocketContext, (us_socket_t *) this, sizeof(WebSocketData) + sizeof(UserData)); + + /* Update corked socket in case we got a new one (assuming we always are corked in handlers). */ + webSocket->AsyncSocket::cork(); + + /* Initialize websocket with any moved backpressure intact */ + + + /* Todo: this is the only use of HttpContext! Move that code in here! */ + /* We should not depend on the HttpContext.h! */ + httpContext->upgradeToWebSocket( + webSocket->init(perMessageDeflate, compressOptions, std::move(backpressure)) + ); + + /* Arm idleTimeout */ + us_socket_timeout(SSL, (us_socket_t *) webSocket, /*behavior.*/idleTimeout); + + /* Default construct the UserData right before calling open handler */ + new (webSocket->getUserData()) UserData; + + /* Emit open event and start the timeout */ + if (webSocketContextData->openHandler) { + webSocketContextData->openHandler(webSocket, /*req*/ nullptr); + } + + // if we weren't corked then uncork here! otherwise we were called from httpContext! } -public: /* Immediately terminate this Http response */ using Super::close; diff --git a/src/WebSocket.h b/src/WebSocket.h index 8154c81..692f8ca 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -30,6 +30,7 @@ namespace uWS { template struct WebSocket : AsyncSocket { template friend struct TemplatedApp; + template friend struct HttpResponse; private: typedef AsyncSocket Super; diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index 3bd6558..30a27bd 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -34,6 +34,7 @@ template struct WebSocket; template struct WebSocketContextData { /* The callbacks for this context */ + fu2::unique_function *, HttpRequest *)> openHandler = nullptr; fu2::unique_function *, std::string_view, uWS::OpCode)> messageHandler = nullptr; fu2::unique_function *)> drainHandler = nullptr; fu2::unique_function *, int, std::string_view)> closeHandler = nullptr; @@ -45,6 +46,9 @@ struct WebSocketContextData { size_t maxPayloadLength = 0; int idleTimeout = 0; + /* We do need these for async upgrade */ + int compression; + /* There needs to be a maxBackpressure which will force close everything over that limit */ size_t maxBackpressure = 0; @@ -77,12 +81,12 @@ struct WebSocketContextData { if (!failed) { asyncSocket->timeout(this->idleTimeout); } - + /* Failing here must not immediately close the socket, as that could result in stack overflow, * iterator invalidation and other TopicTree::drain bugs. We may shutdown the reading side of the socket, * causing next iteration to error-close the socket from that context instead, if we want to */ } - + /* If we have too much backpressure, simply skip sending from here */ /* Reserved, unused */ diff --git a/src/WebSocketData.h b/src/WebSocketData.h index c733981..5e1e24d 100644 --- a/src/WebSocketData.h +++ b/src/WebSocketData.h @@ -21,6 +21,7 @@ #include "WebSocketProtocol.h" #include "AsyncSocketData.h" #include "PerMessageDeflate.h" +#include "TopicTree.h" #include