Initial sender hole experiment

src/TopicTree.h

@@ -28,6 +28,16 @@
 #include <list>
 #include <cstring>
 
+struct Hole {
+    std::pair<size_t, size_t> lengths;
+    unsigned int messageId;
+};
+
+struct Intersection {
+    std::pair<std::string, std::string> dataChannels;
+    std::vector<Hole> holes;
+};
+
 namespace uWS {
 
 /* A Subscriber is an extension of a socket */
@@ -65,8 +75,12 @@ struct Topic {
 };
 
 struct TopicTree {
+
+    /* Sender holes */
+    std::map<Subscriber *, std::vector<unsigned int>> senderHoles;
+
 private:
-    std::function<int(Subscriber *, std::pair<std::string_view, std::string_view>)> cb;
+    std::function<int(Subscriber *, Intersection &)> cb;
 
     Topic *root = new Topic;
 
@@ -187,7 +201,7 @@ private:
 
 public:
 
-    TopicTree(std::function<int(Subscriber *, std::pair<std::string_view, std::string_view>)> cb) {
+    TopicTree(std::function<int(Subscriber *, Intersection &)> cb) {
         this->cb = cb;
     }
 
@@ -251,7 +265,13 @@ public:
         }
     }
 
-    void publish(std::string_view topic, std::pair<std::string_view, std::string_view> message) {
+    void publish(std::string_view topic, std::pair<std::string_view, std::string_view> message, Subscriber *sender = nullptr) {
+
+        /* Add a hole for the sender if one */
+        if (sender) {
+            senderHoles[sender].push_back(messageId);
+        }
+
         publish(root, 0, 0, topic, message);
         messageId++;
     }
@@ -340,6 +360,7 @@ public:
         numTriggeredTopics = numFilteredTriggeredTopics;
 
         if (!numTriggeredTopics) {
+            senderHoles.clear();
             return;
         }
 
@@ -355,7 +376,7 @@ public:
         if (min != (Subscriber *)UINTPTR_MAX) {
 
             /* Up to 64 triggered Topics per batch */
-            std::map<uint64_t, std::pair<std::string, std::string>> intersectionCache;
+            std::map<uint64_t, /*std::pair<std::string, std::string>*/ Intersection> intersectionCache;
 
             /* Loop over these here */
             std::set<Subscriber *>::iterator it[64];
@@ -402,7 +423,7 @@ public:
                 }
 
                 /* Generate cache for intersection */
-                if (intersectionCache[intersection].first.length() == 0) {
+                if (intersectionCache[intersection].dataChannels.first.length() == 0) {
 
                     /* Build the union in order without duplicates */
                     std::map<unsigned int, std::pair<std::string, std::string>> complete;
@@ -411,15 +432,39 @@ public:
                     }
 
                     /* Create the linear cache, {inflated, deflated} */
-                    std::pair<std::string, std::string> res;
+                    /*std::pair<std::string, std::string>*/ Intersection res;
+                    //std::string messageIds; // sorterade id:n för meddelanden
+
+                    //std::vector<
+
+
                     for (auto &p : complete) {
-                        res.first.append(p.second.first);
-                        res.second.append(p.second.second);
+                        printf("messageId = %d\n", p.first);
+
+
+                        res.dataChannels.first.append(p.second.first);
+                        res.dataChannels.second.append(p.second.second);
+
+                        // appenda {id, längd, längd}
+                        Hole h;
+                        h.lengths.first = p.second.first.length();
+                        h.lengths.second = p.second.second.length();
+                        h.messageId = p.first;
+                        res.holes.push_back(h);
                     }
 
+                    //can we know the messageId here and lookup if "min" is the sender?
+
                     cb(min, intersectionCache[intersection] = std::move(res));
                 }
                 else {
+
+                    // vi kan göra en cache som håller inflated, deflated, messageIds
+
+                    // sen, för varje subscriber, kollar vi upp en vektor av messageIds - senderHoles
+
+                    // sen måste vi loopa över
+
                     cb(min, intersectionCache[intersection]);
                 }
 
@@ -434,6 +479,7 @@ public:
             triggeredTopics[i]->triggered = false;
         }
         numTriggeredTopics = 0;
+        senderHoles.clear();
     }
 };
 

tests/TopicTree.cpp

@@ -9,14 +9,14 @@ void testUnsubscribeInside() {
     uWS::TopicTree *topicTree;
     std::map<void *, std::pair<std::string, std::string>> expectedResult;
 
-    topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::pair<std::string_view, std::string_view> dataChannels) {
-        std::string_view data = dataChannels.second;
+    topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, Intersection &intersection) {
+        std::string_view data = intersection.dataChannels.second;
 
         /* Check for unexpected subscribers */
         assert(expectedResult.find(s) != expectedResult.end());
 
         /* Check for unexpected data */
-        assert(expectedResult[s].first == dataChannels.first && expectedResult[s].second == dataChannels.second);
+        assert(expectedResult[s].first == intersection.dataChannels.first && expectedResult[s].second == intersection.dataChannels.second);
 
         /* This one causes mess-up */
         topicTree->unsubscribeAll(s);
@@ -67,13 +67,40 @@ void testPublisherHoles() {
     uWS::TopicTree *topicTree;
     std::map<void *, std::pair<std::string, std::string>> expectedResult;
 
-    topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::pair<std::string_view, std::string_view> dataChannels) {
+    topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, /*std::pair<std::string, std::string> &dataChannels*/ Intersection &intersection) {
+
+
+        std::cout << "Subscriber: " << s << std::endl;
+
+        std::vector<unsigned int> &senderForMessages = topicTree->senderHoles[s];
+
+        for (unsigned int id : senderForMessages) {
+            std::cout << "We are sender for id: " << id << std::endl;
+        }
+
+        // iterate holes
+        for (Hole h : intersection.holes) {
+            std::cout << h.messageId << std::endl;
+
+
+
+            // todo: this linear search is not needed as ids are ordered!
+            if (std::find(senderForMessages.begin(), senderForMessages.end(), h.messageId) != senderForMessages.end()) {
+                std::cout << "WE ARE SENDER FOR THIS MESSAGE!" << std::endl;
+            }
+
+
+
+        }
+
+
+
 
         /* Check for unexpected subscribers */
         assert(expectedResult.find(s) != expectedResult.end());
 
         /* Check for unexpected data */
-        assert(expectedResult[s].first == dataChannels.first && expectedResult[s].second == dataChannels.second);
+        assert(expectedResult[s].first == intersection.dataChannels.first && expectedResult[s].second == intersection.dataChannels.second);
 
         /* We actually don't use this one */
         return 0;
@@ -84,8 +111,8 @@ void testPublisherHoles() {
 
     /* Fill out expectedResult */
     expectedResult = {
-        {s1, {"Två!", "Två!"}}, // todo: this one should not receive what he sent himself
-        {s2, {"Två!", "Två!"}}
+        {s1, {"Två!Två!", "Två!Två!"}}, // todo: this one should not receive what he sent himself
+        {s2, {"Två!Två!", "Två!Två!"}}
     };
 
     /* Make sure s1 < s2 */
@@ -101,7 +128,8 @@ void testPublisherHoles() {
 
     /* This order matters, as it fills triggeredTopics array in order */
     //topicTree->publish("1", {std::string_view("Ett!"), std::string_view("Ett!")});
-    topicTree->publish("1", {std::string_view("Två!"), std::string_view("Två!")});
+    topicTree->publish("1", {std::string_view("Två!"), std::string_view("Två!")}, s1);
+    topicTree->publish("1", {std::string_view("Två!"), std::string_view("Två!")}, s1);
 
     topicTree->drain();