Hook up new pubsub

src/AsyncSocket.h

@@ -31,6 +31,7 @@ template <bool SSL>
 struct AsyncSocket {
     template <bool> friend struct HttpContext;
     template <bool, bool> friend struct WebSocketContext;
+    template <bool> friend struct WebSocketContextData;
     friend struct TopicTree;
 
 protected:

src/TopicTreeDraft.h

@@ -6,6 +6,8 @@
 #include <set>
 #include <chrono>
 
+namespace uWS {
+
 /* A Subscriber is an extension of a socket */
 struct Subscriber {
     /* List of all our subscriptions (subscribersNextSubscription) */
@@ -215,6 +217,10 @@ public:
     /* Drain the tree by emitting what to send with every Subscriber */
     void drain(/*std::function<int(Subscriber *, std::string_view)> cb*/) {
 
+        if (!numTriggeredTopics) {
+            return;
+        }
+
         /* Up to 64 triggered Topics per batch */
         std::map<uint64_t, std::string> intersectionCache;
 
@@ -285,6 +291,13 @@ public:
 
             min = nextMin;
         }
+
+        /* Clear messages of triggered Topics */
+        for (int i = 0; i < numTriggeredTopics; i++) {
+            triggeredTopics[i]->messages.clear();
+            triggeredTopics[i]->triggered = false;
+        }
+        numTriggeredTopics = 0;
     }
 
     void print(Topic *root = nullptr, int indentation = 1) {
@@ -302,3 +315,5 @@ public:
         }
     }
 };
+
+}

src/WebSocket.h

@@ -128,7 +128,7 @@ public:
         }
 
         /* Make sure to unsubscribe from any pub/sub node at exit */
-        webSocketContextData->topicTree.unsubscribeAll(this);
+        webSocketContextData->topicTree.unsubscribeAll((Subscriber *) this);
     }
 
     /* Subscribe to a topic according to MQTT rules and syntax */
@@ -138,9 +138,7 @@ public:
         );
 
         /* Fix this up */
-        bool *valid = new bool;
+        webSocketContextData->topicTree.subscribe(topic, (Subscriber *) this);
-        *valid = true;
-        webSocketContextData->topicTree.subscribe(std::string(topic), this, valid);
     }
 
     /* Publish a message to a topic according to MQTT rules and syntax */
@@ -153,7 +151,7 @@ public:
         char dst[1024];
         size_t dst_length = protocol::formatMessage<true>(dst, message.data(), message.length(), OpCode::TEXT, message.length(), false);
 
-        webSocketContextData->topicTree.publish(std::string(topic), dst, dst_length);
+        webSocketContextData->topicTree.publish(topic, std::string_view(dst, dst_length));
     }
 };
 

src/WebSocketContext.h

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

src/WebSocketContextData.h

@@ -22,7 +22,7 @@
 #include <string_view>
 
 #include "WebSocketProtocol.h"
-#include "TopicTree.h"
+#include "TopicTreeDraft.h"
 
 namespace uWS {
 
@@ -43,6 +43,23 @@ struct WebSocketContextData {
 
     /* Each websocket context has a topic tree for pub/sub */
     TopicTree topicTree;
+
+    WebSocketContextData() : topicTree([](Subscriber *s, std::string_view data) -> int {
+        //std::cout << "Skickar data: " << data << " på sub: " << s << std::endl;
+
+
+        auto *asyncSocket = (AsyncSocket<SSL> *) s;
+
+        asyncSocket->write(data.data(), data.length());
+
+        return 0;
+    }) {
+
+        Loop::get()->addPostHandler([this](Loop *loop) {
+
+            topicTree.drain();
+        });
+    }
 };
 
 }