Don't count dropped publish message in stats

Wiebe Cazemier
1 parent 339a55fe
Showing 3 changed files with 13 additions and 13 deletions
client.cpp
client.h
session.cpp
@@ -177,7 +177,7 @@ void Client::writeText(const std::string &amp;text)
     setReadyForWriting(true);
 }
  
-void Client::writeMqttPacket(const MqttPacket &packet, const char qos)
+int Client::writeMqttPacket(const MqttPacket &packet, const char qos)
 {
     std::lock_guard<std::mutex> locker(writeBufMutex);
  
@@ -192,7 +192,7 @@ void Client::writeMqttPacket(const MqttPacket &amp;packet, const char qos)
     // QoS packet are queued and limited elsewhere.
     if (packet.packetType == PacketType::PUBLISH && qos == 0 && packet.getSizeIncludingNonPresentHeader() > writebuf.freeSpace())
     {
-        return;
+        return 0;
     }
  
     writebuf.ensureFreeSpace(packet.getSizeIncludingNonPresentHeader());
@@ -211,19 +211,22 @@ void Client::writeMqttPacket(const MqttPacket &amp;packet, const char qos)
         setReadyForDisconnect();
  
     setReadyForWriting(true);
+    return 1;
 }
  
 // Helper method to avoid the exception ending up at the sender of messages, which would then get disconnected.
-void Client::writeMqttPacketAndBlameThisClient(const MqttPacket &packet, const char qos)
+int Client::writeMqttPacketAndBlameThisClient(const MqttPacket &packet, const char qos)
 {
     try
     {
-        this->writeMqttPacket(packet, qos);
+        return this->writeMqttPacket(packet, qos);
     }
     catch (std::exception &ex)
     {
         threadData->removeClientQueued(fd);
     }
+
+    return 0;
 }
  
 // Ping responses are always the same, so hardcoding it for optimization.
@@ -121,8 +121,8 @@ public:
  
     void writeText(const std::string &text);
     void writePingResp();
-    void writeMqttPacket(const MqttPacket &packet, const char qos = 0);
-    void writeMqttPacketAndBlameThisClient(const MqttPacket &packet, const char qos);
+    int writeMqttPacket(const MqttPacket &packet, const char qos = 0);
+    int writeMqttPacketAndBlameThisClient(const MqttPacket &packet, const char qos);
     bool writeBufIntoFd();
     bool readyForDisconnecting() const { return disconnectWhenBytesWritten && writebuf.usedBytes() == 0; }
  
@@ -137,8 +137,7 @@ void Session::writePacket(const MqttPacket &amp;packet, char max_qos, bool retain, u
  
             if (c)
             {
-                c->writeMqttPacketAndBlameThisClient(packet, qos);
-                count++;
+                count += c->writeMqttPacketAndBlameThisClient(packet, qos);
             }
         }
         else if (qos > 0)
@@ -165,9 +164,8 @@ void Session::writePacket(const MqttPacket &amp;packet, char max_qos, bool retain, u
             std::shared_ptr<Client> c = makeSharedClient();
             if (c)
             {
-                c->writeMqttPacketAndBlameThisClient(*copyPacket.get(), qos);
+                count += c->writeMqttPacketAndBlameThisClient(*copyPacket.get(), qos);
                 copyPacket->setDuplicate(); // Any dealings with this packet from here will be a duplicate.
-                count++;
             }
         }
     }
@@ -201,16 +199,15 @@ uint64_t Session::sendPendingQosMessages()
         std::lock_guard<std::mutex> locker(qosQueueMutex);
         for (const std::shared_ptr<MqttPacket> &qosMessage : qosPacketQueue)
         {
-            c->writeMqttPacketAndBlameThisClient(*qosMessage.get(), qosMessage->getQos());
+            count += c->writeMqttPacketAndBlameThisClient(*qosMessage.get(), qosMessage->getQos());
             qosMessage->setDuplicate(); // Any dealings with this packet from here will be a duplicate.
-            count++;
         }
  
         for (const uint16_t packet_id : outgoingQoS2MessageIds)
         {
             PubRel pubRel(packet_id);
             MqttPacket packet(pubRel);
-            c->writeMqttPacketAndBlameThisClient(packet, 2);
+            count += c->writeMqttPacketAndBlameThisClient(packet, 2);
         }
     }