Fix up unsubscribeAll

src/TopicTree.h

@@ -1,195 +0,0 @@
-/*
- * Authored by Alex Hultman, 2018-2019.
- * Intellectual property of third-party.
-
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
-
- *     http://www.apache.org/licenses/LICENSE-2.0
-
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/* Every WebSocketContext holds one TopicTree */
-#include "Loop.h"
-#include "AsyncSocket.h"
-
-#ifndef UWS_TOPICTREE_H
-#define UWS_TOPICTREE_H
-
-#include <map>
-#include <string>
-#include <vector>
-#include <set>
-
-// todo: obviously this module is WIP
-
-namespace uWS {
-
-    // publishing to a node, then another node, then another node should prioritize draining that way
-    // sending and publishing will interleave undefined, they are separate streams
-
-struct TopicTree {
-private:
-    struct Node : std::map<std::string, Node *> {
-        /* Add and/or lookup node from topic */
-        Node *get(std::string topic) {
-            std::pair<std::map<std::string, Node *>::iterator, bool> p = insert({topic, nullptr});
-            if (p.second) {
-                return p.first->second = new Node;
-            } else {
-                return p.first->second;
-            }
-        }
-
-        /* Subscribers have int backpressureOffset */
-        std::set<void *> subscribers;
-
-        /* Current shared message */
-        std::string sharedMessage;
-
-        /* Backpressure is stored linearly up to a limit */
-        std::string backpressure;
-        unsigned int backpressureOffset = 0;
-
-
-    } topicToNode;
-
-    std::map<void *, std::vector<Node *>> socketToNodeList;
-
-    /* Nodes that hold something to send this iteration */
-    std::set<Node *> pubNodes;
-
-    /* Settings */
-    bool mergePublishedMessages = false;
-
-    /* Where we store prepared messages to send */
-    //std::string preparedMessage;
-
-public:
-
-    ~TopicTree() {
-        /* We have a few leaks here, I think */
-    }
-
-    TopicTree() {
-        /* Dynamically hook us up with the Loop post handler */
-        Loop::get()->addPostHandler([this](Loop *loop) {
-
-            if (!pubNodes.size()) {
-                return;
-            }
-
-            /* We say that all senders get their own message as well, for now being */
-
-            // if a node is very likely used together with another node?
-
-            for (Node *topicNode : pubNodes) {
-                for (auto /*[*/ws/*, valid]*/ : topicNode->subscribers) {
-                    AsyncSocket<false> *asyncSocket = (AsyncSocket<false> *) ws; // assumes non-SSL for now
-
-                    /* Writing optionally raw data */
-                    auto [written, failed] = asyncSocket->write(topicNode->sharedMessage.data(), topicNode->sharedMessage.length(), true, 0);
-
-                    /* We should probably reset timeout for a WebSocket getting something sent */
-
-
-                    /* Every subscriber to a topicNode will have int backpressure cursor to this room */
-
-                    /* How far we wrote will be stored in the WebSocket's Pub/sub block and drained before any other sending (we need to fail sending if already sending pubsub) */
-
-                    /* All messages not fully sent, will be stored in the topictree with an index so that websocket can refer to it by two index: what buffer, what offset */
-                    /* If total backpressure of the topictree is larger than a set limit we close all the slow receivers */
-
-                    /* It is also possible to move topictree backpressure to the websockets themselves, if only one */
-                }
-
-                /* If not all sockets managed to send this message, move it to backpressure */
-                topicNode->sharedMessage.clear();
-            }
-            pubNodes.clear();
-
-
-        });
-    }
-
-    /* WebSocket.subscribe will lookup the Loop and subscribe in its tree */
-    void subscribe(std::string topic, void *connection, bool *valid) {
-        Node *curr = &topicToNode;
-        for (int i = 0; i < topic.length(); i++) {
-            int start = i;
-            while (topic[i] != '/' && i < topic.length()) {
-                i++;
-            }
-            curr = curr->get(topic.substr(start, i - start));
-        }
-        curr->subscribers.insert(connection);
-        /* Only do this if we did not aleady exist */
-        socketToNodeList[connection].push_back(curr);
-    }
-
-    /* Unsubscribe from all subscriptions */
-    void unsubscribeAll(void *connection) {
-
-        for (Node *node : socketToNodeList[connection]) {
-
-            /* Also make sure to update any backpressure here */
-
-            node->subscribers.erase(connection);
-        }
-
-        socketToNodeList.erase(connection);
-    }
-
-    /* WebSocket.publish looks up its tree and publishes to it */
-    void publish(std::string topic, char *data, size_t length) {
-        Node *curr = &topicToNode;
-        for (int i = 0; i < topic.length(); i++) {
-            int start = i;
-            while (topic[i] != '/' && i < topic.length()) {
-                i++;
-            }
-            std::string path(topic.data() + start, i - start);
-
-            // end wildcard consumes traversal
-            auto it = curr->find("#");
-            if (it != curr->end()) {
-                curr = it->second;
-                //matches.push_back(curr);
-                curr->sharedMessage.append(data, length);
-                if (curr->subscribers.size()) {
-                    pubNodes.insert(curr);
-                }
-                break;
-            } else {
-                it = curr->find(path);
-                if (it == curr->end()) {
-                    it = curr->find("+");
-                    if (it != curr->end()) {
-                        goto skip;
-                    }
-                    break;
-                } else {
-    skip:
-                    curr = it->second;
-                    if (i == topic.length()) {
-                        //matches.push_back(curr);
-                        curr->sharedMessage.append(data, length);
-                        if (curr->subscribers.size()) {
-                            pubNodes.insert(curr);
-                        }
-                        break;
-                    }
-                }
-            }
-        }
-    }
-};
-}
-
-#endif // UWS_TOPICTREE_H

src/TopicTreeDraft.h

@@ -1,3 +1,23 @@
+/*
+ * Authored by Alex Hultman, 2018-2019.
+ * Intellectual property of third-party.
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ *     http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef UWS_TOPICTREE_H
+#define UWS_TOPICTREE_H
+
 #include <iostream>
 #include <vector>
 #include <map>
@@ -5,13 +25,16 @@
 #include <functional>
 #include <set>
 #include <chrono>
+#include <list>
 
 namespace uWS {
 
 /* A Subscriber is an extension of a socket */
 struct Subscriber {
-    /* List of all our subscriptions (subscribersNextSubscription) */
+    std::list<struct Topic *> subscriptions;
-    struct Subscription *subscriptions;
+    void *user;
+
+    Subscriber(void *user) : user(user) {}
 };
 
 struct Topic {
@@ -40,24 +63,6 @@ struct Topic {
     std::set<Subscriber *> subs;
 };
 
-/* A Subscription is a link between Topic and Subscriber */
-struct Subscription {
-    /* The Topic we are subscribed to */
-    Topic *topic;
-
-    /* What subscriber we are linked to */
-    Subscriber *subscriber;
-
-    /* Backpressure relating to the Topic */
-    int backpressure;
-
-    /* Next subscription from subscriber point of view */
-    Subscription *subscribersNextSubscription;
-
-    /* Next subscription from topic point of view */
-    Subscription *topicNextSubscription;
-};
-
 struct TopicTree {
 private:
     std::function<int(Subscriber *, std::string_view)> cb;
@@ -194,6 +199,9 @@ public:
 
         /* Add socket to Topic's Set */
         iterator->subs.insert(subscriber);
+
+        /* Add Topic to list of subscriptions */
+        subscriber->subscriptions.push_back(iterator);
     }
 
     void publish(std::string_view topic, std::string_view message) {
@@ -207,89 +215,108 @@ public:
 
     }
 
+    /* Can be called with nullptr, ignore it then */
     void unsubscribeAll(Subscriber *subscriber) {
-        for (Subscription *iterator = subscriber->subscriptions; iterator; iterator = iterator->subscribersNextSubscription) {
+        if (subscriber) {
-            iterator->topic->subs.erase(subscriber);
+            for (Topic *topic : subscriber->subscriptions) {
-            trimTree(iterator->topic);
+                topic->subs.erase(subscriber);
+                trimTree(topic);
+            }
         }
     }
 
     /* Drain the tree by emitting what to send with every Subscriber */
-    void drain(/*std::function<int(Subscriber *, std::string_view)> cb*/) {
+    void drain() {
 
+        /* Do nothing if nothing to send */
         if (!numTriggeredTopics) {
             return;
         }
 
-        /* Up to 64 triggered Topics per batch */
+        /* Fast path for one topic (can also be used with heuristics) */
-        std::map<uint64_t, std::string> intersectionCache;
+        if (numTriggeredTopics == -555555) {
+            /* Disabled */
+            /*std::string res;
+            for (auto &p : triggeredTopics[0]->messages) {
+                res.append(p.second);
+            }
 
-        /* Loop over these here */
+            for (Subscriber *s : triggeredTopics[0]->subs) {
-        std::set<Subscriber *>::iterator it[64];
+                cb(s, res);
-        std::set<Subscriber *>::iterator end[64];
+            }*/
-        for (int i = 0; i < numTriggeredTopics; i++) {
+        } else {
-            it[i] = triggeredTopics[i]->subs.begin();
-            end[i] = triggeredTopics[i]->subs.end();
-        }
-        
-        /* Empty all sets from unique subscribers */
-        for (int nonEmpty = numTriggeredTopics; nonEmpty; ) {
 
-            Subscriber *nextMin = (Subscriber *)UINTPTR_MAX;
+            /* Up to 64 triggered Topics per batch */
-
+            std::map<uint64_t, std::string> intersectionCache;
-            /* The message sets relevant for this intersection */
-            std::map<int, std::string> *perSubscriberIntersectingTopicMessages[64];
-            int numPerSubscriberIntersectingTopicMessages = 0;
-
-            uint64_t intersection = 0;
 
+            /* Loop over these here */
+            std::set<Subscriber *>::iterator it[64];
+            std::set<Subscriber *>::iterator end[64];
             for (int i = 0; i < numTriggeredTopics; i++) {
-                if ((it[i] != end[i]) && (*it[i] == min)) {
+                it[i] = triggeredTopics[i]->subs.begin();
+                end[i] = triggeredTopics[i]->subs.end();
+            }
+            
+            /* Empty all sets from unique subscribers */
+            for (int nonEmpty = numTriggeredTopics; nonEmpty; ) {
 
-                    /* Mark this intersection */
+                Subscriber *nextMin = (Subscriber *)UINTPTR_MAX;
-                    intersection |= (1 << i);
-                    perSubscriberIntersectingTopicMessages[numPerSubscriberIntersectingTopicMessages++] = &triggeredTopics[i]->messages;
 
-                    it[i]++;
+                /* The message sets relevant for this intersection */
-                    if (it[i] == end[i]) {
+                std::map<int, std::string> *perSubscriberIntersectingTopicMessages[64];
-                        nonEmpty--;
+                int numPerSubscriberIntersectingTopicMessages = 0;
+
+                uint64_t intersection = 0;
+
+                for (int i = 0; i < numTriggeredTopics; i++) {
+                    if ((it[i] != end[i]) && (*it[i] == min)) {
+
+                        /* Mark this intersection */
+                        intersection |= (1 << i);
+                        perSubscriberIntersectingTopicMessages[numPerSubscriberIntersectingTopicMessages++] = &triggeredTopics[i]->messages;
+
+                        it[i]++;
+                        if (it[i] == end[i]) {
+                            nonEmpty--;
+                        }
+                        else {
+                            if (nextMin > *it[i]) {
+                                nextMin = *it[i];
+                            }
+                        }
                     }
                     else {
-                        if (nextMin > *it[i]) {
+                        /* We need to lower nextMin to us, in the case of min being the last in a set */
+                        if ((it[i] != end[i]) && (nextMin > *it[i])) {
                             nextMin = *it[i];
                         }
                     }
                 }
-                else {
+
-                    /* We need to lower nextMin to us, in the case of min being the last in a set */
+                /* Generate cache for intersection */
-                    if ((it[i] != end[i]) && (nextMin > *it[i])) {
+                if (intersectionCache[intersection].length() == 0) {
-                        nextMin = *it[i];
+
+                    /* Build the union in order without duplicates */
+                    std::map<int, std::string> complete;
+                    for (int i = 0; i < numPerSubscriberIntersectingTopicMessages; i++) {
+                        complete.insert(perSubscriberIntersectingTopicMessages[i]->begin(), perSubscriberIntersectingTopicMessages[i]->end());
                     }
+
+                    /* Create the linear cache */
+                    std::string res;
+                    for (auto &p : complete) {
+                        res.append(p.second);
+                    }
+
+                    cb(min, intersectionCache[intersection] = std::move(res));
                 }
-            }
+                else {
-
+                    cb(min, intersectionCache[intersection]);
-            /* Generate cache for intersection */
-            if (intersectionCache[intersection].length() == 0) {
-
-                /* Build the union in order without duplicates */
-                std::map<int, std::string> complete;
-                for (int i = 0; i < numPerSubscriberIntersectingTopicMessages; i++) {
-                    complete.insert(perSubscriberIntersectingTopicMessages[i]->begin(), perSubscriberIntersectingTopicMessages[i]->end());
                 }
 
-                /* Create the linear cache */
+                min = nextMin;
-                std::string res;
-                for (auto &p : complete) {
-                    res.append(p.second);
-                }
-
-                cb(min, intersectionCache[intersection] = std::move(res));
-            }
-            else {
-                cb(min, intersectionCache[intersection]);
             }
 
-            min = nextMin;
         }
 
         /* Clear messages of triggered Topics */
@@ -316,4 +343,6 @@ public:
     }
 };
 
-}
+}
+
+#endif

src/WebSocket.h

@@ -128,7 +128,7 @@ public:
         }
 
         /* Make sure to unsubscribe from any pub/sub node at exit */
-        webSocketContextData->topicTree.unsubscribeAll((Subscriber *) this);
+        webSocketContextData->topicTree.unsubscribeAll(webSocketData->subscriber);
     }
 
     /* Subscribe to a topic according to MQTT rules and syntax */
@@ -137,8 +137,13 @@ public:
             (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this)
         );
 
-        /* Fix this up */
+        /* Make us a subscriber if we aren't yet */
-        webSocketContextData->topicTree.subscribe(topic, (Subscriber *) this);
+        WebSocketData *webSocketData = (WebSocketData *) us_socket_ext(SSL, (us_socket_t *) this);
+        if (!webSocketData->subscriber) {
+            webSocketData->subscriber = new Subscriber(this);
+        }
+
+        webSocketContextData->topicTree.subscribe(topic, webSocketData->subscriber);
     }
 
     /* Publish a message to a topic according to MQTT rules and syntax */

src/WebSocketContext.h

@@ -237,7 +237,7 @@ private:
                 }
 
                 /* Make sure to unsubscribe from any pub/sub node at exit */
-                webSocketContextData->topicTree.unsubscribeAll((Subscriber *) s);
+                webSocketContextData->topicTree.unsubscribeAll(webSocketData->subscriber);
             }
 
             /* Destruct in-placed data struct */

src/WebSocketContextData.h

@@ -48,7 +48,7 @@ struct WebSocketContextData {
         //std::cout << "Skickar data: " << data << " på sub: " << s << std::endl;
 
 
-        auto *asyncSocket = (AsyncSocket<SSL> *) s;
+        auto *asyncSocket = (AsyncSocket<SSL> *) s->user;
 
         asyncSocket->write(data.data(), data.length());
 

src/WebSocketData.h

@@ -41,6 +41,9 @@ private:
 
     /* We might have a dedicated compressor */
     DeflationStream *deflationStream = nullptr;
+
+    /* We could be a subscriber */
+    Subscriber *subscriber = nullptr;
 public:
     WebSocketData(bool perMessageDeflate, bool slidingCompression, std::string &&backpressure) : AsyncSocketData<false>(std::move(backpressure)), WebSocketState<true>() {
         compressionStatus = perMessageDeflate ? ENABLED : DISABLED;
@@ -55,6 +58,10 @@ public:
         if (deflationStream) {
             delete deflationStream;
         }
+
+        if (subscriber) {
+            delete subscriber;
+        }
     }
 };