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

Fix up WebSocketContext.h

Browse Source
This commit is contained in:
Alex Hultman
2019-02-10 01:08:10 +01:00
parent f1bdbc968b
commit a419221a28
1 changed files with 30 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/WebSocketContext.h
+30 -59
View File
@@ -23,8 +23,6 @@
#include "WebSocketData.h" #include "WebSocketData.h"
#include "WebSocket.h" #include "WebSocket.h"
// todo: this module needs fixing! see below!
namespace uWS { namespace uWS {
template <bool SSL, bool isServer> template <bool SSL, bool isServer>
@@ -61,9 +59,7 @@ private:
/* Returns true on breakage */ /* Returns true on breakage */
static bool handleFragment(char *data, size_t length, unsigned int remainingBytes, int opCode, bool fin, uWS::WebSocketState<isServer> *webSocketState, void *s) { static bool handleFragment(char *data, size_t length, unsigned int remainingBytes, int opCode, bool fin, uWS::WebSocketState<isServer> *webSocketState, void *s) {
/* WebSocketData and WebSocketContextData */ /* WebSocketData and WebSocketContextData */
WebSocketContextData<SSL> *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, WebSocketContextData<SSL> *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, us_new_socket_context(SSL, (us_new_socket_t *) s));
us_new_socket_context(SSL, (us_new_socket_t *) s)
);
WebSocketData *webSocketData = (WebSocketData *) us_new_socket_ext(SSL, (us_new_socket_t *) s); WebSocketData *webSocketData = (WebSocketData *) us_new_socket_ext(SSL, (us_new_socket_t *) s);
/* Is this a non-control frame? */ /* Is this a non-control frame? */
@@ -75,12 +71,7 @@ private:
if (webSocketData->compressionStatus == WebSocketData::CompressionStatus::COMPRESSED_FRAME) { if (webSocketData->compressionStatus == WebSocketData::CompressionStatus::COMPRESSED_FRAME) {
webSocketData->compressionStatus = WebSocketData::CompressionStatus::ENABLED; webSocketData->compressionStatus = WebSocketData::CompressionStatus::ENABLED;
LoopData *loopData = (LoopData *)us_loop_ext( LoopData *loopData = (LoopData *) us_loop_ext(us_new_socket_context_loop(SSL, us_new_socket_context(SSL, (us_new_socket_t *) s)));
us_new_socket_context_loop(SSL,
us_new_socket_context(SSL, (us_new_socket_t *)s)
)
);
std::string_view inflatedFrame = loopData->inflationStream->inflate(loopData->zlibContext, {data, length}, webSocketContextData->maxPayloadLength); std::string_view inflatedFrame = loopData->inflationStream->inflate(loopData->zlibContext, {data, length}, webSocketContextData->maxPayloadLength);
if (!inflatedFrame.length()) { if (!inflatedFrame.length()) {
forceClose(webSocketState, s); forceClose(webSocketState, s);
@@ -100,7 +91,7 @@ private:
/* Emit message event & break if we are closed or shut down when returning */ /* Emit message event & break if we are closed or shut down when returning */
if (webSocketContextData->messageHandler) { if (webSocketContextData->messageHandler) {
webSocketContextData->messageHandler((WebSocket<SSL, isServer> *) s, std::string_view(data, length), (uWS::OpCode) opCode); webSocketContextData->messageHandler((WebSocket<SSL, isServer> *) s, std::string_view(data, length), (uWS::OpCode) opCode);
if (us_new_socket_is_closed(SSL, (us_new_socket_t *)s) || webSocketData->isShuttingDown) { if (us_new_socket_is_closed(SSL, (us_new_socket_t *) s) || webSocketData->isShuttingDown) {
return true; return true;
} }
} }
@@ -153,7 +144,7 @@ private:
/* Emit message and check for shutdown or close */ /* Emit message and check for shutdown or close */
if (webSocketContextData->messageHandler) { if (webSocketContextData->messageHandler) {
webSocketContextData->messageHandler((WebSocket<SSL, isServer> *) s, std::string_view(data, length), (uWS::OpCode) opCode); webSocketContextData->messageHandler((WebSocket<SSL, isServer> *) s, std::string_view(data, length), (uWS::OpCode) opCode);
if (us_new_socket_is_closed(SSL, (us_new_socket_t *)s) || webSocketData->isShuttingDown) { if (us_new_socket_is_closed(SSL, (us_new_socket_t *) s) || webSocketData->isShuttingDown) {
return true; return true;
} }
} }
@@ -221,9 +212,7 @@ private:
} }
static bool refusePayloadLength(uint64_t length, uWS::WebSocketState<isServer> *wState, void *s) { static bool refusePayloadLength(uint64_t length, uWS::WebSocketState<isServer> *wState, void *s) {
WebSocketContextData<SSL> *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, auto *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, us_new_socket_context(SSL, (us_new_socket_t *) s));
us_new_socket_context(SSL, (us_new_socket_t *)s)
);
/* Return true for refuse, false for accept */ /* Return true for refuse, false for accept */
return webSocketContextData->maxPayloadLength < length; return webSocketContextData->maxPayloadLength < length;
@@ -241,9 +230,7 @@ private:
WebSocketData *webSocketData = (WebSocketData *) (us_new_socket_ext(SSL, s)); WebSocketData *webSocketData = (WebSocketData *) (us_new_socket_ext(SSL, s));
if (!webSocketData->isShuttingDown) { if (!webSocketData->isShuttingDown) {
/* Emit close event */ /* Emit close event */
WebSocketContextData<SSL> *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, auto *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, us_new_socket_context(SSL, (us_new_socket_t *) s));
us_new_socket_context(SSL, (us_new_socket_t *)s)
);
if (webSocketContextData->closeHandler) { if (webSocketContextData->closeHandler) {
webSocketContextData->closeHandler((WebSocket<SSL, true> *) s, 1006, {}); webSocketContextData->closeHandler((WebSocket<SSL, true> *) s, 1006, {});
@@ -271,40 +258,28 @@ private:
return s; return s;
} }
/* Everytime we get data, we reset the timeout to our idleTimeout, that's the only timer we have */ auto *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, us_new_socket_context(SSL, (us_new_socket_t *) s));
auto *asyncSocket = (AsyncSocket<SSL> *) s;
/* If not in websocket shutdown state, for every */
// återställ inte om vi är i shutdown state, dvs, ge den inte massa tid på sig att skicka massa skit-frames mellan och upphålla oss!
WebSocketContextData<SSL> *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL,
us_new_socket_context(SSL, (us_new_socket_t *) s)
);
us_new_socket_timeout(SSL, (us_new_socket_t *) s, webSocketContextData->idleTimeout);
/* Every time we get data and not in shutdown state we simply reset the timeout */
asyncSocket->timeout(webSocketContextData->idleTimeout);
/* We always cork on data */ /* We always cork on data */
AsyncSocket<SSL> *webSocket = (AsyncSocket<SSL> *) s; asyncSocket->cork();
webSocket->cork();
/* This parser has virtually no overhead */ /* This parser has virtually no overhead */
uWS::WebSocketProtocol<isServer, WebSocketContext<SSL, isServer>>::consume(data, length, (WebSocketState<isServer> *) webSocketData, s); uWS::WebSocketProtocol<isServer, WebSocketContext<SSL, isServer>>::consume(data, length, (WebSocketState<isServer> *) webSocketData, s);
// todo: we need to check for close and shutdown here? as we just emitted a bunch of message/close events! /* Uncorking a closed socekt is fine, in fact it is needed */
asyncSocket->uncork();
// todo: check for failures here just like for HTTP /* If uncorking was successful and we are in shutdown state then send TCP FIN */
webSocket->uncork(); if (asyncSocket->getBufferedAmount() == 0) {
/* We can now be in shutdown state */
// cannot do anything else if closed if (webSocketData->isShuttingDown) {
if (us_new_socket_is_closed(SSL, (us_new_socket_t *) s)) { /* Shutting down a closed socket is handled by uSockets and just fine */
return s; asyncSocket->shutdown();
} }
// I guess we need to check drain here - emit drain if we had to poll for writable
// are we shutdown? can onnly call this if we did succeed uncork!
if (webSocketData->isShuttingDown) {
webSocket->shutdown();
} }
return s; return s;
@@ -318,35 +293,31 @@ private:
return s; return s;
} }
AsyncSocket<SSL> *webSocket = (AsyncSocket<SSL> *) s; AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
WebSocketData *webSocketData = (WebSocketData *)(us_new_socket_ext(SSL, s)); WebSocketData *webSocketData = (WebSocketData *)(us_new_socket_ext(SSL, s));
/* We store old backpressure since it is unclear whether write drained anything */ /* We store old backpressure since it is unclear whether write drained anything */
int backpressure = webSocket->getBufferedAmount(); int backpressure = asyncSocket->getBufferedAmount();
/* Drain as much as possible */ /* Drain as much as possible */
webSocket->write(nullptr, 0); asyncSocket->write(nullptr, 0);
/* Behavior: if we actively drain backpressure, always reset timeout (even if we are in shutdown) */ /* Behavior: if we actively drain backpressure, always reset timeout (even if we are in shutdown) */
if (backpressure < webSocket->getBufferedAmount()) { if (backpressure < asyncSocket->getBufferedAmount()) {
WebSocketContextData<SSL> *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, auto *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, us_new_socket_context(SSL, (us_new_socket_t *) s));
us_new_socket_context(SSL, (us_new_socket_t *)s) asyncSocket->timeout(webSocketContextData->idleTimeout);
);
webSocket->timeout(webSocketContextData->idleTimeout);
} }
/* Are we in (WebSocket) shutdown mode? */ /* Are we in (WebSocket) shutdown mode? */
if (webSocketData->isShuttingDown) { if (webSocketData->isShuttingDown) {
/* Check if we just now drained completely */ /* Check if we just now drained completely */
if (webSocket->getBufferedAmount() == 0) { if (asyncSocket->getBufferedAmount() == 0) {
/* Now perform the actual TCP/TLS shutdown which was postponed due to backpressure */ /* Now perform the actual TCP/TLS shutdown which was postponed due to backpressure */
webSocket->shutdown(); asyncSocket->shutdown();
} }
} else if (backpressure > webSocket->getBufferedAmount()) { } else if (backpressure > asyncSocket->getBufferedAmount()) {
/* Only call drain if we actually drained backpressure */ /* Only call drain if we actually drained backpressure */
WebSocketContextData<SSL> *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, auto *webSocketContextData = (WebSocketContextData<SSL> *) us_new_socket_context_ext(SSL, us_new_socket_context(SSL, (us_new_socket_t *) s));
us_new_socket_context(SSL, (us_new_socket_t *)s)
);
if (webSocketContextData->drainHandler) { if (webSocketContextData->drainHandler) {
webSocketContextData->drainHandler((WebSocket<SSL, isServer> *) s); webSocketContextData->drainHandler((WebSocket<SSL, isServer> *) s);
} }