Move message counting to the end of the call stack

This makes much more sense than returning the amount of messages sent all the way up the call stack.

Move message counting to the end of the call stack
This makes much more sense than returning the amount of messages sent all the way up the call stack.
Wiebe Cazemier
1 parent 44de6670
Showing 7 changed files with 33 additions and 51 deletions
FlashMQTests/tst_maintests.cpp
client.cpp
client.h
session.cpp
session.h
subscriptionstore.cpp
subscriptionstore.h
@@ -1033,15 +1033,13 @@ void MainTests::testSavingSessions()
         splitTopic(topic4, subtopics);
         store->addSubscription(c2, topic4, subtopics, 0);
  
-        uint64_t count = 0;
-
         Publish publish("a/b/c", "Hello Barry", 1);
  
         std::shared_ptr<Session> c1ses = c1->getSession();
         c1.reset();
         MqttPacket publishPacket(ProtocolVersion::Mqtt311, publish);
         PublishCopyFactory fac(&publishPacket);
-        c1ses->writePacket(fac, 1, count);
+        c1ses->writePacket(fac, 1);
  
         store->saveSessionsAndSubscriptions("/tmp/flashmqtests_sessions.db");
  
@@ -191,7 +191,7 @@ void Client::writeText(const std::string &amp;text)
     setReadyForWriting(true);
 }
  
-int Client::writeMqttPacket(const MqttPacket &packet)
+void Client::writeMqttPacket(const MqttPacket &packet)
 {
     const size_t packetSize = packet.getSizeIncludingNonPresentHeader();
  
@@ -199,7 +199,7 @@ int Client::writeMqttPacket(const MqttPacket &amp;packet)
     // sending that Application Message [MQTT-3.1.2-25]."
     if (packetSize > this->maxOutgoingPacketSize)
     {
-        return 0;
+        return;
     }
  
     std::lock_guard<std::mutex> locker(writeBufMutex);
@@ -215,19 +215,23 @@ int Client::writeMqttPacket(const MqttPacket &amp;packet)
     // QoS packet are queued and limited elsewhere.
     if (packet.packetType == PacketType::PUBLISH && packet.getQos() == 0 && packetSize > writebuf.freeSpace())
     {
-        return 0;
+        return;
     }
  
     packet.readIntoBuf(writebuf);
  
-    if (packet.packetType == PacketType::DISCONNECT)
+    if (packet.packetType == PacketType::PUBLISH)
+    {
+        ThreadData *td = ThreadGlobals::getThreadData();
+        td->incrementSentMessageCount(1);
+    }
+    else if (packet.packetType == PacketType::DISCONNECT)
         setReadyForDisconnect();
  
     setReadyForWriting(true);
-    return 1;
 }
  
-int Client::writeMqttPacketAndBlameThisClient(PublishCopyFactory &copyFactory, char max_qos, uint16_t packet_id)
+void Client::writeMqttPacketAndBlameThisClient(PublishCopyFactory &copyFactory, char max_qos, uint16_t packet_id)
 {
     uint16_t topic_alias = 0;
     bool skip_topic = false;
@@ -256,22 +260,20 @@ int Client::writeMqttPacketAndBlameThisClient(PublishCopyFactory &amp;copyFactory, c
         p->setQos(max_qos);
     }
  
-    return writeMqttPacketAndBlameThisClient(*p);
+    writeMqttPacketAndBlameThisClient(*p);
 }
  
 // Helper method to avoid the exception ending up at the sender of messages, which would then get disconnected.
-int Client::writeMqttPacketAndBlameThisClient(const MqttPacket &packet)
+void Client::writeMqttPacketAndBlameThisClient(const MqttPacket &packet)
 {
     try
     {
-        return this->writeMqttPacket(packet);
+        this->writeMqttPacket(packet);
     }
     catch (std::exception &ex)
     {
         threadData->removeClientQueued(fd);
     }
-
-    return 0;
 }
  
 // Ping responses are always the same, so hardcoding it for optimization.
@@ -151,9 +151,9 @@ public:
  
     void writeText(const std::string &text);
     void writePingResp();
-    int writeMqttPacket(const MqttPacket &packet);
-    int writeMqttPacketAndBlameThisClient(PublishCopyFactory &copyFactory, char max_qos, uint16_t packet_id);
-    int writeMqttPacketAndBlameThisClient(const MqttPacket &packet);
+    void writeMqttPacket(const MqttPacket &packet);
+    void writeMqttPacketAndBlameThisClient(PublishCopyFactory &copyFactory, char max_qos, uint16_t packet_id);
+    void writeMqttPacketAndBlameThisClient(const MqttPacket &packet);
     bool writeBufIntoFd();
     bool isBeingDisconnected() const { return disconnectWhenBytesWritten; }
     bool readyForDisconnecting() const { return disconnectWhenBytesWritten && writebuf.usedBytes() == 0; }
@@ -130,7 +130,7 @@ void Session::assignActiveConnection(std::shared_ptr&lt;Client&gt; &amp;client)
  * @param retain. Keep MQTT-3.3.1-9 in mind: existing subscribers don't get retain=1 on packets.
  * @param count. Reference value is updated. It's for statistics.
  */
-void Session::writePacket(PublishCopyFactory &copyFactory, const char max_qos, uint64_t &count)
+void Session::writePacket(PublishCopyFactory &copyFactory, const char max_qos)
 {
     assert(max_qos <= 2);
  
@@ -148,7 +148,7 @@ void Session::writePacket(PublishCopyFactory &amp;copyFactory, const char max_qos, u
         {
             if (c)
             {
-                count += c->writeMqttPacketAndBlameThisClient(copyFactory, effectiveQos, 0);
+                c->writeMqttPacketAndBlameThisClient(copyFactory, effectiveQos, 0);
             }
         }
         else if (effectiveQos > 0)
@@ -174,7 +174,7 @@ void Session::writePacket(PublishCopyFactory &amp;copyFactory, const char max_qos, u
  
             if (c)
             {
-                count += c->writeMqttPacketAndBlameThisClient(copyFactory, effectiveQos, nextPacketId);
+                c->writeMqttPacketAndBlameThisClient(copyFactory, effectiveQos, nextPacketId);
             }
         }
     }
@@ -212,7 +212,6 @@ bool Session::clearQosMessage(uint16_t packet_id, bool qosHandshakeEnds)
  
 /**
  * @brief Session::sendAllPendingQosData sends pending publishes and QoS2 control packets.
- * @return the amount of messages/packets published.
  *
  * [MQTT-4.4.0-1] (about MQTT 3.1.1): "When a Client reconnects with CleanSession set to 0, both the Client and Server MUST
  * re-send any unacknowledged PUBLISH Packets (where QoS > 0) and PUBREL Packets using their original Packet Identifiers. This
@@ -225,10 +224,8 @@ bool Session::clearQosMessage(uint16_t packet_id, bool qosHandshakeEnds)
  * never know that, because IT will have received the PUBACK from FlashMQ. The QoS system is not between publisher
  * and subscriber. Users are required to implement something themselves.
  */
-uint64_t Session::sendAllPendingQosData()
+void Session::sendAllPendingQosData()
 {
-    uint64_t count = 0;
-
     std::shared_ptr<Client> c = makeSharedClient();
     if (c)
     {
@@ -259,7 +256,7 @@ uint64_t Session::sendAllPendingQosData()
             p.setPacketId(queuedPublish.getPacketId());
             //p.setDuplicate(); // TODO: this is wrong. Until we have a retransmission system, no packets can have the DUP bit set.
  
-            count += c->writeMqttPacketAndBlameThisClient(p);
+            c->writeMqttPacketAndBlameThisClient(p);
  
             pos++;
         }
@@ -268,11 +265,9 @@ uint64_t Session::sendAllPendingQosData()
         {
             PubResponse pubRel(c->getProtocolVersion(), PacketType::PUBREL, ReasonCodes::Success, packet_id);
             MqttPacket packet(pubRel);
-            count += c->writeMqttPacketAndBlameThisClient(packet);
+            c->writeMqttPacketAndBlameThisClient(packet);
         }
     }
-
-    return count;
 }
  
 bool Session::hasActiveClient() const
@@ -78,9 +78,9 @@ public:
     const std::string &getClientId() const { return client_id; }
     std::shared_ptr<Client> makeSharedClient() const;
     void assignActiveConnection(std::shared_ptr<Client> &client);
-    void writePacket(PublishCopyFactory &copyFactory, const char max_qos, uint64_t &count);
+    void writePacket(PublishCopyFactory &copyFactory, const char max_qos);
     bool clearQosMessage(uint16_t packet_id, bool qosHandshakeEnds);
-    uint64_t sendAllPendingQosData();
+    void sendAllPendingQosData();
     bool hasActiveClient() const;
     void clearWill();
     std::shared_ptr<WillPublish> &getWill();
@@ -147,8 +147,7 @@ void SubscriptionStore::addSubscription(std::shared_ptr&lt;Client&gt; &amp;client, const s
             const std::shared_ptr<Session> &ses = session_it->second;
             deepestNode->addSubscriber(ses, qos);
             lock_guard.unlock();
-            uint64_t count = giveClientRetainedMessages(ses, subtopics, qos);
-            client->getThreadData()->incrementSentMessageCount(count);
+            giveClientRetainedMessages(ses, subtopics, qos);
         }
     }
 }
@@ -264,8 +263,7 @@ void SubscriptionStore::registerClientAndKickExistingOne(std::shared_ptr&lt;Client&gt;
     session->assignActiveConnection(client);
     client->assignSession(session);
     session->setSessionProperties(clientReceiveMax, sessionExpiryInterval, clean_start, client->getProtocolVersion());
-    uint64_t count = session->sendAllPendingQosData();
-    client->getThreadData()->incrementSentMessageCount(count);
+    session->sendAllPendingQosData();
 }
  
 /**
@@ -451,7 +449,6 @@ void SubscriptionStore::queuePacketAtSubscribers(PublishCopyFactory &amp;copyFactory
 {
     SubscriptionNode *startNode = dollar ? &rootDollar : &root;
  
-    uint64_t count = 0;
     std::forward_list<ReceivingSubscriber> subscriberSessions;
  
     {
@@ -463,13 +460,7 @@ void SubscriptionStore::queuePacketAtSubscribers(PublishCopyFactory &amp;copyFactory
  
     for(const ReceivingSubscriber &x : subscriberSessions)
     {
-        x.session->writePacket(copyFactory, x.qos, count);
-    }
-
-    std::shared_ptr<Client> sender = copyFactory.getSender();
-    if (sender)
-    {
-        sender->getThreadData()->incrementSentMessageCount(count);
+        x.session->writePacket(copyFactory, x.qos);
     }
 }
  
@@ -524,11 +515,9 @@ void SubscriptionStore::giveClientRetainedMessagesRecursively(std::vector&lt;std::s
     }
 }
  
-uint64_t SubscriptionStore::giveClientRetainedMessages(const std::shared_ptr<Session> &ses,
-                                                       const std::vector<std::string> &subscribeSubtopics, char max_qos)
+void SubscriptionStore::giveClientRetainedMessages(const std::shared_ptr<Session> &ses,
+                                                   const std::vector<std::string> &subscribeSubtopics, char max_qos)
 {
-    uint64_t count = 0;
-
     RetainedMessageNode *startNode = &retainedMessagesRoot;
     if (!subscribeSubtopics.empty() && !subscribeSubtopics[0].empty() > 0 && subscribeSubtopics[0][0] == '$')
         startNode = &retainedMessagesRootDollar;
@@ -544,10 +533,8 @@ uint64_t SubscriptionStore::giveClientRetainedMessages(const std::shared_ptr&lt;Ses
     for(Publish &publish : packetList)
     {
         PublishCopyFactory copyFactory(&publish);
-        ses->writePacket(copyFactory, max_qos, count);
+        ses->writePacket(copyFactory, max_qos);
     }
-
-    return count;
 }
  
 void SubscriptionStore::setRetainedMessage(const Publish &publish, const std::vector<std::string> &subtopics)
@@ -150,8 +150,8 @@ public:
     void sendQueuedWillMessages();
     void queueWillMessage(const std::shared_ptr<WillPublish> &willMessage, const std::shared_ptr<Session> &session, bool forceNow = false);
     void queuePacketAtSubscribers(PublishCopyFactory &copyFactory, bool dollar = false);
-    uint64_t giveClientRetainedMessages(const std::shared_ptr<Session> &ses,
-                                        const std::vector<std::string> &subscribeSubtopics, char max_qos);
+    void giveClientRetainedMessages(const std::shared_ptr<Session> &ses,
+                                    const std::vector<std::string> &subscribeSubtopics, char max_qos);
  
     void setRetainedMessage(const Publish &publish, const std::vector<std::string> &subtopics);