Emit new 'subscription' events

Author: uNetworkingAB

README.md

@@ -4,7 +4,7 @@
 <i>Simple, secure</i><sup><a href="https://github.com/uNetworking/uWebSockets/tree/master/fuzzing#fuzz-testing-of-various-parsers-and-mocked-examples">1</a></sup><i> & standards compliant</i><sup><a href="https://unetworking.github.io/uWebSockets.js/report.pdf">2</a></sup><i> web server for the most demanding</i><sup><a href="https://github.com/uNetworking/uWebSockets/tree/master/benchmarks#benchmark-driven-development">3</a></sup><i> of applications.</i> <a href="https://github.com/uNetworking/uWebSockets/blob/master/misc/READMORE.md">Read more...</a>
 <br><br>
 
-<a href="https://github.com/uNetworking/uWebSockets/releases"><img src="https://img.shields.io/github/v/release/uNetworking/uWebSockets"></a> <a href="https://lgtm.com/projects/g/uNetworking/uWebSockets/context:cpp"><img alt="Language grade: C/C++" src="https://img.shields.io/lgtm/grade/cpp/g/uNetworking/uWebSockets.svg?logo=lgtm&logoWidth=18"/></a> <a href="https://osv.dev/list?q=uwebsockets&affected_only=true&page=1&ecosystem=OSS-Fuzz"><img src="https://oss-fuzz-build-logs.storage.googleapis.com/badges/uwebsockets.svg" /></a> <img src="https://img.shields.io/badge/downloads-65%20million-pink" />
+<a href="https://github.com/uNetworking/uWebSockets/releases"><img src="https://img.shields.io/github/v/release/uNetworking/uWebSockets"></a> <a href="https://osv.dev/list?q=uwebsockets&affected_only=true&page=1&ecosystem=OSS-Fuzz"><img src="https://oss-fuzz-build-logs.storage.googleapis.com/badges/uwebsockets.svg" /></a> <img src="https://img.shields.io/badge/downloads-70%20million-pink" />
 
 </div>
 <br><br>

src/TopicTree.h

@@ -158,7 +158,7 @@ struct TopicTree {
     }
 
     /* Subscribe fails if we already are subscribed */
-    bool subscribe(Subscriber *s, std::string_view topic) {
+    Topic *subscribe(Subscriber *s, std::string_view topic) {
         /* Notify user that they are doing something wrong here */
         checkIteratingSubscriber(s);
 
@@ -173,42 +173,44 @@ struct TopicTree {
         /* Insert us in topic, insert topic in us */
         auto [it, inserted] = s->topics.insert(topicPtr);
         if (!inserted) {
-            return false;
+            return nullptr;
         }
         topicPtr->insert(s);
 
         /* Success */
-        return true;
+        return topicPtr;
     }
 
-    /* Returns ok, last */
-    std::pair<bool, bool> unsubscribe(Subscriber *s, std::string_view topic) {
+    /* Returns ok, last, newCount */
+    std::tuple<bool, bool, int> unsubscribe(Subscriber *s, std::string_view topic) {
         /* Notify user that they are doing something wrong here */
         checkIteratingSubscriber(s);
 
         /* Lookup topic */
         Topic *topicPtr = lookupTopic(topic);
         if (!topicPtr) {
             /* If the topic doesn't exist we are assumed to still be subscribers of something */
-            return {false, false};
+            return {false, false, -1};
         }
 
         /* Erase from our list first */
         if (s->topics.erase(topicPtr) == 0) {
-            return {false, false};
+            return {false, false, -1};
         }
 
         /* Remove us from topic */
         topicPtr->erase(s);
 
+        int newCount = topicPtr->size();
+
         /* If there is no subscriber to this topic, remove it */
         if (!topicPtr->size()) {
             /* Unique_ptr deletes the topic */
             topics.erase(topic);
         }
 
         /* If we don't hold any topics we are to be freed altogether */
-        return {true, topics.size() == 0};
+        return {true, s->topics.size() == 0, newCount};
     }
 
     /* Factory function for creating a Subscriber */

src/WebSocket.h

@@ -195,6 +195,11 @@ struct WebSocket : AsyncSocket<SSL> {
         /* Set shorter timeout (use ping-timeout) to avoid long hanging sockets after end() on broken connections */
         Super::timeout(webSocketContextData->idleTimeoutComponents.second);
 
+        /* At this point we iterate all currently held subscriptions and emit an event for all of them */
+        for (Topic *t : webSocketData->subscriber->topics) {
+            webSocketContextData->subscriptionHandler(this, t->name, (int) t->size() - 1, (int) t->size());
+        }
+
         /* Make sure to unsubscribe from any pub/sub node at exit */
         webSocketContextData->topicTree->freeSubscriber(webSocketData->subscriber);
         webSocketData->subscriber = nullptr;
@@ -234,7 +239,11 @@ struct WebSocket : AsyncSocket<SSL> {
         }
 
         /* Cannot return numSubscribers as this is only for this particular websocket context */
-        webSocketContextData->topicTree->subscribe(webSocketData->subscriber, topic);
+        Topic *topicOrNull = webSocketContextData->topicTree->subscribe(webSocketData->subscriber, topic);
+        if (topicOrNull && webSocketContextData->subscriptionHandler) {
+            /* Emit this socket, the topic, new count, old count */
+            webSocketContextData->subscriptionHandler(this, topic, (int) topicOrNull->size(), (int) topicOrNull->size() - 1);
+        }
 
         /* Subscribe always succeeds */
         return true;
@@ -251,7 +260,11 @@ struct WebSocket : AsyncSocket<SSL> {
         if (!webSocketData->subscriber) { return false; }
 
         /* Cannot return numSubscribers as this is only for this particular websocket context */
-        auto [ok, last] = webSocketContextData->topicTree->unsubscribe(webSocketData->subscriber, topic);
+        auto [ok, last, newCount] = webSocketContextData->topicTree->unsubscribe(webSocketData->subscriber, topic);
+        /* Emit subscription event if last */
+        if (ok && webSocketContextData->subscriptionHandler) {
+            webSocketContextData->subscriptionHandler(this, topic, newCount, newCount + 1);
+        }
 
         /* Free us as subscribers if we unsubscribed from our last topic */
         if (ok && last) {