Add basic maxBackpressure for pubsub

src/App.h

@@ -89,6 +89,7 @@ public:
         CompressOptions compression = DISABLED;
         int maxPayloadLength = 16 * 1024;
         int idleTimeout = 120;
+        int maxBackpressure = 1 * 1024 * 1204;
         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;
@@ -134,6 +135,7 @@ public:
         /* Copy settings */
         webSocketContext->getExt()->maxPayloadLength = behavior.maxPayloadLength;
         webSocketContext->getExt()->idleTimeout = behavior.idleTimeout;
+        webSocketContext->getExt()->maxBackpressure = behavior.maxBackpressure;
 
         return std::move(get(pattern, [webSocketContext, httpContext = this->httpContext, behavior = std::move(behavior)](auto *res, auto *req) mutable {
 

src/WebSocketContextData.h

@@ -42,7 +42,7 @@ struct WebSocketContextData {
     int idleTimeout = 0;
 
     /* There needs to be a maxBackpressure which will force close everything over that limit */
-    size_t maxBackpressure = 16 * 1024;
+    size_t maxBackpressure = 0;
 
     /* Each websocket context has a topic tree for pub/sub */
     TopicTree topicTree;
@@ -51,8 +51,17 @@ struct WebSocketContextData {
         /* We rely on writing to regular asyncSockets */
         auto *asyncSocket = (AsyncSocket<SSL> *) s->user;
 
-        asyncSocket->timeout(this->idleTimeout);
-        asyncSocket->write(data.data(), data.length());
+        auto [written, failed] = asyncSocket->write(data.data(), data.length());
+        if (!failed) {
+            asyncSocket->timeout(this->idleTimeout);
+        } else {
+            /* Note: this assumes we are not corked, as corking will swallow things and fail later on */
+
+            /* Check if we now have too much backpressure (todo: don't buffer up before check) */
+            if (asyncSocket->getBufferedAmount() > maxBackpressure) {
+                asyncSocket->close();
+            }
+        }
 
         /* Reserved, unused */
         return 0;