* FIX [sqlite] add mqtt_version field for nni_mqtt_sqlite_option;

emqx · Dec 2, 2022 · f0c3723 · f0c3723
1 parent 3ec935a
commit f0c3723
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 12 deletions.
diff --git a/src/mqtt/protocol/mqtt/sqlite_handler.c b/src/mqtt/protocol/mqtt/sqlite_handler.c
@@ -28,11 +28,10 @@ inline void
 sqlite_flush_lmq(nni_mqtt_sqlite_option *sqlite, nni_lmq *lmq)
 {
 	if (sqlite_is_enabled(sqlite)) {
-		nni_mqtt_qos_db_set_client_offline_msg_batch(sqlite->db, lmq,
-		    sqlite->db_name, MQTT_PROTOCOL_VERSION_v311);
-		nni_mqtt_qos_db_remove_oldest_client_offline_msg(sqlite->db,
-		    sqlite->disk_cache_size,
-		    sqlite->db_name);
+		nni_mqtt_qos_db_set_client_offline_msg_batch(
+		    sqlite->db, lmq, sqlite->db_name, sqlite->mqtt_version);
+		nni_mqtt_qos_db_remove_oldest_client_offline_msg(
+		    sqlite->db, sqlite->disk_cache_size, sqlite->db_name);
 	}
 }
 

diff --git a/src/supplemental/mqtt/mqtt_public.c b/src/supplemental/mqtt/mqtt_public.c
@@ -855,6 +855,7 @@ nng_mqtt_alloc_sqlite_opt(nng_mqtt_sqlite_option **sqlite)
 	opt->disk_cache_size     = 102400;
 	opt->mounted_file_path   = NULL;
 	opt->flush_mem_threshold = 100;
+	opt->mqtt_version        = MQTT_PROTOCOL_VERSION_v311;
 
 	*sqlite                  = opt;
 	return (rv);

diff --git a/src/supplemental/mqtt/mqtt_qos_db.c b/src/supplemental/mqtt/mqtt_qos_db.c
@@ -674,9 +674,15 @@ nni_mqtt_sqlite_db_init(
 		sqlite->db_name = nni_strdup(db_name);
 		nni_mqtt_qos_db_init(
 		    &sqlite->db, sqlite->mounted_file_path, db_name);
-		nni_mqtt_qos_db_set_client_info(sqlite->db, db_name, NULL,
-		    "MQTT",
-		    proto_ver == 0 ? MQTT_PROTOCOL_VERSION_v311 : proto_ver);
+
+		if (proto_ver == 0 ||
+		    (proto_ver != MQTT_PROTOCOL_VERSION_v311 &&
+		        proto_ver != MQTT_PROTOCOL_VERSION_v5)) {
+			sqlite->mqtt_version = proto_ver;
+		}
+
+		nni_mqtt_qos_db_set_client_info(
+		    sqlite->db, db_name, NULL, "MQTT", sqlite->mqtt_version);
 	}
 }
 

diff --git a/src/supplemental/mqtt/mqtt_qos_db.h b/src/supplemental/mqtt/mqtt_qos_db.h
@@ -37,10 +37,11 @@
 **/
 
 struct nng_mqtt_sqlite_option {
-	bool   enable;
-	size_t disk_cache_size;   // specify the max rows of sqlite table
-	char * mounted_file_path; // specify the db file path
-	char * db_name;
+	bool    enable;
+	uint8_t mqtt_version;      // mqtt version
+	size_t  disk_cache_size;   // specify the max rows of sqlite table
+	char *  mounted_file_path; // specify the db file path
+	char *  db_name;
 	size_t
 	    flush_mem_threshold; // flush to sqlite table when count of message
 	                         // is equal or greater than this value