GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(mqtt): remove mqtt cache db when node deleted

Browse Source
This commit is contained in:
eeff
2024-08-01 15:50:49 +08:00
parent cc84c7f620
commit bd0f786406
4 changed files with 119 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/connection/mqtt_client.h
+20 -19
View File
@@ -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);
plugins/mqtt/mqtt_plugin_intf.c
+10 -2
View File
@@ -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;
src/connection/mqtt_client.c
+77 -18
View File
@@ -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
src/core/manager.c
+12
View File
@@ -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);