Fix MQTT QoS 2 Subscriber Handling (PUBREC/PUBREL/PUBCOMP) and Add Tests (#90)

* Fix QoS handling in handlePacket for MQTT messages

* Add QoS 2 message handling test case

* Update CHANGELOG.md to reflect improvements in QoS handling and test coverage

Author: MarcAntoineCRUE

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to
 [Semantic Versioning](https://semver.org/).
 
+## Unreleased
+
+### Changed
+
+* Improved test coverage for all QoS levels (0, 1, 2)
+* Increased robustness and correctness in client-side handling of QoS 1 and QoS 2 messages (proper handling of PUBACK, PUBREC, PUBREL, PUBCOMP sequences)
+* Merged and cleaned up tests to ensure MQTT protocol compliance for all QoS scenarios
+
 ## [3.3.0] - 2025-12-14
 
 ### Added

src/PubSubClient.cpp

@@ -393,10 +393,13 @@ bool PubSubClient::handlePacket(uint8_t hdrLen, size_t length) {
                         ERROR_PSC_PRINTF_P("handlePacket(): Missing msgId in QoS 1/2 message\n");
                         return false;
                     }
+                    uint8_t publishQos = MQTT_HDR_GET_QOS(_buffer[0]);  // save QoS before _buffer[0] is overwritten
                     uint16_t msgId = (_buffer[payloadOffset] << 8) + _buffer[payloadOffset + 1];
                     callback(topic, payload + 2, payloadLen - 2);  // remove the msgId from the callback payload
 
-                    _buffer[0] = MQTTPUBACK;
+                    // QoS 1: respond with PUBACK
+                    // QoS 2: respond with PUBREC (first step of the QoS 2 subscriber handshake)
+                    _buffer[0] = (publishQos == MQTT_QOS1) ? MQTTPUBACK : MQTTPUBREC;
                     _buffer[1] = 2;
                     _buffer[2] = (msgId >> 8);
                     _buffer[3] = (msgId & 0xFF);
@@ -415,20 +418,36 @@ bool PubSubClient::handlePacket(uint8_t hdrLen, size_t length) {
             // No futher action here, as resending is not supported.
             break;
         case MQTTPUBREC:
-            // MQTT Publish Received (QoS 2 publish received, part 1): See section 3.5 MQTT v3.1.1 protocol specification
+            // MQTT Publish Received (QoS 2 publisher handshake, part 1): broker acknowledges our QoS 2 PUBLISH.
+            // See section 3.5 MQTT v3.1.1 protocol specification.
             if (length < 4) {
                 ERROR_PSC_PRINTF_P("handlePacket(): Received PUBREC packet with length %zu, expected at least 4 bytes\n", length);
                 return false;
             }
-            // MQTT Publish Release (QoS 2 publish received, part 2): See section 3.6 MQTT v3.1.1 protocol specification
-            _buffer[0] = MQTTPUBREL | 2;  // PUBREL with bit 1 set
-            // bytes 1-3 of PUBREL are the same as of PUBREC
+            // MQTT Publish Release (QoS 2 publisher handshake, part 2): See section 3.6 MQTT v3.1.1 protocol specification
+            _buffer[0] = MQTTPUBREL | 2;  // PUBREL fixed header: bit 1 must be set per spec
+            // bytes 1-3 of PUBREL are the same as of PUBREC (remaining length + msgId)
+            if (_client->write(_buffer, 4) == 4) {
+                _lastOutActivity = millis();
+            }
+            break;
+        case MQTTPUBREL:
+            // MQTT Publish Release (QoS 2 subscriber handshake, part 2): broker releases the message to us.
+            // See section 3.6 MQTT v3.1.1 protocol specification.
+            if (length < 4) {
+                ERROR_PSC_PRINTF_P("handlePacket(): Received PUBREL packet with length %zu, expected at least 4 bytes\n", length);
+                return false;
+            }
+            // MQTT Publish Complete (QoS 2 subscriber handshake, part 3): See section 3.7 MQTT v3.1.1 protocol specification
+            _buffer[0] = MQTTPUBCOMP;
+            // bytes 1-3 of PUBCOMP are the same as of PUBREL (remaining length + msgId)
             if (_client->write(_buffer, 4) == 4) {
                 _lastOutActivity = millis();
             }
             break;
         case MQTTPUBCOMP:
-            // MQTT Publish Complete (QoS 2 publish received, part 3): See section 3.7 MQTT v3.1.1 protocol specification
+            // MQTT Publish Complete (QoS 2 publisher handshake, part 3): broker confirms delivery of our QoS 2 PUBLISH.
+            // See section 3.7 MQTT v3.1.1 protocol specification.
             if (length < 4) {
                 ERROR_PSC_PRINTF_P("handlePacket(): Received PUBCOMP packet with length %zu, expected at least 4 bytes\n", length);
                 return false;

tests/src/receive_spec.cpp

@@ -28,6 +28,7 @@ int test_receive_oversized_message();
 int test_resize_buffer();
 int test_receive_oversized_stream_message();
 int test_receive_qos1();
+int test_receive_qos2();
 
 void reset_callback() {
     callback_called = false;
@@ -330,6 +331,57 @@ int test_receive_qos1() {
     END_IT
 }
 
+int test_receive_qos2() {
+    IT("receives a qos2 message - responds PUBREC then PUBCOMP");
+    reset_callback();
+
+    ShimClient shimClient;
+    shimClient.setAllowConnect(true);
+
+    byte connack[] = {0x20, 0x02, 0x00, 0x00};
+    shimClient.respond(connack, 4);
+
+    PubSubClient client(server, 1883, callback, shimClient);
+    bool rc = client.connect("client_test1");
+    IS_TRUE(rc);
+
+    // QoS 2 PUBLISH from broker (0x34 = MQTTPUBLISH | QoS2 bits)
+    // Fixed header 0x34, remaining length 0x10 (16), topic len 0x0005, topic "topic",
+    // msgId 0x1234, payload "payload"
+    byte publish[] = {0x34, 0x10, 0x0, 0x5, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x34, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
+    shimClient.respond(publish, 18);
+
+    // Client must respond with PUBREC (0x50), remaining length 2, msgId 0x1234
+    byte pubrec[] = {0x50, 0x02, 0x12, 0x34};
+    shimClient.expect(pubrec, 4);
+
+    rc = client.loop();
+    IS_TRUE(rc);
+    IS_TRUE(callback_called);
+    IS_TRUE(strcmp(lastTopic, "topic") == 0);
+    IS_TRUE(memcmp(lastPayload, "payload", 7) == 0);
+    IS_TRUE(lastLength == 7);
+    IS_FALSE(shimClient.error());
+
+    reset_callback();
+
+    // Broker sends PUBREL (0x62 = MQTTPUBREL | bit1), remaining length 2, msgId 0x1234
+    byte pubrel[] = {0x62, 0x02, 0x12, 0x34};
+    shimClient.respond(pubrel, 4);
+
+    // Client must respond with PUBCOMP (0x70), remaining length 2, msgId 0x1234
+    byte pubcomp[] = {0x70, 0x02, 0x12, 0x34};
+    shimClient.expect(pubcomp, 4);
+
+    rc = client.loop();
+    IS_TRUE(rc);
+    IS_FALSE(callback_called);  // callback must NOT fire again on PUBREL
+
+    IS_FALSE(shimClient.error());
+
+    END_IT
+}
+
 int main() {
     SUITE("Receive");
     test_receive_callback();
@@ -340,6 +392,7 @@ int main() {
     test_resize_buffer();
     test_receive_oversized_stream_message();
     test_receive_qos1();
+    test_receive_qos2();
 
     FINISH
 }