Initial pubsub preparatory work

src/Loop.h

@@ -21,12 +21,8 @@
 /* The loop is lazily created per-thread and run with uWS::run() */
 
 #include "LoopData.h"
-
 #include <libusockets_new.h>
 
-
-
-
 #include <iostream>
 
 namespace uWS {
@@ -62,6 +58,11 @@ private:
         if (loopData->postHandler) {
             loopData->postHandler((Loop *) loop);
         }
+
+        /* We should move over to using only these */
+        for (auto &f : loopData->postHandlers) {
+            f((Loop *) loop);
+        }
     }
 
     Loop() = delete;
@@ -109,6 +110,13 @@ public:
         us_loop_free((us_loop *) this);
     }
 
+    /* We want to have multiple of these */
+    void addPostHandler(fu2::unique_function<void(Loop *)> &&handler) {
+        LoopData *loopData = (LoopData *) us_loop_ext((us_loop *) this);
+
+        loopData->postHandlers.emplace_back(std::move(handler));
+    }
+
     /* Set postCb callback */
     void setPostHandler(fu2::unique_function<void(Loop *)> &&handler) {
         LoopData *loopData = (LoopData *) us_loop_ext((us_loop *) this);

src/LoopData.h

@@ -40,6 +40,9 @@ private:
 
     fu2::unique_function<void(Loop *)> postHandler, preHandler;
 
+    /* Move over to these later on */
+    std::vector<fu2::unique_function<void(Loop *)>> postHandlers;
+
 public:
     ~LoopData() {
         /* If we have had App.ws called with compression we need to clear this */

src/TopicTree.h

@@ -15,7 +15,9 @@
  * limitations under the License.
  */
 
-/* Every Loop holds one TopicTree */
+/* Every WebSocketContext holds one TopicTree */
+#include "Loop.h"
+#include "WebSocket.h"
 
 #ifndef TOPICTREE_H
 #define TOPICTREE_H
@@ -25,6 +27,8 @@
 #include <vector>
 #include <set>
 
+namespace uWS {
+
 class TopicTree {
 private:
     struct Node : std::map<std::string, Node *> {
@@ -46,7 +50,42 @@ private:
 public:
 
     TopicTree() {
-        std::cout << "Constructing TopicTree" << std::endl;
+        /* Dynamically hook us up with the Loop post handler */
+        Loop::defaultLoop()->addPostHandler([this](Loop *loop) {
+
+            if (!pubNodes.size()) {
+                return;
+            }
+
+            // messages need to be prepared twice: compressed and non compressed
+            // if using dedicated compression, don't prepare
+
+            // user should be something like a std::string with formatted content
+            std::string *preparedMessage = new std::string;
+
+            // prepare, send, ref, user
+            drain([](void *user, char *sharedMessage, size_t sharedMessageLength) {
+
+                //std::cout << "Preparing " << std::string_view(sharedMessage, sharedMessageLength) << std::endl;
+
+                std::string *preparedMessage = (std::string *) user;
+                preparedMessage->append(sharedMessage, sharedMessageLength);
+
+            }, [](void *user, void *ws) {
+
+                /* This would be where we send the preformatted pre-compessed message in user */
+
+                //std::cout << "Sending " << std::endl;
+
+                WebSocket<false, true> *webSocket = (WebSocket<false, true> *) ws; // assumes non-SSL
+
+                std::string *preparedMessage = (std::string *) user;
+                webSocket->send(*preparedMessage, OpCode::TEXT, false);
+
+            }, [](void *ws) {
+                std::cout << "Refing" << std::endl;
+            }, preparedMessage);
+        });
     }
 
     /* WebSocket.subscribe will lookup the Loop and subscribe in its tree */
@@ -116,6 +155,9 @@ public:
 
     /* I forgot what this does but probably needs lots of changes anyways */
     void drain(void (*prepareCb)(void *user, char *, size_t), void (*sendCb)(void *, void *), void (*refCb)(void *), void *user) {
+
+        std::cout << "pubNodes: " << pubNodes.size() << std::endl;
+
         if (pubNodes.size()) {
 
             //if(pubNodes.size() > 1) {
@@ -128,6 +170,8 @@ public:
                 }
             //}
 
+            std::cout << "Now we are here" << std::endl;
+
             for (Node *topicNode : pubNodes) {
                 prepareCb(user, (char *) topicNode->sharedMessage.data(), topicNode->sharedMessage.length());
                 for (auto it = topicNode->subscribers.begin(); it != topicNode->subscribers.end(); ) {
@@ -145,6 +189,6 @@ public:
         }
     }
 };
-
+}
 
 #endif // TOPICTREE_H

src/WebSocket.h

@@ -133,7 +133,9 @@ public:
         );
 
         /* Fix this up */
-        webSocketContextData->topicTree.subscribe(topic, this, nullptr);
+        bool *valid = new bool;
+        *valid = true;
+        webSocketContextData->topicTree.subscribe(std::string(topic), this, valid);
     }
 
     /* Publish a message to a topic according to MQTT rules and syntax */
@@ -142,7 +144,7 @@ public:
             (us_new_socket_context_t *) us_new_socket_context(SSL, (us_new_socket_t *) this)
         );
 
-        webSocketContextData->topicTree.publish(topic, message.data(), message.length());
+        webSocketContextData->topicTree.publish(std::string(topic), (char *) message.data(), message.length());
     }
 };