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

Make it stick together just barely

Browse Source
This commit is contained in:
Alex Hultman
2020-06-02 12:40:51 +02:00
parent a8539e7463
commit cb48712b3d
7 changed files with 134 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/EchoServer.cpp
+57 -4
View File
@@ -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<PerSocketData>(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<PerSocketData>(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 */
src/App.h
+12 -35
View File
@@ -86,7 +86,7 @@ public:
int maxPayloadLength = 16 * 1024;
int idleTimeout = 120;
int maxBackpressure = 1 * 1024 * 1024;
fu2::unique_function<void(HttpResponse<SSL> *, HttpRequest *)> upgrade = nullptr;
fu2::unique_function<void(HttpResponse<SSL> *, HttpRequest *, struct us_socket_context_t *)> upgrade = nullptr;
fu2::unique_function<void(uWS::WebSocket<SSL, true> *, HttpRequest *)> open = nullptr;
fu2::unique_function<void(uWS::WebSocket<SSL, true> *, std::string_view, uWS::OpCode)> message = nullptr;
fu2::unique_function<void(uWS::WebSocket<SSL, true> *)> 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<SSL, true> *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<UserData>(secWebSocketKey, secWebSocketProtocol, secWebSocketExtensions, (struct us_socket_context_t *) webSocketContext);
}
/* Move any backpressure */
std::string backpressure(std::move(((AsyncSocketData<SSL> *) 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<SSL, true> *webSocket = (WebSocket<SSL, true> *) 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<SSL>::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 */
src/HttpContext.h
+2 -1
View File
@@ -35,6 +35,7 @@ template<bool> struct HttpResponse;
template <bool SSL>
struct HttpContext {
template<bool> friend struct TemplatedApp;
template<bool> 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) {
src/HttpResponse.h
+55 -3
View File
@@ -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 <typename UserData>
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<SSL> *webSocketContextData = (WebSocketContextData<SSL> *) 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<SSL> *httpContext = (HttpContext<SSL> *) us_socket_context(SSL, (struct us_socket_t *) this);
/* Move any backpressure */
std::string backpressure(std::move(((AsyncSocketData<SSL> *) 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<SSL, true> *webSocket = (WebSocket<SSL, true> *) 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<SSL>::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;
src/WebSocket.h
+1
View File
@@ -30,6 +30,7 @@ namespace uWS {
template <bool SSL, bool isServer>
struct WebSocket : AsyncSocket<SSL> {
template <bool> friend struct TemplatedApp;
template <bool> friend struct HttpResponse;
private:
typedef AsyncSocket<SSL> Super;
src/WebSocketContextData.h
+6 -2
View File
@@ -34,6 +34,7 @@ template <bool, bool> struct WebSocket;
template <bool SSL>
struct WebSocketContextData {
/* The callbacks for this context */
fu2::unique_function<void(uWS::WebSocket<SSL, true> *, HttpRequest *)> openHandler = nullptr;
fu2::unique_function<void(WebSocket<SSL, true> *, std::string_view, uWS::OpCode)> messageHandler = nullptr;
fu2::unique_function<void(WebSocket<SSL, true> *)> drainHandler = nullptr;
fu2::unique_function<void(WebSocket<SSL, true> *, 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 */
src/WebSocketData.h
+1
View File
@@ -21,6 +21,7 @@
#include "WebSocketProtocol.h"
#include "AsyncSocketData.h"
#include "PerMessageDeflate.h"
#include "TopicTree.h"
#include <string>