From bd0f786406d534ef9c60a45a345e34cf063f6389 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Thu, 1 Aug 2024 15:50:49 +0800 Subject: [PATCH] feat(mqtt): remove mqtt cache db when node deleted --- include/neuron/connection/mqtt_client.h | 39 +++++----- plugins/mqtt/mqtt_plugin_intf.c | 12 +++- src/connection/mqtt_client.c | 95 ++++++++++++++++++++----- src/core/manager.c | 12 ++++ 4 files changed, 119 insertions(+), 39 deletions(-) diff --git a/include/neuron/connection/mqtt_client.h b/include/neuron/connection/mqtt_client.h index 91641db5..135de534 100644 --- a/include/neuron/connection/mqtt_client.h +++ b/include/neuron/connection/mqtt_client.h @@ -85,25 +85,26 @@ bool neu_mqtt_client_is_open(neu_mqtt_client_t *client); bool neu_mqtt_client_is_connected(neu_mqtt_client_t *client); size_t neu_mqtt_client_get_cached_msgs_num(neu_mqtt_client_t *client); -int neu_mqtt_client_set_addr(neu_mqtt_client_t *client, const char *host, - uint16_t port); -int neu_mqtt_client_set_id(neu_mqtt_client_t *client, const char *id); -int neu_mqtt_client_set_will_msg(neu_mqtt_client_t *client, const char *topic, - uint8_t *msg, uint32_t len, bool retain, - uint8_t qos); -int neu_mqtt_client_set_user(neu_mqtt_client_t *client, const char *username, - const char *password); -int neu_mqtt_client_set_connect_cb(neu_mqtt_client_t * client, - neu_mqtt_client_connection_cb_t cb, - void * data); -int neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t * client, - neu_mqtt_client_connection_cb_t cb, - void * data); -int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled, - const char *ca, const char *cert, const char *key, - const char *keypass); -int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client, - size_t mem_size_bytes, size_t db_size_bytes); +int neu_mqtt_client_set_addr(neu_mqtt_client_t *client, const char *host, + uint16_t port); +int neu_mqtt_client_set_id(neu_mqtt_client_t *client, const char *id); +int neu_mqtt_client_set_will_msg(neu_mqtt_client_t *client, const char *topic, + uint8_t *msg, uint32_t len, bool retain, + uint8_t qos); +int neu_mqtt_client_set_user(neu_mqtt_client_t *client, const char *username, + const char *password); +int neu_mqtt_client_set_connect_cb(neu_mqtt_client_t * client, + neu_mqtt_client_connection_cb_t cb, + void * data); +int neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t * client, + neu_mqtt_client_connection_cb_t cb, + void * data); +int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled, + const char *ca, const char *cert, const char *key, + const char *keypass); +void neu_mqtt_client_remove_cache_db(neu_mqtt_client_t *client); +int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client, + size_t mem_size_bytes, size_t db_size_bytes); // default to NEU_MQTT_CACHE_SYNC_INTERVAL_DEFAULT if not set int neu_mqtt_client_set_cache_sync_interval(neu_mqtt_client_t *client, uint32_t interval); diff --git a/plugins/mqtt/mqtt_plugin_intf.c b/plugins/mqtt/mqtt_plugin_intf.c index 8fed2b82..66ae96de 100644 --- a/plugins/mqtt/mqtt_plugin_intf.c +++ b/plugins/mqtt/mqtt_plugin_intf.c @@ -402,9 +402,17 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, case NEU_REQ_UPDATE_NODE: error = handle_update_driver(plugin, data); break; - case NEU_REQRESP_NODE_DELETED: - error = handle_del_driver(plugin, data); + case NEU_REQRESP_NODE_DELETED: { + neu_reqresp_node_deleted_t *req = data; + if (0 != strcmp(plugin->common.name, req->node)) { + error = handle_del_driver(plugin, data); + } else { + if (plugin->client) { + neu_mqtt_client_remove_cache_db(plugin->client); + } + } break; + } default: error = NEU_ERR_MQTT_FAILURE; break; diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c index 75e3fe9b..93518edd 100644 --- a/src/connection/mqtt_client.c +++ b/src/connection/mqtt_client.c @@ -123,6 +123,7 @@ struct neu_mqtt_client_s { neu_mqtt_client_connection_cb_t disconnect_cb; void * disconnect_cb_data; nng_mqtt_sqlite_option * sqlite_cfg; + char * db; bool receiving; nng_aio * recv_aio; subscription_t * subscriptions; @@ -901,26 +902,10 @@ static inline nng_mqtt_sqlite_option * alloc_sqlite_config(neu_mqtt_client_t *client) { int rv; - char * db = NULL; nng_mqtt_sqlite_option *cfg = NULL; - const mqtt_buf client_id = - nng_mqtt_msg_get_connect_client_id(client->conn_msg); - - if (NULL == client_id.buf || 0 == client_id.length) { - log(error, "nng_mqtt_msg_get_connect_client_id fail"); - return NULL; - } - - neu_asprintf(&db, "neuron-mqtt-client-%.*s.db", (unsigned) client_id.length, - (char *) client_id.buf); - if (NULL == db) { - log(error, "neu_asprintf nano-mqtt cilent db fail"); - return NULL; - } if ((rv = nng_mqtt_alloc_sqlite_opt(&cfg)) != 0) { log(error, "nng_mqtt_alloc_sqlite_opt fail: %s", nng_strerror(rv)); - free(db); return NULL; } @@ -928,9 +913,7 @@ alloc_sqlite_config(neu_mqtt_client_t *client) nng_mqtt_set_sqlite_flush_threshold(cfg, 256); nng_mqtt_set_sqlite_max_rows(cfg, 256); nng_mqtt_set_sqlite_db_dir(cfg, "persistence/"); - nng_mqtt_sqlite_db_init(cfg, db, client->version); - free(db); return cfg; } @@ -989,6 +972,7 @@ void neu_mqtt_client_free(neu_mqtt_client_t *client) subscriptions_free(client->subscriptions); tasks_free(client->task_free_list); nng_msg_free(client->conn_msg); + free(client->db); free(client->url); free(client->host); nng_mtx_free(client->mtx); @@ -1213,6 +1197,76 @@ end: return rv; } +static void rand_str(char *dst, size_t len) +{ + char alphabet[] = "0123456789" + "abcdefghijklmnopqrstuvwxyz" + "ABCDEFGHIJKLMNOPQRSTUVWXYZ"; + + while (len-- > 1) { + size_t i = (double) rand() / RAND_MAX * ((sizeof alphabet) - 1); + *dst++ = alphabet[i]; + } + *dst = '\0'; +} + +static int neu_mqtt_client_set_cache_db(neu_mqtt_client_t *client) +{ + char str[8] = { 0 }; + rand_str(str, sizeof(str)); + + char *db = NULL; + neu_asprintf(&db, "neuron-mqtt-client-%s.db", str); + if (NULL == db) { + log(error, "neu_asprintf nano-mqtt cilent db fail"); + return NEU_ERR_EINTERNAL; + } + + if (client->db && 0 == strcmp(client->db, db)) { + free(db); + return 0; + } + + neu_mqtt_client_remove_cache_db(client); + + free(client->db); + client->db = db; + + if (client->sqlite_cfg) { + nng_mqtt_sqlite_db_init(client->sqlite_cfg, db, client->version); + } + + log(notice, "set cache db %s", db); + + return 0; +} + +void neu_mqtt_client_remove_cache_db(neu_mqtt_client_t *client) +{ + if (NULL == client->db) { + return; + } + + const char *dir = "persistence"; + int path_size = strlen(dir) + strlen(client->db) + 6; + char * path = calloc(path_size, 1); + if (NULL == path) { + return; + } + + snprintf(path, path_size, "%s/%s", dir, client->db); + log(notice, "rm cache db %s", path); + remove(path); + snprintf(path, path_size, "%s/%s-shm", dir, client->db); + log(notice, "rm cache db %s", path); + remove(path); + snprintf(path, path_size, "%s/%s-wal", dir, client->db); + log(notice, "rm cache db %s", path); + remove(path); + + free(path); +} + int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client, size_t mem_size_bytes, size_t db_size_bytes) { @@ -1227,6 +1281,7 @@ int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client, if (client->sqlite_cfg) { nng_mqtt_free_sqlite_opt(client->sqlite_cfg); client->sqlite_cfg = NULL; + neu_mqtt_client_remove_cache_db(client); } goto end; } @@ -1240,6 +1295,10 @@ int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client, } } + if (0 != (rv = neu_mqtt_client_set_cache_db(client))) { + return rv; + } + // FIXME: until NanoSDK cache size limit feature out // Assume that, // 1. every upload message follows the below template diff --git a/src/core/manager.c b/src/core/manager.c index ebaff7a4..43921057 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -801,10 +801,22 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) } utarray_free(subscriptions); } + neu_reqresp_node_deleted_t resp = { 0 }; strcpy(resp.node, header->receiver); + // notify MQTT about node removal + if (0 == strcmp(adapter->module->module_name, "MQTT")) { + msg = neu_msg_new(NEU_REQRESP_NODE_DELETED, NULL, &resp); + if (NULL != msg) { + neu_reqresp_head_t *hd = neu_msg_get_header(msg); + strcpy(hd->receiver, cmd->node); + strcpy(hd->sender, "manager"); + reply(manager, hd, &resp); + } + } header->type = NEU_REQ_NODE_UNINIT; forward_msg(manager, header, header->receiver); + if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_DRIVER) { UT_array *apps = neu_subscribe_manager_find_by_driver( manager->subscribe_manager, resp.node);