From 973a975915bd36b148b1cc3e07bc361db53fc006 Mon Sep 17 00:00:00 2001
From: Alex Hultman <alexhultman@gmail.com>
Date: Thu, 30 Nov 2023 17:38:12 +0100
Subject: [PATCH] Add .dropped events for pub/sub backpressure notification

---
 examples/EchoServer.cpp    | 3 +++
 src/App.h                  | 2 ++
 src/WebSocket.h            | 6 ++++++
 src/WebSocketContextData.h | 1 +
 4 files changed, 12 insertions(+)

diff --git a/examples/EchoServer.cpp b/examples/EchoServer.cpp
index 1e73416..671ffde 100644
--- a/examples/EchoServer.cpp
+++ b/examples/EchoServer.cpp
@@ -38,6 +38,9 @@ int main() {
              * benchmarking of large message sending without compression */
             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*/) {
             /* Check ws->getBufferedAmount() here */
         },
diff --git a/src/App.h b/src/App.h
index 87f5c14..a9ffd4c 100644
--- a/src/App.h
+++ b/src/App.h
@@ -242,6 +242,7 @@ public:
         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> *, 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> *, std::string_view)> ping = nullptr;
         MoveOnlyFunction<void(WebSocket<SSL, true, UserData> *, std::string_view)> pong = nullptr;
@@ -372,6 +373,7 @@ public:
         /* Copy all handlers */
         webSocketContext->getExt()->openHandler = std::move(behavior.open);
         webSocketContext->getExt()->messageHandler = std::move(behavior.message);
+        webSocketContext->getExt()->droppedHandler = std::move(behavior.dropped);
         webSocketContext->getExt()->drainHandler = std::move(behavior.drain);
         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 {
diff --git a/src/WebSocket.h b/src/WebSocket.h
index 1eedc22..f2663e5 100644
--- a/src/WebSocket.h
+++ b/src/WebSocket.h
@@ -100,6 +100,12 @@ public:
             if (webSocketContextData->closeOnBackpressureLimit) {
                 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;
         }
 
diff --git a/src/WebSocketContextData.h b/src/WebSocketContextData.h
index e5ae49a..7f04aa6 100644
--- a/src/WebSocketContextData.h
+++ b/src/WebSocketContextData.h
@@ -59,6 +59,7 @@ public:
     /* The callbacks for this context */
     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)> droppedHandler = 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> *, int, std::string_view)> closeHandler = nullptr;