Add .dropped events for pub/sub backpressure notification
This commit is contained in:
@@ -38,6 +38,9 @@ int main() {
|
|||||||
* benchmarking of large message sending without compression */
|
* benchmarking of large message sending without compression */
|
||||||
ws->send(message, opCode, message.length() < 16 * 1024);
|
ws->send(message, opCode, message.length() < 16 * 1024);
|
||||||
},
|
},
|
||||||
|
.dropped = [](auto */*ws*/, std::string_view /*message*/, uWS::OpCode /*opCode*/) {
|
||||||
|
/* A message was dropped due to set maxBackpressure and closeOnBackpressureLimit limit */
|
||||||
|
},
|
||||||
.drain = [](auto */*ws*/) {
|
.drain = [](auto */*ws*/) {
|
||||||
/* Check ws->getBufferedAmount() here */
|
/* Check ws->getBufferedAmount() here */
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -242,6 +242,7 @@ public:
|
|||||||
MoveOnlyFunction<void(HttpResponse<SSL> *, HttpRequest *, struct us_socket_context_t *)> upgrade = nullptr;
|
MoveOnlyFunction<void(HttpResponse<SSL> *, HttpRequest *, struct us_socket_context_t *)> upgrade = nullptr;
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *)> open = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *)> open = nullptr;
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *, std::string_view, OpCode)> message = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *, std::string_view, OpCode)> message = nullptr;
|
||||||
|
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *, std::string_view, OpCode)> dropped = nullptr;
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *)> drain = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *)> drain = nullptr;
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *, std::string_view)> ping = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *, std::string_view)> ping = nullptr;
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *, std::string_view)> pong = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *, std::string_view)> pong = nullptr;
|
||||||
@@ -372,6 +373,7 @@ public:
|
|||||||
/* Copy all handlers */
|
/* Copy all handlers */
|
||||||
webSocketContext->getExt()->openHandler = std::move(behavior.open);
|
webSocketContext->getExt()->openHandler = std::move(behavior.open);
|
||||||
webSocketContext->getExt()->messageHandler = std::move(behavior.message);
|
webSocketContext->getExt()->messageHandler = std::move(behavior.message);
|
||||||
|
webSocketContext->getExt()->droppedHandler = std::move(behavior.dropped);
|
||||||
webSocketContext->getExt()->drainHandler = std::move(behavior.drain);
|
webSocketContext->getExt()->drainHandler = std::move(behavior.drain);
|
||||||
webSocketContext->getExt()->subscriptionHandler = std::move(behavior.subscription);
|
webSocketContext->getExt()->subscriptionHandler = std::move(behavior.subscription);
|
||||||
webSocketContext->getExt()->closeHandler = std::move([closeHandler = std::move(behavior.close)](WebSocket<SSL, true, UserData> *ws, int code, std::string_view message) mutable {
|
webSocketContext->getExt()->closeHandler = std::move([closeHandler = std::move(behavior.close)](WebSocket<SSL, true, UserData> *ws, int code, std::string_view message) mutable {
|
||||||
|
|||||||
@@ -100,6 +100,12 @@ public:
|
|||||||
if (webSocketContextData->closeOnBackpressureLimit) {
|
if (webSocketContextData->closeOnBackpressureLimit) {
|
||||||
us_socket_shutdown_read(SSL, (us_socket_t *) this);
|
us_socket_shutdown_read(SSL, (us_socket_t *) this);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* It is okay to call send again from within this callback since we immediately return with DROPPED afterwards */
|
||||||
|
if (webSocketContextData->droppedHandler) {
|
||||||
|
webSocketContextData->droppedHandler(this, message, opCode);
|
||||||
|
}
|
||||||
|
|
||||||
return DROPPED;
|
return DROPPED;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -59,6 +59,7 @@ public:
|
|||||||
/* The callbacks for this context */
|
/* The callbacks for this context */
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *)> openHandler = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *)> openHandler = nullptr;
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *, std::string_view, OpCode)> messageHandler = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *, std::string_view, OpCode)> messageHandler = nullptr;
|
||||||
|
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *, std::string_view, OpCode)> droppedHandler = nullptr;
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *)> drainHandler = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *)> drainHandler = nullptr;
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *, std::string_view, int, int)> subscriptionHandler = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *, std::string_view, int, int)> subscriptionHandler = nullptr;
|
||||||
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *, int, std::string_view)> closeHandler = nullptr;
|
MoveOnlyFunction<void(WebSocket<SSL, true, USERDATA> *, int, std::string_view)> closeHandler = nullptr;
|
||||||
|
|||||||
Reference in New Issue
Block a user