From 722632f6d6232021e8bf12a0ead5c4c1dc55d3bc Mon Sep 17 00:00:00 2001 From: gc87 Date: Mon, 29 Aug 2022 21:27:24 +0800 Subject: [PATCH] feat(mqtt): add cache setting to ui & webapi --- include/neuron/connection/mqtt_client_intf.h | 1 + plugins/mqtt/mqtt.json | 11 ++++++++ plugins/mqtt/mqtt_plugin.c | 23 ++++++++++------ plugins/mqtt/mqtt_plugin.h | 1 - plugins/mqtt/mqtt_util.c | 14 ++++++++++ src/utils/mem_cache.c | 15 ++++++---- tests/mqtt_test.cc | 29 ++++++++++++++++++++ 7 files changed, 79 insertions(+), 15 deletions(-) diff --git a/include/neuron/connection/mqtt_client_intf.h b/include/neuron/connection/mqtt_client_intf.h index 4d8fad73..786e3a44 100644 --- a/include/neuron/connection/mqtt_client_intf.h +++ b/include/neuron/connection/mqtt_client_intf.h @@ -61,6 +61,7 @@ typedef struct { int format; char * upload_topic; char * heartbeat_topic; + size_t cache; char * group_id; state_update state_update_func; zlog_category_t *log; diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index b5768950..a1ec4165 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -38,6 +38,17 @@ ] } }, + "cache": { + "name": "cache size(MB)", + "description": "Upload data memory cache", + "type": "int", + "attribute": "optional", + "default": 64, + "valid": { + "min": 1, + "max": 256 + } + }, "ssl": { "name": "ssl", "description": "Enable SSL connection", diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index dfe049b9..dc711df1 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -57,6 +57,7 @@ struct mqtt_routine { neu_mqtt_option_t option; neu_mqtt_client_t client; UT_array * topics; + neu_mem_cache_t * cache; nng_mtx * mutex; neu_events_t * events; neu_event_timer_t *send; @@ -176,7 +177,7 @@ static int mqtt_routine_cache_send(void *data) // size_t used_bytes = 0; // size_t used_items = 0; - // neu_mem_cache_used(routine->plugin->cache, &used_bytes, &used_items); + // neu_mem_cache_used(routine->cache, &used_bytes, &used_items); // plog_info(routine->plugin, "used bytes:%lu, used items:%lu", used_bytes, // used_items); @@ -186,7 +187,7 @@ static int mqtt_routine_cache_send(void *data) } nng_mtx_lock(routine->mutex); - cache_item_t item = neu_mem_cache_earliest(routine->plugin->cache); + cache_item_t item = neu_mem_cache_earliest(routine->cache); if (NULL == item.data) { nng_mtx_unlock(routine->mutex); return -1; @@ -299,6 +300,10 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, topics_subscribe(routine->topics, routine->client); // cache send + const size_t max_bytes = routine->option.cache * 1024 * 1024; + const size_t max_items = 0; + routine->cache = neu_mem_cache_create(max_bytes, max_items); + nng_mtx_alloc(&routine->mutex); routine->events = neu_event_new(); neu_event_timer_param_t param = { @@ -320,6 +325,7 @@ static void mqtt_routine_stop(mqtt_routine_t *routine) neu_event_del_timer(routine->events, routine->send); neu_event_close(routine->events); nng_mtx_free(routine->mutex); + neu_mem_cache_destroy(routine->cache); // stop recevied neu_mqtt_client_suspend(routine->client); @@ -360,11 +366,6 @@ static int mqtt_plugin_init(neu_plugin_t *plugin) plugin->running = false; plugin->config = NULL; - // cache setting - const size_t max_bytes = 1024; - const size_t max_items = 1; - plugin->cache = neu_mem_cache_create(max_bytes, max_items); - const char *name = neu_plugin_module.module_name; plog_info(plugin, "initialize plugin: %s", name); return NEU_ERR_SUCCESS; @@ -423,7 +424,6 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin) plugin_stop_running(plugin); plugin_config_free(plugin); - neu_mem_cache_destroy(plugin->cache); const char *name = neu_plugin_module.module_name; plog_info(plugin, "uninitialize plugin: %s", name); @@ -570,8 +570,13 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data) }; nng_mtx_lock(routine->mutex); - neu_mem_cache_add(plugin->cache, &item); + rc = neu_mem_cache_add(routine->cache, &item); nng_mtx_unlock(routine->mutex); + + if (0 != rc) { + item.release(item.data); + } + return NEU_ERR_MQTT_CONNECT_FAILURE; } diff --git a/plugins/mqtt/mqtt_plugin.h b/plugins/mqtt/mqtt_plugin.h index 9becf55b..8bcfccf8 100644 --- a/plugins/mqtt/mqtt_plugin.h +++ b/plugins/mqtt/mqtt_plugin.h @@ -35,7 +35,6 @@ struct neu_plugin { bool running; mqtt_routine_t * routine; char * config; - neu_mem_cache_t * cache; }; #ifdef __cplusplus diff --git a/plugins/mqtt/mqtt_util.c b/plugins/mqtt/mqtt_util.c index cdffdb98..dc91ef03 100644 --- a/plugins/mqtt/mqtt_util.c +++ b/plugins/mqtt/mqtt_util.c @@ -117,6 +117,7 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, neu_json_elem_t heartbeat = { .name = "heartbeat-topic", .t = NEU_JSON_STR }; neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT }; + neu_json_elem_t cache = { .name = "cache", .t = NEU_JSON_INT }; neu_json_elem_t ssl = { .name = "ssl", .t = NEU_JSON_BOOL }; neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR }; neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT }; @@ -175,6 +176,19 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config, option->format = format.v.val_int; } + // cache, optional + ret = neu_parse_param(config, &error, 1, &cache); + if (0 != ret) { + free(error); + option->cache = 64; + } else { + if (1 > cache.v.val_int || 265 < cache.v.val_int) { + option->cache = 64; + } else { + option->cache = cache.v.val_int; + } + } + // ssl, optional ret = neu_parse_param(config, &error, 1, &ssl); if (0 != ret) { diff --git a/src/utils/mem_cache.c b/src/utils/mem_cache.c index 14e9d5b4..a3242d78 100644 --- a/src/utils/mem_cache.c +++ b/src/utils/mem_cache.c @@ -55,18 +55,23 @@ int neu_mem_cache_add(neu_mem_cache_t *cache, cache_item_t *item) assert(NULL != cache); assert(NULL != item); - element *el = calloc(1, sizeof(element)); - if (NULL == el) { - return -1; - } - while (queue_is_full(cache, item)) { cache_item_t remove_item = neu_mem_cache_earliest(cache); + if (NULL == remove_item.data) { + // size(item.data) > cache->max_bytes + return -1; + } + if (NULL != remove_item.release && NULL != remove_item.data) { remove_item.release(remove_item.data); } } + element *el = calloc(1, sizeof(element)); + if (NULL == el) { + return -1; + } + el->item = *item; DL_PREPEND(cache->head, el); diff --git a/tests/mqtt_test.cc b/tests/mqtt_test.cc index 0373a64d..e38aaaaf 100644 --- a/tests/mqtt_test.cc +++ b/tests/mqtt_test.cc @@ -213,6 +213,35 @@ TEST(MQTTTest, mqtt_option_init_by_config) "\"cert\":\"ttttt\", \"key\":\"tttt\", \"keypass\":\"\" }}"; rc = mqtt_option_init(plugin, config, &option); EXPECT_EQ(0, rc); + EXPECT_EQ(option.cache, 64); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + config = + (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"cache\": 100, \"ssl\":true, " + "\"host\":\"192.168.10.116\", \"port\":1883, " + "\"username\":\"\", \"password\":\"\", \"ca\":\"tttttt\", " + "\"cert\":\"ttttt\", \"key\":\"tttt\", \"keypass\":\"\" }}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(0, rc); + EXPECT_EQ(option.cache, 100); + mqtt_option_uninit(plugin, &option); + + memset(&option, 0, sizeof(neu_mqtt_option_t)); + config = + (char *) "{\"node_id\":5,\"params\":{\"upload-topic\":\"/neuron/" + "node_name/upload\", " + "\"heartbeat-topic\":\"/neuron/node_name/heartbeat\", " + "\"format\":1, \"cache\": -100, \"ssl\":true, " + "\"host\":\"192.168.10.116\", \"port\":1883, " + "\"username\":\"\", \"password\":\"\", \"ca\":\"tttttt\", " + "\"cert\":\"ttttt\", \"key\":\"tttt\", \"keypass\":\"\" }}"; + rc = mqtt_option_init(plugin, config, &option); + EXPECT_EQ(0, rc); + EXPECT_EQ(option.cache, 64); mqtt_option_uninit(plugin, &option); free(plugin);