GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-23 18:37:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(mqtt): add cache setting to ui & webapi

Browse Source
This commit is contained in:
gc87
2022-08-29 21:27:24 +08:00
parent cd3c9213f6
commit 722632f6d6
7 changed files with 79 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/connection/mqtt_client_intf.h
+1
View File
@@ -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;
plugins/mqtt/mqtt.json
+11
View File
@@ -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",
plugins/mqtt/mqtt_plugin.c
+14 -9
View File
@@ -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;
}
plugins/mqtt/mqtt_plugin.h
-1
View File
@@ -35,7 +35,6 @@ struct neu_plugin {
bool running;
mqtt_routine_t * routine;
char * config;
neu_mem_cache_t * cache;
};
#ifdef __cplusplus
plugins/mqtt/mqtt_util.c
+14
View File
@@ -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) {
src/utils/mem_cache.c
+10 -5
View File
@@ -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);
tests/mqtt_test.cc
+29
View File
@@ -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);