Store subscribers as map

The initial concept of iterating over a vector being fast didn't help, and made having many subscribers to one topic very slow.

Store subscribers as map
The initial concept of iterating over a vector being fast didn't help, and made having many subscribers to one topic very slow.
Wiebe Cazemier
1 parent 84fe398d
Showing 2 changed files with 15 additions and 22 deletions
subscriptionstore.cpp
subscriptionstore.h
@@ -28,7 +28,7 @@ SubscriptionNode::SubscriptionNode(const std::string &amp;subtopic) :
 }
-std::vector<Subscription> &SubscriptionNode::getSubscribers()
+std::unordered_map<std::string, Subscription> &SubscriptionNode::getSubscribers()
 {
     return subscribers;
 }
@@ -44,18 +44,8 @@ void SubscriptionNode::addSubscriber(const std::shared_ptr&lt;Session&gt; &amp;subscriber,
     sub.session = subscriber;
     sub.qos = qos;
-    // I'll have to decide whether to keep the subscriber as a vector. Vectors are
-    // fast, and relatively, you don't often add subscribers.
-    auto subscriber_it = std::find(subscribers.begin(), subscribers.end(), sub);
-    if (subscriber_it == subscribers.end())
-    {
-        subscribers.push_back(sub);
-    }
-    else
-    {
-        Subscription &existingSub = *subscriber_it;
-        existingSub = sub;
-    }
+    const std::string &client_id = subscriber->getClientId();
+    subscribers[client_id] = sub;
 }
 void SubscriptionNode::removeSubscriber(const std::shared_ptr<Session> &subscriber)
@@ -64,7 +54,7 @@ void SubscriptionNode::removeSubscriber(const std::shared_ptr&lt;Session&gt; &amp;subscrib
     sub.session = subscriber;
     sub.qos = 0;
-    auto it = std::find(subscribers.begin(), subscribers.end(), sub);
+    auto it = subscribers.find(subscriber->getClientId());
     if (it != subscribers.end())
     {
@@ -261,10 +251,12 @@ bool SubscriptionStore::sessionPresent(const std::string &amp;clientid)
     return result;
 }
-void SubscriptionStore::publishNonRecursively(const MqttPacket &packet, const std::vector<Subscription> &subscribers, uint64_t &count) const
+void SubscriptionStore::publishNonRecursively(const MqttPacket &packet, const std::unordered_map<std::string, Subscription> &subscribers, uint64_t &count) const
 {
-    for (const Subscription &sub : subscribers)
+    for (auto &pair : subscribers)
     {
+        const Subscription &sub = pair.second;
+
         const std::shared_ptr<Session> session = sub.session.lock();
         if (session) // Shared pointer expires when session has been cleaned by 'clean session' connect.
         {
@@ -479,10 +471,10 @@ int SubscriptionNode::cleanSubscriptions()
     auto it = subscribers.begin();
     while (it != subscribers.end())
     {
-        std::shared_ptr<Session> ses = it->session.lock();
+        std::shared_ptr<Session> ses = it->second.session.lock();
         if (!ses)
         {
-            Logger::getInstance()->logf(LOG_DEBUG, "Removing empty spot in subscribers vector");
+            Logger::getInstance()->logf(LOG_DEBUG, "Removing empty spot in subscribers map");
             it = subscribers.erase(it);
         }
         else
@@ -567,8 +559,9 @@ void SubscriptionStore::getRetainedMessages(RetainedMessageNode *this_node, std:
 void SubscriptionStore::getSubscriptions(SubscriptionNode *this_node, const std::string &composedTopic, bool root,
                                          std::unordered_map<std::string, std::list<SubscriptionForSerializing>> &outputList) const
 {
-    for (const Subscription &node : this_node->getSubscribers())
+    for (auto &pair : this_node->getSubscribers())
     {
+        const Subscription &node = pair.second;
         std::shared_ptr<Session> ses = node.session.lock();
         if (ses)
         {
@@ -44,14 +44,14 @@ struct Subscription
 class SubscriptionNode
 {
     std::string subtopic;
-    std::vector<Subscription> subscribers;
+    std::unordered_map<std::string, Subscription> subscribers;
 public:
     SubscriptionNode(const std::string &subtopic);
     SubscriptionNode(const SubscriptionNode &node) = delete;
     SubscriptionNode(SubscriptionNode &&node) = delete;
-    std::vector<Subscription> &getSubscribers();
+    std::unordered_map<std::string, Subscription> &getSubscribers();
     const std::string &getSubtopic() const;
     void addSubscriber(const std::shared_ptr<Session> &subscriber, char qos);
     void removeSubscriber(const std::shared_ptr<Session> &subscriber);
@@ -94,7 +94,7 @@ class SubscriptionStore
     Logger *logger = Logger::getInstance();
-    void publishNonRecursively(const MqttPacket &packet, const std::vector<Subscription> &subscribers, uint64_t &count) const;
+    void publishNonRecursively(const MqttPacket &packet, const std::unordered_map<std::string, Subscription> &subscribers, uint64_t &count) const;
     void publishRecursively(std::vector<std::string>::const_iterator cur_subtopic_it, std::vector<std::string>::const_iterator end,
                             SubscriptionNode *this_node, const MqttPacket &packet, uint64_t &count) const;
     void getRetainedMessages(RetainedMessageNode *this_node, std::vector<RetainedMessage> &outputList) const;