From 3fa0fbc5fa17b3b6c6560549465d3092acb95a8a Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Fri, 21 Dec 2018 04:31:37 +0100 Subject: [PATCH] Add a few more events --- examples/EchoServer.cpp | 42 +++++++++++++++++-------------- misc/main.cpp | 51 ++++++++++++++++++++++---------------- src/App.h | 9 +++++-- src/AsyncSocket.h | 7 ++++++ src/WebSocket.h | 11 ++++++++ src/WebSocketContext.h | 16 +++++++----- src/WebSocketContextData.h | 4 ++- 7 files changed, 91 insertions(+), 49 deletions(-) diff --git a/examples/EchoServer.cpp b/examples/EchoServer.cpp index 0df38cd..cff3a49 100644 --- a/examples/EchoServer.cpp +++ b/examples/EchoServer.cpp @@ -1,29 +1,35 @@ #include "App.h" int main() { + /* ws->getUserData returns one of these */ + struct PerSocketData { + + }; + /* Very simple WebSocket echo server */ - uWS::App().ws("/*", { - /*.compression = */true, - /*.maxPayloadLength*/ - - /*.open = */[](auto *ws, auto *req) { + uWS::App().ws("/*", { + /* Settings */ + .compression = true, + .maxPayloadLength = 16 * 1024, + /* Handlers */ + .open = [](auto *ws, auto *req) { }, - /*.message = */[](auto *ws, std::string_view message, uWS::OpCode opCode) { + .message = [](auto *ws, std::string_view message, uWS::OpCode opCode) { ws->send(message, opCode); + }, + .drain = [](auto *ws) { + /* Check getBufferedAmount here */ + }, + .ping = [](auto *ws) { + + }, + .pong = [](auto *ws) { + + }, + .close = [](auto *ws, int code, std::string_view message) { + } - /*.drain = []() { - - }, - .ping = []() { - - }, - .pong = []() { - - }, - .close = []() { - - }*/ }).listen(9001, [](auto *token) { if (token) { std::cout << "Listening on port " << 9001 << std::endl; diff --git a/misc/main.cpp b/misc/main.cpp index b561b7b..2b9a85c 100644 --- a/misc/main.cpp +++ b/misc/main.cpp @@ -6,7 +6,7 @@ int main(int argc, char **argv) { struct PerSocketData { - + int hello; }; /*const char *key_file_name; @@ -14,34 +14,41 @@ int main(int argc, char **argv) { const char *passphrase; const char *dh_params_file_name;*/ - uWS::SSLApp({ + uWS::App(/*SSLApp({ "/home/alexhultman/key.pem", "/home/alexhultman/cert.pem", "1234" - }).get("/hello", [](auto *res, auto *req) { + }*/).get("/hello", [](auto *res, auto *req) { res->end("Hello HTTP!"); - }).ws("/*", { - /*.compression = */true, - /*.maxPayloadLength*/ - - /*.open = */[](auto *ws, auto *req) { - + }).ws("/*", { + /* Settings */ + .compression = true, + .maxPayloadLength = 16 * 1024, + /* Handlers */ + .open = [](auto *ws, auto *req) { + std::cout << "WebSocket connected" << std::endl; + /* Access per socket data */ + /*PerSocketData *perSocketData = */ws->getUserData(); + /*perSocketData->hello = 13;*/ }, - /*.message = */[](auto *ws, std::string_view message, uWS::OpCode opCode) { + .message = [](auto *ws, std::string_view message, uWS::OpCode opCode) { ws->send(message, opCode); + }, + .drain = [](auto *ws) { + std::cout << "Drainage: " << ws->getBufferedAmount() << std::endl; + }, + .ping = [](auto *ws) { + std::cout << "Ping" << std::endl; + }, + .pong = [](auto *ws) { + std::cout << "Pong" << std::endl; + }, + .close = [](auto *ws, int code, std::string_view message) { + std::cout << "WebSocket disconnected: " << code << "[" << message << "]" << std::endl; + /* Access per socket data */ + //PerSocketData *perSocketData = ws->getUserData(); + //std::cout << "OK per socket data: " << (perSocketData->hello == 13) << std::endl; } - /*.drain = []() { - - }, - .ping = []() { - - }, - .pong = []() { - - }, - .close = []() { - - }*/ }).listen(9001, [](auto *token) { if (token) { std::cout << "Listening on port " << 3000 << std::endl; diff --git a/src/App.h b/src/App.h index 90f982e..a6fcf8d 100644 --- a/src/App.h +++ b/src/App.h @@ -54,8 +54,13 @@ public: struct WebSocketBehavior { bool compression = false; + int maxPayloadLength = 16 * 1024; std::function *, HttpRequest *)> open = nullptr; std::function *, std::string_view, uWS::OpCode)> message = nullptr; + std::function *)> drain = nullptr; + std::function *)> ping = nullptr; + std::function *)> pong = nullptr; + std::function *, int, std::string_view)> close = nullptr; }; template @@ -75,6 +80,8 @@ public: /* Copy all handlers */ webSocketContext->getExt()->messageHandler = behavior.message; + webSocketContext->getExt()->drainHandler = behavior.drain; + webSocketContext->getExt()->closeHandler = behavior.close; return get(pattern, [webSocketContext, this, behavior](auto *res, auto *req) { /* If we have this header set, it's a websocket */ @@ -125,8 +132,6 @@ public: behavior.open(webSocket, req); } - std::cout << "oh hey!" << std::endl; - } else { /* For now we do not support having HTTP and websocket routes on the same URL */ res->close(); diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index bbbe533..a289c69 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -95,6 +95,13 @@ protected: } } + /* Returns the user space backpressure. */ + int getBufferedAmount() { + AsyncSocketData *asyncSocketData = (AsyncSocketData *) getExt(); + + return asyncSocketData->buffer.size(); + } + /* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer. Always drain if possible. * Returns pair of bytes written (anywhere) and wheter or not this call resulted in the polling for * writable (or we are in a state that implies polling for writable). */ diff --git a/src/WebSocket.h b/src/WebSocket.h index 728ed03..ebcb9b5 100644 --- a/src/WebSocket.h +++ b/src/WebSocket.h @@ -39,6 +39,17 @@ private: } public: + /* Returns pointer to the per socket user data */ + //template + void *getUserData() { + WebSocketData *webSocketData = (WebSocketData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this); + + return nullptr; + } + + /* See AsyncSocket */ + using Super::getBufferedAmount; + /* Send or buffer a WebSocket frame, compressed or not. Returns false on increased user space backpressure. */ bool send(std::string_view message, uWS::OpCode opCode = uWS::OpCode::BINARY, bool compress = false) { /* Transform the message to compressed domain if requested */ diff --git a/src/WebSocketContext.h b/src/WebSocketContext.h index 992d934..65a4bc2 100644 --- a/src/WebSocketContext.h +++ b/src/WebSocketContext.h @@ -47,7 +47,7 @@ private: static bool setCompressed(uWS::WebSocketState *wState, void *s) { //WebSocketData *webSocketData = (WebSocketData *) us_socket_ext((us_socket *) s); - std::cout << "set compressed" << std::endl; + //std::cout << "set compressed" << std::endl; WebSocketData *webSocketData = (WebSocketData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) s); @@ -264,14 +264,13 @@ private: std::cout << "close!" << std::endl; + + return s; }); /* Handle WebSocket data streams */ static_dispatch(us_ssl_socket_context_on_data, us_socket_context_on_data)(getSocketContext(), [](auto *s, char *data, int length) { - - //std::cout << "websocket data" << std::endl; - /* We always cork on data */ AsyncSocket *webSocket = (AsyncSocket *) s; webSocket->cork(); @@ -279,13 +278,14 @@ private: /* We need the websocket data */ WebSocketData *wsState = (WebSocketData *) (static_dispatch(us_ssl_socket_ext, us_socket_ext)(s)); - // this parser requires almost no time -> 215k req/sec of 215k possible + /* This parser has virtually no overhead */ uWS::WebSocketProtocol>::consume(data, length, wsState, s); - // todo: check for failures here just like for HTTP webSocket->uncork(); + // I guess we need to check drain here + // are we shutdown? if (wsState->isShuttingDown) { webSocket->shutdown(); @@ -306,6 +306,10 @@ private: // check for failures and shutdown just like in data event webSocket->write(nullptr, 0); // drainage - also check for shutdown! + // call drain here + + + return s; }); diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h index da2ec78..abeda42 100644 --- a/src/WebSocketContextData.h +++ b/src/WebSocketContextData.h @@ -28,8 +28,10 @@ template struct WebSocket; template struct WebSocketContextData { - + /* The callbacks for this context */ std::function *, std::string_view, uWS::OpCode)> messageHandler; + std::function *)> drainHandler; + std::function *, int, std::string_view)> closeHandler; }; }