Flow control based on receive maximum

It merely drops packets when they exceed it. The specs are unclear about whether you're supposed to delay transmission until the quota is non-negative again. I decided against it because of increased complexity, and because on a continously overloaded client, this makes no sense. Effectively, this formalizes the 'max qos pending' mechanism that was already in place. It also includes PUBACK/PUBREL/PUBCOMP error handling, because that needed to be done for proper quota control.

Flow control based on receive maximum
It merely drops packets when they exceed it. The specs are unclear about whether you're supposed to delay transmission until the quota is non-negative again. I decided against it because of increased complexity, and because on a continously overloaded client, this makes no sense. Effectively, this formalizes the 'max qos pending' mechanism that was already in place. It also includes PUBACK/PUBREL/PUBCOMP error handling, because that needed to be done for proper quota control.
Wiebe Cazemier
1 parent ac53fea0
Showing 12 changed files with 143 additions and 97 deletions
client.cpp
client.h
configfileparser.cpp
mqttpacket.cpp
qospacketqueue.cpp
qospacketqueue.h
session.cpp
session.h
sessionsandsubscriptionsdb.cpp
subscriptionstore.cpp
subscriptionstore.h
types.h
@@ -27,9 +27,9 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #include "utils.h"
 #include "threadglobals.h"
-StowedClientRegistrationData::StowedClientRegistrationData(bool clean_start, uint16_t maxQosPackets, uint32_t sessionExpiryInterval) :
+StowedClientRegistrationData::StowedClientRegistrationData(bool clean_start, uint16_t clientReceiveMax, uint32_t sessionExpiryInterval) :
     clean_start(clean_start),
-    maxQosPackets(maxQosPackets),
+    clientReceiveMax(clientReceiveMax),
     sessionExpiryInterval(sessionExpiryInterval)
 {
@@ -448,9 +448,9 @@ void Client::serverInitiatedDisconnect(ReasonCodes reason)
  * @param maxQosPackets
  * @param sessionExpiryInterval
  */
-void Client::setRegistrationData(bool clean_start, uint16_t maxQosPackets, uint32_t sessionExpiryInterval)
+void Client::setRegistrationData(bool clean_start, uint16_t client_receive_max, uint32_t sessionExpiryInterval)
 {
-    this->registrationData = std::make_unique<StowedClientRegistrationData>(clean_start, maxQosPackets, sessionExpiryInterval);
+    this->registrationData = std::make_unique<StowedClientRegistrationData>(clean_start, client_receive_max, sessionExpiryInterval);
 }
 const std::unique_ptr<StowedClientRegistrationData> &Client::getRegistrationData() const
@@ -48,10 +48,10 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 struct StowedClientRegistrationData
 {
     const bool clean_start;
-    const uint16_t maxQosPackets;
+    const uint16_t clientReceiveMax;
     const uint32_t sessionExpiryInterval;
-    StowedClientRegistrationData(bool clean_start, uint16_t maxQosPackets, uint32_t sessionExpiryInterval);
+    StowedClientRegistrationData(bool clean_start, uint16_t clientReceiveMax, uint32_t sessionExpiryInterval);
 };
 class Client
@@ -171,7 +171,7 @@ public:
     void sendOrQueueWill();
     void serverInitiatedDisconnect(ReasonCodes reason);
-    void setRegistrationData(bool clean_start, uint16_t maxQosPackets, uint32_t sessionExpiryInterval);
+    void setRegistrationData(bool clean_start, uint16_t client_receive_max, uint32_t sessionExpiryInterval);
     const std::unique_ptr<StowedClientRegistrationData> &getRegistrationData() const;
     void clearRegistrationData();
@@ -445,9 +445,9 @@ void ConfigFileParser::loadFile(bool test)
                 if (key == "max_qos_msg_pending_per_client")
                 {
                     int newVal = std::stoi(value);
-                    if (newVal < 32 || newVal > 65530)
+                    if (newVal < 32 || newVal > 65535)
                     {
-                        throw ConfigFileException(formatString("max_qos_msg_pending_per_client value '%d' is invalid. Valid values between 32 and 65530.", newVal));
+                        throw ConfigFileException(formatString("max_qos_msg_pending_per_client value '%d' is invalid. Valid values between 32 and 65535.", newVal));
                     }
                     tmpSettings->maxQosMsgPendingPerClient = newVal;
                 }
@@ -385,7 +385,7 @@ void MqttPacket::handleConnect()
     uint16_t keep_alive = readTwoBytesToUInt16();
-    uint16_t max_qos_packets = settings.maxQosMsgPendingPerClient;
+    uint16_t client_receive_max = settings.maxQosMsgPendingPerClient;
     uint32_t session_expire = settings.getExpireSessionAfterSeconds();
     uint32_t max_outgoing_packet_size = settings.maxPacketSize;
     uint16_t max_outgoing_topic_aliases = 0; // Default MUST BE 0, meaning server won't initiate aliases
@@ -410,7 +410,7 @@ void MqttPacket::handleConnect()
                 session_expire = std::min<uint32_t>(readFourBytesToUint32(), session_expire);
                 break;
             case Mqtt5Properties::ReceiveMaximum:
-                max_qos_packets = std::min<int16_t>(readTwoBytesToUInt16(), max_qos_packets);
+                client_receive_max = std::min<int16_t>(readTwoBytesToUInt16(), client_receive_max);
                 break;
             case Mqtt5Properties::MaximumPacketSize:
                 max_outgoing_packet_size = std::min<uint32_t>(readFourBytesToUint32(), max_outgoing_packet_size);
@@ -445,6 +445,11 @@ void MqttPacket::handleConnect()
         }
     }
+    if (client_receive_max == 0 || max_outgoing_packet_size == 0)
+    {
+        throw ProtocolError("Receive max or max outgoing packet size can't be 0.", ReasonCodes::ProtocolError);
+    }
+
     std::string client_id = readBytesToString();
     std::string username;
@@ -610,7 +615,7 @@ void MqttPacket::handleConnect()
         {
             connAck->propertyBuilder = std::make_shared<Mqtt5PropertyBuilder>();
             connAck->propertyBuilder->writeSessionExpiry(session_expire);
-            connAck->propertyBuilder->writeReceiveMax(max_qos_packets);
+            connAck->propertyBuilder->writeReceiveMax(settings.maxQosMsgPendingPerClient);
             connAck->propertyBuilder->writeRetainAvailable(1);
             connAck->propertyBuilder->writeMaxPacketSize(sender->getMaxIncomingPacketSize());
             if (clientIdGenerated)
@@ -629,7 +634,7 @@ void MqttPacket::handleConnect()
         sender->stageConnack(std::move(connAck));
     }
-    sender->setRegistrationData(clean_start, max_qos_packets, session_expire);
+    sender->setRegistrationData(clean_start, client_receive_max, session_expire);
     Authentication &authentication = *ThreadGlobals::getAuth();
     AuthResult authResult = AuthResult::login_denied;
@@ -1176,7 +1181,7 @@ void MqttPacket::handlePublish()
 void MqttPacket::handlePubAck()
 {
     uint16_t packet_id = readTwoBytesToUInt16();
-    sender->getSession()->clearQosMessage(packet_id);
+    sender->getSession()->clearQosMessage(packet_id, true);
 }
 /**
@@ -1185,12 +1190,29 @@ void MqttPacket::handlePubAck()
 void MqttPacket::handlePubRec()
 {
     const uint16_t packet_id = readTwoBytesToUInt16();
-    sender->getSession()->clearQosMessage(packet_id);
-    sender->getSession()->addOutgoingQoS2MessageId(packet_id);
-    PubResponse pubRel(this->protocolVersion, PacketType::PUBREL, ReasonCodes::Success, packet_id);
-    MqttPacket response(pubRel);
-    sender->writeMqttPacket(response);
+    ReasonCodes reasonCode = ReasonCodes::Success; // Default when not specified, or MQTT3
+
+    if (!atEnd())
+    {
+        reasonCode = static_cast<ReasonCodes>(readByte());
+    }
+
+    const bool publishTerminatesHere = reasonCode >= ReasonCodes::UnspecifiedError;
+    const bool foundAndRemoved = sender->getSession()->clearQosMessage(packet_id, publishTerminatesHere);
+
+    // "If it has sent a PUBREC with a Reason Code of 0x80 or greater, the receiver MUST treat any subsequent PUBLISH packet
+    // that contains that Packet Identifier as being a new Application Message."
+    if (!publishTerminatesHere)
+    {
+        sender->getSession()->addOutgoingQoS2MessageId(packet_id);
+
+        // MQTT5: "[The sender] MUST send a PUBREL packet when it receives a PUBREC packet from the receiver with a Reason Code value less than 0x80"
+        const ReasonCodes reason = foundAndRemoved ? ReasonCodes::Success : ReasonCodes::PacketIdentifierNotFound;
+        PubResponse pubRel(this->protocolVersion, PacketType::PUBREL, reason, packet_id);
+        MqttPacket response(pubRel);
+        sender->writeMqttPacket(response);
+    }
 }
 /**
@@ -1203,9 +1225,10 @@ void MqttPacket::handlePubRel()
         throw ProtocolError("PUBREL first byte LSB must be 0010.", ReasonCodes::MalformedPacket);
     const uint16_t packet_id = readTwoBytesToUInt16();
-    sender->getSession()->removeIncomingQoS2MessageId(packet_id);
+    const bool foundAndRemoved = sender->getSession()->removeIncomingQoS2MessageId(packet_id);
+    const ReasonCodes reason = foundAndRemoved ? ReasonCodes::Success : ReasonCodes::PacketIdentifierNotFound;
-    PubResponse pubcomp(this->protocolVersion, PacketType::PUBCOMP, ReasonCodes::Success, packet_id);
+    PubResponse pubcomp(this->protocolVersion, PacketType::PUBCOMP, reason, packet_id);
     MqttPacket response(pubcomp);
     sender->writeMqttPacket(response);
 }
@@ -27,8 +27,10 @@ size_t QueuedPublish::getApproximateMemoryFootprint() const
 }
-void QoSPublishQueue::erase(const uint16_t packet_id)
+bool QoSPublishQueue::erase(const uint16_t packet_id)
 {
+    bool result = false;
+
     auto it = queue.begin();
     auto end = queue.end();
     while (it != end)
@@ -43,12 +45,15 @@ void QoSPublishQueue::erase(const uint16_t packet_id)
                 qosQueueBytes = 0;
             queue.erase(it);
+            result = true;
             break;
         }
         it++;
     }
+
+    return  result;
 }
 std::list<QueuedPublish>::iterator QoSPublishQueue::erase(std::list<QueuedPublish>::iterator pos)
@@ -30,7 +30,7 @@ class QoSPublishQueue
     ssize_t qosQueueBytes = 0;
 public:
-    void erase(const uint16_t packet_id);
+    bool erase(const uint16_t packet_id);
     std::list<QueuedPublish>::iterator erase(std::list<QueuedPublish>::iterator pos);
     size_t size() const;
     size_t getByteSize() const;
@@ -27,12 +27,20 @@ Session::Session()
     const Settings &settings = *ThreadGlobals::getSettings();
     // Sessions also get defaults from the handleConnect() method, but when you create sessions elsewhere, we do need some sensible defaults.
-    this->maxQosMsgPending = settings.maxQosMsgPendingPerClient;
+    this->flowControlQuota = settings.maxQosMsgPendingPerClient;
     this->sessionExpiryInterval = settings.expireSessionsAfterSeconds;
 }
-bool Session::requiresPacketRetransmission() const
+void Session::increaseFlowControlQuota()
 {
+    flowControlQuota++;
+    this->flowControlQuota = std::min<int>(flowControlQuota, flowControlCealing);
+}
+
+bool Session::requiresQoSQueueing() const
+{
+    return  true;
+
     const std::shared_ptr<Client> client = makeSharedClient();
     if (!client)
@@ -49,8 +57,7 @@ bool Session::requiresPacketRetransmission() const
 void Session::increasePacketId()
 {
     nextPacketId++;
-    if (nextPacketId == 0)
-        nextPacketId++;
+    nextPacketId = std::max<uint16_t>(nextPacketId, 1);
 }
 /**
@@ -146,81 +153,65 @@ void Session::writePacket(PublishCopyFactory &amp;copyFactory, const char max_qos, u
         }
         else if (effectiveQos > 0)
         {
-            const bool requiresRetransmission = requiresPacketRetransmission();
+            std::unique_lock<std::mutex> locker(qosQueueMutex);
-            if (requiresRetransmission)
+            if (this->flowControlQuota <= 0 || (qosPacketQueue.getByteSize() >= settings->maxQosBytesPendingPerClient && qosPacketQueue.size() > 0))
             {
-                std::unique_lock<std::mutex> locker(qosQueueMutex);
-
-                const size_t totalQosPacketsInTransit = qosPacketQueue.size() + incomingQoS2MessageIds.size() + outgoingQoS2MessageIds.size();
-                if (totalQosPacketsInTransit >= maxQosMsgPending
-                    || (qosPacketQueue.getByteSize() >= settings->maxQosBytesPendingPerClient && qosPacketQueue.size() > 0))
+                if (QoSLogPrintedAtId != nextPacketId)
                 {
-                    if (QoSLogPrintedAtId != nextPacketId)
-                    {
-                        logger->logf(LOG_WARNING, "Dropping QoS message(s) for client '%s', because max in-transit packet count reached.", client_id.c_str());
-                        QoSLogPrintedAtId = nextPacketId;
-                    }
-                    return;
+                    logger->logf(LOG_WARNING, "Dropping QoS message(s) for client '%s', because it hasn't seen enough PUBACK/PUBCOMP/PUBRECs to release places "
+                                              "or it exceeded 'max_qos_bytes_pending_per_client'.", client_id.c_str());
+                    QoSLogPrintedAtId = nextPacketId;
                 }
+                return;
+            }
-                increasePacketId();
+            increasePacketId();
+            flowControlQuota--;
+            if (requiresQoSQueueing())
                 qosPacketQueue.queuePublish(copyFactory, nextPacketId, effectiveQos);
-                if (c)
-                {
-                    count += c->writeMqttPacketAndBlameThisClient(copyFactory, effectiveQos, nextPacketId);
-                }
-            }
-            else
+            if (c)
             {
-                // We don't need to make a copy of the packet in this branch, because:
-                // - The packet to give the client won't shrink in size because source and client have a packet_id.
-                // - We don't have to store the copy in the session for retransmission, see Session::requiresPacketRetransmission()
-                // So, we just keep altering the original published packet.
-
-                std::unique_lock<std::mutex> locker(qosQueueMutex);
-
-                if (qosInFlightCounter >= 65530) // Includes a small safety margin.
-                {
-                    if (QoSLogPrintedAtId != nextPacketId)
-                    {
-                        logger->logf(LOG_WARNING, "Dropping QoS message(s) for client '%s', because it hasn't seen enough PUBACKs to release places.", client_id.c_str());
-                        QoSLogPrintedAtId = nextPacketId;
-                    }
-                    return;
-                }
-
-                increasePacketId();
-
-                qosInFlightCounter++;
-                assert(c); // with requiresRetransmission==false, there must be a client.
                 count += c->writeMqttPacketAndBlameThisClient(copyFactory, effectiveQos, nextPacketId);
             }
         }
     }
 }
-void Session::clearQosMessage(uint16_t packet_id)
+/**
+ * @brief Session::clearQosMessage clears a QOS message from the queue. Note that in QoS 2, that doesn't complete the handshake.
+ * @param packet_id
+ * @param qosHandshakeEnds can be set to true when you know the QoS handshake ends, (like) when PUBREC contains an error.
+ * @return whether the packet_id in question was found.
+ */
+bool Session::clearQosMessage(uint16_t packet_id, bool qosHandshakeEnds)
 {
 #ifndef NDEBUG
     logger->logf(LOG_DEBUG, "Clearing QoS message for '%s', packet id '%d'. Left in queue: %d", client_id.c_str(), packet_id, qosPacketQueue.size());
 #endif
+    bool result = false;
+
     std::lock_guard<std::mutex> locker(qosQueueMutex);
-    if (requiresPacketRetransmission())
-        qosPacketQueue.erase(packet_id);
+    if (requiresQoSQueueing())
+        result = qosPacketQueue.erase(packet_id);
     else
     {
-        qosInFlightCounter--;
-        qosInFlightCounter = std::max<int>(0, qosInFlightCounter); // Should never happen, but in case we receive too many PUBACKs.
+        result = true;
     }
-}
+    if (qosHandshakeEnds)
+    {
+        increaseFlowControlQuota();
+    }
+
+    return result;
+}
 /**
- * @brief Session::sendPendingQosMessages sends pending publishes and QoS2 control packets.
+ * @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
@@ -234,7 +225,7 @@ void Session::clearQosMessage(uint16_t packet_id)
  * 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::sendPendingQosMessages()
+uint64_t Session::sendAllPendingQosData()
 {
     uint64_t count = 0;
@@ -254,12 +245,23 @@ uint64_t Session::sendPendingQosMessages()
                 pos = qosPacketQueue.erase(pos);
                 continue;
             }
-            pos++;
+
+            if (flowControlQuota <= 0)
+            {
+                logger->logf(LOG_WARNING, "Dropping QoS message(s) for client '%s', because it exceeds its receive maximum.", client_id.c_str());
+                pos = qosPacketQueue.erase(pos);
+                continue;
+            }
+
+            flowControlQuota--;
             MqttPacket p(c->getProtocolVersion(), pub);
-            p.setDuplicate();
+            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);
+
+            pos++;
         }
         for (const uint16_t packet_id : outgoingQoS2MessageIds)
@@ -310,7 +312,7 @@ bool Session::incomingQoS2MessageIdInTransit(uint16_t packet_id)
     return it != incomingQoS2MessageIds.end();
 }
-void Session::removeIncomingQoS2MessageId(u_int16_t packet_id)
+bool Session::removeIncomingQoS2MessageId(u_int16_t packet_id)
 {
     assert(packet_id > 0);
@@ -320,9 +322,16 @@ void Session::removeIncomingQoS2MessageId(u_int16_t packet_id)
     logger->logf(LOG_DEBUG, "As QoS 2 receiver: publish released (PUBREL) for '%s', packet id '%d'. Left in queue: %d", client_id.c_str(), packet_id, incomingQoS2MessageIds.size());
 #endif
+    bool result = false;
+
     const auto it = incomingQoS2MessageIds.find(packet_id);
     if (it != incomingQoS2MessageIds.end())
+    {
         incomingQoS2MessageIds.erase(it);
+        result = true;
+    }
+
+    return result;
 }
 void Session::addOutgoingQoS2MessageId(uint16_t packet_id)
@@ -342,6 +351,8 @@ void Session::removeOutgoingQoS2MessageId(u_int16_t packet_id)
     const auto it = outgoingQoS2MessageIds.find(packet_id);
     if (it != outgoingQoS2MessageIds.end())
         outgoingQoS2MessageIds.erase(it);
+
+    increaseFlowControlQuota();
 }
 /**
@@ -355,9 +366,10 @@ bool Session::getDestroyOnDisconnect() const
     return destroyOnDisconnect;
 }
-void Session::setSessionProperties(uint16_t maxQosPackets, uint32_t sessionExpiryInterval, bool clean_start, ProtocolVersion protocol_version)
+void Session::setSessionProperties(uint16_t clientReceiveMax, uint32_t sessionExpiryInterval, bool clean_start, ProtocolVersion protocol_version)
 {
-    this->maxQosMsgPending = maxQosPackets;
+    this->flowControlQuota = clientReceiveMax;
+    this->flowControlCealing = clientReceiveMax;
     this->sessionExpiryInterval = sessionExpiryInterval;
     if (protocol_version <= ProtocolVersion::Mqtt311 && clean_start)
@@ -45,9 +45,15 @@ class Session
     std::set<uint16_t> outgoingQoS2MessageIds;
     std::mutex qosQueueMutex;
     uint16_t nextPacketId = 0;
-    uint16_t qosInFlightCounter = 0;
+
+    /**
+     * Even though flow control data is not part of the session state, I'm keeping it here because there are already
+     * mutexes that they can be placed under, saving additional synchronization.
+     */
+    int flowControlCealing = 0xFFFF;
+    int flowControlQuota = 0xFFFF;
+
     uint32_t sessionExpiryInterval = 0;
-    uint16_t maxQosMsgPending;
     uint16_t QoSLogPrintedAtId = 0;
     bool destroyOnDisconnect = false;
     std::shared_ptr<WillPublish> willPublish;
@@ -55,7 +61,9 @@ class Session
     std::chrono::time_point<std::chrono::steady_clock> removalQueuedAt;
     Logger *logger = Logger::getInstance();
-    bool requiresPacketRetransmission() const;
+    void increaseFlowControlQuota();
+
+    bool requiresQoSQueueing() const;
     void increasePacketId();
     Session(const Session &other);
@@ -71,8 +79,8 @@ public:
     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 clearQosMessage(uint16_t packet_id);
-    uint64_t sendPendingQosMessages();
+    bool clearQosMessage(uint16_t packet_id, bool qosHandshakeEnds);
+    uint64_t sendAllPendingQosData();
     bool hasActiveClient() const;
     void clearWill();
     std::shared_ptr<WillPublish> &getWill();
@@ -80,14 +88,14 @@ public:
     void addIncomingQoS2MessageId(uint16_t packet_id);
     bool incomingQoS2MessageIdInTransit(uint16_t packet_id);
-    void removeIncomingQoS2MessageId(u_int16_t packet_id);
+    bool removeIncomingQoS2MessageId(u_int16_t packet_id);
     void addOutgoingQoS2MessageId(uint16_t packet_id);
     void removeOutgoingQoS2MessageId(u_int16_t packet_id);
     bool getDestroyOnDisconnect() const;
-    void setSessionProperties(uint16_t maxQosPackets, uint32_t sessionExpiryInterval, bool clean_start, ProtocolVersion protocol_version);
+    void setSessionProperties(uint16_t clientReceiveMax, uint32_t sessionExpiryInterval, bool clean_start, ProtocolVersion protocol_version);
     void setSessionExpiryInterval(uint32_t newVal);
     void setQueuedRemovalAt();
     uint32_t getSessionExpiryInterval() const;
@@ -161,12 +161,10 @@ SessionsAndSubscriptionsResult SessionsAndSubscriptionsDB::readDataV2()
             const uint32_t compensatedSessionExpiry = persistence_state_age > originalSessionExpiryInterval ? 0 : originalSessionExpiryInterval - persistence_state_age;
             const uint32_t sessionExpiryInterval = std::min<uint32_t>(compensatedSessionExpiry, settings->getExpireSessionAfterSeconds());
-            const uint16_t maxQosPending = std::min<uint16_t>(readUint16(eofFound), settings->maxQosMsgPendingPerClient);
-
             // We will set the session expiry interval as it would have had time continued. If a connection picks up session, it will update
             // it with a more relevant value.
             // The protocol version 5 is just dummy, to get the behavior I want.
-            ses->setSessionProperties(maxQosPending, sessionExpiryInterval, 0, ProtocolVersion::Mqtt5);
+            ses->setSessionProperties(0xFFFF, sessionExpiryInterval, 0, ProtocolVersion::Mqtt5);
             const uint16_t hasWill = readUint16(eofFound);
@@ -314,7 +312,6 @@ void SessionsAndSubscriptionsDB::saveData(const std::vector&lt;std::unique_ptr&lt;Sess
         writeUint16(ses->nextPacketId);
         writeUint32(ses->getCurrentSessionExpiryInterval());
-        writeUint16(ses->maxQosMsgPending);
         const bool hasWillThatShouldSurviveRestart = ses->getWill().operator bool() && ses->getWill()->will_delay > 0;
         writeUint16(static_cast<uint16_t>(hasWillThatShouldSurviveRestart));
@@ -213,7 +213,7 @@ void SubscriptionStore::registerClientAndKickExistingOne(std::shared_ptr&lt;Client&gt;
     if (registrationData)
     {
-        registerClientAndKickExistingOne(client, registrationData->clean_start, registrationData->maxQosPackets, registrationData->sessionExpiryInterval);
+        registerClientAndKickExistingOne(client, registrationData->clean_start, registrationData->clientReceiveMax, registrationData->sessionExpiryInterval);
         client->clearRegistrationData();
     }
     else
@@ -224,7 +224,7 @@ void SubscriptionStore::registerClientAndKickExistingOne(std::shared_ptr&lt;Client&gt;
 }
 // Removes an existing client when it already exists [MQTT-3.1.4-2].
-void SubscriptionStore::registerClientAndKickExistingOne(std::shared_ptr<Client> &client, bool clean_start, uint16_t maxQosPackets, uint32_t sessionExpiryInterval)
+void SubscriptionStore::registerClientAndKickExistingOne(std::shared_ptr<Client> &client, bool clean_start, uint16_t clientReceiveMax, uint32_t sessionExpiryInterval)
 {
     RWLockGuard lock_guard(&subscriptionsRwlock);
     lock_guard.wrlock();
@@ -261,8 +261,8 @@ void SubscriptionStore::registerClientAndKickExistingOne(std::shared_ptr&lt;Client&gt;
     session->assignActiveConnection(client);
     client->assignSession(session);
-    session->setSessionProperties(maxQosPackets, sessionExpiryInterval, clean_start, client->getProtocolVersion());
-    uint64_t count = session->sendPendingQosMessages();
+    session->setSessionProperties(clientReceiveMax, sessionExpiryInterval, clean_start, client->getProtocolVersion());
+    uint64_t count = session->sendAllPendingQosData();
     client->getThreadData()->incrementSentMessageCount(count);
 }
@@ -161,7 +161,7 @@ public:
     void addSubscription(std::shared_ptr<Client> &client, const std::string &topic, const std::vector<std::string> &subtopics, char qos);
     void removeSubscription(std::shared_ptr<Client> &client, const std::string &topic);
     void registerClientAndKickExistingOne(std::shared_ptr<Client> &client);
-    void registerClientAndKickExistingOne(std::shared_ptr<Client> &client, bool clean_start, uint16_t maxQosPackets, uint32_t sessionExpiryInterval);
+    void registerClientAndKickExistingOne(std::shared_ptr<Client> &client, bool clean_start, uint16_t clientReceiveMax, uint32_t sessionExpiryInterval);
     std::shared_ptr<Session> lockSession(const std::string &clientid);
     void sendQueuedWillMessages();
@@ -132,6 +132,7 @@ enum class ReasonCodes
     TopicFilterInvalid = 143,
     TopicNameInvalid = 144,
     PacketIdentifierInUse = 145,
+    PacketIdentifierNotFound = 146,
     ReceiveMaximumExceeded = 147,
     TopicAliasInvalid = 148,
     PacketTooLarge = 149,