diff --git a/src/TopicTree.h b/src/TopicTree.h index 43f3db5..20250e8 100644 --- a/src/TopicTree.h +++ b/src/TopicTree.h @@ -28,6 +28,16 @@ #include #include +struct Hole { + std::pair lengths; + unsigned int messageId; +}; + +struct Intersection { + std::pair dataChannels; + std::vector holes; +}; + namespace uWS { /* A Subscriber is an extension of a socket */ @@ -65,8 +75,12 @@ struct Topic { }; struct TopicTree { + + /* Sender holes */ + std::map> senderHoles; + private: - std::function)> cb; + std::function cb; Topic *root = new Topic; @@ -187,7 +201,7 @@ private: public: - TopicTree(std::function)> cb) { + TopicTree(std::function cb) { this->cb = cb; } @@ -251,7 +265,13 @@ public: } } - void publish(std::string_view topic, std::pair message) { + void publish(std::string_view topic, std::pair 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> intersectionCache; + std::map*/ Intersection> intersectionCache; /* Loop over these here */ std::set::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> complete; @@ -411,15 +432,39 @@ public: } /* Create the linear cache, {inflated, deflated} */ - std::pair res; + /*std::pair*/ 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(); } }; diff --git a/tests/TopicTree.cpp b/tests/TopicTree.cpp index db974d4..61c3e69 100644 --- a/tests/TopicTree.cpp +++ b/tests/TopicTree.cpp @@ -9,14 +9,14 @@ void testUnsubscribeInside() { uWS::TopicTree *topicTree; std::map> expectedResult; - topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::pair 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> expectedResult; - topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, std::pair dataChannels) { + topicTree = new uWS::TopicTree([&topicTree, &expectedResult](uWS::Subscriber *s, /*std::pair &dataChannels*/ Intersection &intersection) { + + + std::cout << "Subscriber: " << s << std::endl; + + std::vector &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();