mqtt: qos2

Author: lws-team

include/libwebsockets/lws-callbacks.h

@@ -885,20 +885,26 @@ enum lws_callback_reasons {
 	 * Return nonzero to close the wsi.
 	 */
 
-	LWS_CALLBACK_HTTP_INTERCEPTOR_CHECK				= 213,
+	LWS_CALLBACK_MQTT_QOS2_RX_COMPLETE			= 213,
+	/**< When a QoS2 message has fully completed the transaction (PUBREL
+	 * received, PUBCOMP sent), this callback is generated.
+	 * `in` will point to the `uint16_t` packet ID that completed.
+	 */
+
+	LWS_CALLBACK_HTTP_INTERCEPTOR_CHECK			= 214,
 	/**< A mount has a interceptor_path enabled, this callback asks the
 	 * protocol bound to that mount if it is OK for this request to
 	 * proceed.  If returning 0, the request proceeds to the original
 	 * mount.  If nonzero, the request is diverted to the interceptor_path
 	 * mount.
 	 */
 
-	LWS_CALLBACK_GET_PSS_SIZE					= 214,
+	LWS_CALLBACK_GET_PSS_SIZE					= 215,
 	/**< Called when a protocol wants to specify its PSS size at runtime.
 	 *   If the protocol structure has per_session_data_size == 0, lws will
 	 *   call this to get the size to allocate for the session. */
 
-	LWS_CALLBACK_DHT_VERB_DISPATCH					= 215,
+	LWS_CALLBACK_DHT_VERB_DISPATCH					= 216,
 	/**< Sent to the user protocol handler callback when a DHT message
 	 * carrying a registered verb has been matched by lws-dht.
 	 * `in` is a pointer to `struct lws_dht_verb_dispatch_args` containing

include/libwebsockets/lws-mqtt.h

@@ -73,7 +73,7 @@ typedef struct lws_mqtt_str_st lws_mqtt_str_t;
 typedef enum {
 	QOS0,
 	QOS1,
-	QOS2,				/* not supported */
+	QOS2,
 	RESERVED_QOS_LEVEL,
 	FAILURE_QOS_LEVEL = 0x80
 } lws_mqtt_qos_levels_t;
@@ -88,6 +88,11 @@ typedef union {
 	uint8_t 		bits;
 } lws_mqtt_fixed_hdr_t;
 
+typedef struct lws_mqtt_qos2_state_ops {
+	int (*rx_add)(struct lws *wsi, const char *client_id, uint16_t pkt_id);
+	int (*rx_remove)(struct lws *wsi, const char *client_id, uint16_t pkt_id);
+} lws_mqtt_qos2_state_ops_t;
+
 /*
  * MQTT connection parameters, passed into struct
  * lws_client_connect_info to establish a connection using
@@ -122,6 +127,7 @@ typedef struct lws_mqtt_client_connect_param_s {
 						   parameters */
 	const char 			*username;
 	const char 			*password;
+	const lws_mqtt_qos2_state_ops_t	*qos2_state_ops;
 	uint8_t				aws_iot;
 } lws_mqtt_client_connect_param_t;
 
@@ -383,4 +389,16 @@ LWS_VISIBLE LWS_EXTERN int LWS_WARN_UNUSED_RESULT
 lws_mqtt_client_send_unsubcribe(struct lws *wsi,
 				const lws_mqtt_subscribe_param_t *unsub);
 
+/**
+ * lws_mqtt_client_qos2_rx_add() - inject a saved QoS2 packet ID into the rx list
+ *
+ * \param wsi: the mqtt child wsi
+ * \param pkt_id: the packet ID to inject
+ *
+ * This allows the application to repopulate the unacknowledged QoS2 receives
+ * from persistent storage when a session is resumed.
+ */
+LWS_VISIBLE LWS_EXTERN int
+lws_mqtt_client_qos2_rx_add(struct lws *wsi, uint16_t pkt_id);
+
 #endif /* _LWS_MQTT_H */

lib/roles/mqtt/client/client-mqtt.c

@@ -131,6 +131,7 @@ lws_create_client_mqtt_object(const struct lws_client_connect_info *i,
 		lws_free((void *)cp->client_id);
 
 	c->keep_alive_secs = cp->keep_alive;
+	c->qos2_state_ops = cp->qos2_state_ops;
 	c->aws_iot = cp->aws_iot;
 
 	if (cp->will_param.topic &&

lib/roles/mqtt/mqtt.c

@@ -929,6 +929,32 @@ _lws_mqtt_rx_parser(struct lws *wsi, lws_mqtt_parser_t *par,
 			buf += 2;
 			len -= 2;
 			wsi->mqtt->peer_ack_pkt_id = par->cpkt_id;
+			wsi->mqtt->qos2_duplicate = 0;
+
+			if (pub->qos == QOS2) {
+				lws_mqtt_qos2_rx_t *rx;
+
+				lws_start_foreach_dll(struct lws_dll2 *, p, wsi->mqtt->qos2_rx_list.head) {
+					rx = lws_container_of(p, lws_mqtt_qos2_rx_t, list);
+					if (rx->packet_id == par->cpkt_id) {
+						wsi->mqtt->qos2_duplicate = 1;
+						break;
+					}
+				} lws_end_foreach_dll(p);
+
+				if (!wsi->mqtt->qos2_duplicate) {
+					rx = lws_malloc(sizeof(*rx), "qos2 rx");
+					if (rx) {
+						const char *cid = wsi->mqtt->client.id ? (const char *)wsi->mqtt->client.id->buf : "unknown";
+						rx->packet_id = par->cpkt_id;
+						lws_dll2_add_tail(&rx->list, &wsi->mqtt->qos2_rx_list);
+						if (wsi->mqtt->client.qos2_state_ops &&
+						    wsi->mqtt->client.qos2_state_ops->rx_add)
+							wsi->mqtt->client.qos2_state_ops->rx_add(wsi, cid, par->cpkt_id);
+					}
+				}
+			}
+
 			lwsl_debug("%s: Packet ID %d\n",
 					__func__, (int)par->cpkt_id);
 			par->state = LMQCPP_PAYLOAD;
@@ -1471,11 +1497,40 @@ _lws_mqtt_rx_parser(struct lws *wsi, lws_mqtt_parser_t *par,
 				break;
 
 			case LMQCP_PUBREL:
-				lwsl_err("%s: cmd_completion: PUBREL\n",
+			{
+				lws_mqtt_qos2_rx_t *rx;
+
+				lwsl_info("%s: cmd_completion: PUBREL\n",
 						__func__);
+
+				lws_start_foreach_dll_safe(struct lws_dll2 *, p, tp, wsi->mqtt->qos2_rx_list.head) {
+					rx = lws_container_of(p, lws_mqtt_qos2_rx_t, list);
+					if (rx->packet_id == par->cpkt_id) {
+						const char *cid = wsi->mqtt->client.id ? (const char *)wsi->mqtt->client.id->buf : "unknown";
+						lws_dll2_remove(&rx->list);
+						lws_free(rx);
+						if (wsi->mqtt->client.qos2_state_ops &&
+						    wsi->mqtt->client.qos2_state_ops->rx_remove)
+							wsi->mqtt->client.qos2_state_ops->rx_remove(wsi, cid, par->cpkt_id);
+						break;
+					}
+				} lws_end_foreach_dll_safe(p, tp);
+
+				lws_start_foreach_ll(struct lws *, w,
+						      wsi->mux.child_list) {
+					uint16_t pid = par->cpkt_id;
+					if (w->a.protocol->callback(w,
+						    LWS_CALLBACK_MQTT_QOS2_RX_COMPLETE,
+						    w->user_space,
+						    (void *)&pid, 0)) {
+						return 1;
+					}
+				} lws_end_foreach_ll(w, mux.sibling_list);
+
 				wsi->mqtt->send_pubcomp = 1;
 				lws_callback_on_writable(wsi);
 				break;
+			}
 
 			case LMQCP_PUBACK:
 				lwsl_info("%s: cmd_completion: PUBACK\n",
@@ -1671,7 +1726,8 @@ _lws_mqtt_rx_parser(struct lws *wsi, lws_mqtt_parser_t *par,
 
 				lws_start_foreach_ll(struct lws *, w,
 						      wsi->mux.child_list) {
-					if (lws_mqtt_find_sub(w->mqtt,
+					if (!wsi->mqtt->qos2_duplicate &&
+					    lws_mqtt_find_sub(w->mqtt,
 							      pub->topic))
 						if (w->a.protocol->callback(
 							    w, (enum lws_callback_reasons)n,
@@ -2531,3 +2587,22 @@ lws_wsi_mqtt_adopt(struct lws *parent_wsi, struct lws *wsi)
 	return NULL;
 }
 
+int
+lws_mqtt_client_qos2_rx_add(struct lws *wsi, uint16_t pkt_id)
+{
+	struct lws *nwsi = lws_get_network_wsi(wsi);
+	lws_mqtt_qos2_rx_t *rx;
+
+	if (!nwsi || !nwsi->mqtt)
+		return 1;
+
+	rx = lws_malloc(sizeof(*rx), "qos2 rx");
+	if (!rx)
+		return 1;
+
+	rx->packet_id = pkt_id;
+	lws_dll2_add_tail(&rx->list, &nwsi->mqtt->qos2_rx_list);
+
+	return 0;
+}
+

lib/roles/mqtt/ops-mqtt.c

@@ -485,6 +485,17 @@ rops_close_role_mqtt(struct lws_context_per_thread *pt, struct lws *wsi)
 		s = s1;
 	}
 
+	/* clean up QoS2 rx list */
+	{
+		lws_mqtt_qos2_rx_t *rx;
+
+		lws_start_foreach_dll_safe(struct lws_dll2 *, p, tp, wsi->mqtt->qos2_rx_list.head) {
+			rx = lws_container_of(p, lws_mqtt_qos2_rx_t, list);
+			lws_dll2_remove(&rx->list);
+			lws_free(rx);
+		} lws_end_foreach_dll_safe(p, tp);
+	}
+
 	lws_mqtt_publish_param_t *pub =
 			(lws_mqtt_publish_param_t *)
 				wsi->mqtt->rx_cpkt_param;

lib/roles/mqtt/private-lib-roles-mqtt.h

@@ -327,6 +327,11 @@ typedef struct lws_mqtt_subs {
 	char			topic[];
 } lws_mqtt_subs_t;
 
+typedef struct lws_mqtt_qos2_rx {
+	struct lws_dll2		list;
+	uint16_t		packet_id;
+} lws_mqtt_qos2_rx_t;
+
 typedef struct lws_mqtts {
 	lws_mqtt_parser_t	par;
 	lwsgs_mqtt_states_t	estate;
@@ -348,6 +353,7 @@ typedef struct lws_mqttc {
 	} will;
 	uint16_t		keep_alive_secs;
 	uint16_t			conn_flags;
+	const lws_mqtt_qos2_state_ops_t *qos2_state_ops;
 	uint8_t			aws_iot;
 } lws_mqttc_t;
 
@@ -358,6 +364,7 @@ struct _lws_mqtt_related {
 	lws_sorted_usec_list_t	sul_unsuback_wait; /* unsuback wait TO */
 	lws_sorted_usec_list_t	sul_qos2_pubrec_wait; /* QoS2 pubrec wait TO */
 	lws_sorted_usec_list_t	sul_shadow_wait; /* Device Shadow wait TO */
+	struct lws_dll2_owner	qos2_rx_list;
 	struct lws		*wsi; /**< so sul can use lws_container_of */
 	lws_mqtt_subs_t		*subs_head; /**< Linked-list of heap-allocated subscription objects */
 	void			*rx_cpkt_param;
@@ -382,6 +389,8 @@ struct _lws_mqtt_related {
 	uint8_t			unacked_publish:1;
 	uint8_t			unacked_pubrel:1;
 
+	uint8_t			qos2_duplicate:1;
+
 	uint8_t			done_subscribe:1;
 	uint8_t			done_birth:1;
 	uint8_t			inside_shadow:1;

minimal-examples-lowlevel/mqtt-client/minimal-mqtt-client-qos2/CMakeLists.txt

@@ -0,0 +1,31 @@
+project(lws-minimal-mqtt-client-qos2 C)
+cmake_minimum_required(VERSION 3.10)
+include(CheckCSourceCompiles)
+include(LwsCheckRequirements)
+
+set(SAMP lws-minimal-mqtt-client-qos2)
+set(SRCS minimal-mqtt-client-qos2.c)
+
+set(requirements 1)
+require_lws_config(LWS_ROLE_MQTT 1 requirements)
+require_lws_config(LWS_WITH_CLIENT 1 requirements)
+require_lws_config(LWS_WITH_SYS_STATE 1 requirements)
+
+if (requirements)
+	add_executable(${SAMP} ${SRCS})
+
+	if (websockets_shared)
+		target_link_libraries(${SAMP} websockets_shared ${LIBWEBSOCKETS_DEP_LIBS})
+		add_dependencies(${SAMP} websockets_shared)
+	else()
+		target_link_libraries(${SAMP} websockets ${LIBWEBSOCKETS_DEP_LIBS})
+	endif()
+
+	#
+	# requires mosquitto running locally
+	#
+	add_test(NAME mqtt-client-qos2 COMMAND lws-minimal-mqtt-client-qos2 -f)
+	set_tests_properties(mqtt-client-qos2 PROPERTIES
+		WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}/minimal-examples-lowlevel/mqtt-client/minimal-mqtt-client-qos2
+		TIMEOUT 20)
+endif()

minimal-examples-lowlevel/mqtt-client/minimal-mqtt-client-qos2/README.md

@@ -0,0 +1,46 @@
+# lws-minimal-mqtt-client-qos2
+
+This is a minimal example demonstrating MQTT QoS 2 functionality with deduplication and session restoration capabilities.
+
+The client connects to a local Mosquitto broker, subscribes to a topic (`test/topic0`), and publishes a QoS 2 message to that same topic. It tracks the `packet_id` of unacknowledged QoS 2 receives using the operations API.
+
+## Build
+
+```bash
+$ cmake . && make
+```
+
+## Usage
+
+This example requires a local `mosquitto` broker to run against.
+
+### 1. Standard Testing
+Run the client normally to see the full QoS 2 handshake without any interruptions:
+
+```bash
+$ ./lws-minimal-mqtt-client-qos2
+```
+
+The client will successfully connect, subscribe, publish a message, receive it, and log the payload, followed by the normal completion.
+
+### 2. Fault Injection Mode (Session Resumption)
+Run the client with the `-f` flag to simulate a dropped connection during the QoS 2 handshake:
+
+```bash
+$ ./lws-minimal-mqtt-client-qos2 -f
+```
+
+**Sequence of Events:**
+1. The client receives the QoS 2 `PUBLISH` from the broker.
+2. The custom `my_rx_add` callback fires, saving the unacknowledged `packet_id` to the simulated state store.
+3. The client immediately drops its connection (forceful simulation).
+4. The built-in retry policy triggers a reconnection.
+5. On connection establishment, the client injects the stashed `packet_id` back into the library using `lws_mqtt_client_qos2_rx_add`.
+6. Mosquitto resends the `PUBLISH` with `DUP=1`.
+7. `libwebsockets` recognizes the duplicate, drops the payload to maintain the *Exactly-Once* guarantee, and seamlessly resumes the handshake (`PUBREC` -> `PUBREL` -> `PUBCOMP`).
+
+## Commandline Options
+
+- `-d <log level>`: Set the logging level (e.g., `-d 1039`)
+- `-s`: Use TLS / HTTPS
+- `-f`: Simulate connection drop for testing QoS 2 restoration