Some changes to pubsub

src/AsyncSocket.h

@@ -30,6 +30,7 @@ template <bool SSL>
 struct AsyncSocket {
     template <bool> friend struct HttpContext;
     template <bool, bool> friend struct WebSocketContext;
+    friend class TopicTree;
 protected:
 
     /* Get loop data for socket */

src/Loop.h

@@ -50,19 +50,24 @@ private:
         if (loopData->preHandler) {
             loopData->preHandler((Loop *) loop);
         }
+
+        /* trying this one here */
+        for (auto &f : loopData->postHandlers) {
+            f((Loop *) loop);
+        }
     }
 
     static void postCb(us_loop *loop) {
         LoopData *loopData = (LoopData *) us_loop_ext(loop);
 
-        if (loopData->postHandler) {
-            loopData->postHandler((Loop *) loop);
-        }
-
         /* We should move over to using only these */
         for (auto &f : loopData->postHandlers) {
             f((Loop *) loop);
         }
+
+        if (loopData->postHandler) {
+            loopData->postHandler((Loop *) loop);
+        }
     }
 
     Loop() = delete;

src/TopicTree.h

@@ -17,7 +17,7 @@
 
 /* Every WebSocketContext holds one TopicTree */
 #include "Loop.h"
-#include "WebSocket.h"
+#include "AsyncSocket.h"
 
 #ifndef TOPICTREE_H
 #define TOPICTREE_H
@@ -29,6 +29,9 @@
 
 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
+
 class TopicTree {
 private:
     struct Node : std::map<std::string, Node *> {
@@ -40,13 +43,24 @@ private:
                 return p.first->second;
             }
         }
+        /* Every subscriber should hold some backpressure cursor */
         std::vector<std::pair<void *, bool *>> subscribers;
         std::string sharedMessage;
-    };
 
-    Node *root = new Node;
+        /* We need backpressure stored */
+        /* vector */
+        std::string backpressure;
+    } *root = new Node;
+
+    /* 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() {
@@ -57,34 +71,32 @@ public:
                 return;
             }
 
-            // messages need to be prepared twice: compressed and non compressed
+            /* We say that all senders get their own message as well, for now being */
-            // if using dedicated compression, don't prepare
 
-            // user should be something like a std::string with formatted content
+            for (Node *topicNode : pubNodes) {
-            std::string *preparedMessage = new std::string;
+                for (auto [ws, valid] : topicNode->subscribers) {
+                    AsyncSocket<false> *asyncSocket = (AsyncSocket<false> *) ws; // assumes non-SSL for now
 
-            // prepare, send, ref, user
+                    /* Writing optionally raw data */
-            drain([](void *user, char *sharedMessage, size_t sharedMessageLength) {
+                    auto [written, failed] = asyncSocket->write(topicNode->sharedMessage.data(), topicNode->sharedMessage.length(), true, 0);
 
-                //std::cout << "Preparing " << std::string_view(sharedMessage, sharedMessageLength) << std::endl;
 
-                std::string *preparedMessage = (std::string *) user;
+                    /* Every subscriber to a topicNode will have int backpressure cursor to this room */
-                preparedMessage->append(sharedMessage, sharedMessageLength);
 
-            }, [](void *user, void *ws) {
+                    /* 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) */
 
-                /* This would be where we send the preformatted pre-compessed message in user */
+                    /* 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 */
 
-                //std::cout << "Sending " << std::endl;
+                    /* It is also possible to move topictree backpressure to the websockets themselves, if only one */
+                }
 
-                WebSocket<false, true> *webSocket = (WebSocket<false, true> *) ws; // assumes non-SSL
+                /* If not all sockets managed to send this message, move it to backpressure */
+                topicNode->sharedMessage.clear();
+            }
+            pubNodes.clear();
 
-                std::string *preparedMessage = (std::string *) user;
-                webSocket->send(*preparedMessage, OpCode::TEXT, false);
 
-            }, [](void *ws) {
-                std::cout << "Refing" << std::endl;
-            }, preparedMessage);
         });
     }
 
@@ -99,8 +111,6 @@ public:
             curr = curr->get(topic.substr(start, i - start));
         }
         curr->subscribers.push_back({connection, valid});
-
-        std::cout << "Subscribed to topicTree" << std::endl;
     }
 
     /* WebSocket.publish looks up its tree and publishes to it */
@@ -145,48 +155,6 @@ public:
                 }
             }
         }
-
-        std::cout << "Published to topicTree" << std::endl;
-    }
-
-    void reset() {
-        root = new Node;
-    }
-
-    /* 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) {
-                for (Node *topicNode : pubNodes) {
-                    for (std::pair<void *, bool *> p : topicNode->subscribers) {
-                        if (*p.second) {
-                            refCb(p.first);
-                        }
-                    }
-                }
-            //}
-
-            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(); ) {
-                    if (!*it->second) {
-                        it = topicNode->subscribers.erase(it);
-                    } else {
-                        sendCb(user, it->first);
-                        it++;
-                    }
-                }
-
-                topicNode->sharedMessage.clear();
-            }
-            pubNodes.clear();
-        }
     }
 };
 }

src/WebSocket.h

@@ -144,7 +144,11 @@ public:
             (us_new_socket_context_t *) us_new_socket_context(SSL, (us_new_socket_t *) this)
         );
 
-        webSocketContextData->topicTree.publish(std::string(topic), (char *) message.data(), message.length());
+        /* We frame the message right here and only pass raw bytes to the pub/subber */
+        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);
     }
 };