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

Merge pull request #1210 from eeff/eeff/2.4

Browse Source 
Pick to 2.4 feat(mqtt): make offline cache setting more explicit
This commit is contained in:
eeff
2023-04-20 18:09:29 +08:00
committed by GitHub
parent 7cb61f6fa5 92f387ae7b
commit fc12d7fbff
2 changed files with 105 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/mqtt.json
+20 -4
View File
@@ -57,6 +57,16 @@
]
}
},
"offline-cache": {
"name": "Offline Data Caching",
"name_zh": "离线缓存",
"description": "Offline caching switch. Cache MQTT messages when offline, and sync cached messages when back online.",
"description_zh": "离线缓存开关。连接断开时缓存 MQTT 消息，连接重建时同步缓存的 MQTT 消息到服务器。",
"attribute": "optional",
"type": "bool",
"default": false,
"valid": {}
},
"cache-mem-size": {
"name": "Cache Memory Size (MB)",
"name_zh": "缓存内存大小（MB）",
@@ -64,9 +74,12 @@
"description_zh": "当 MQTT 连接异常时，最大的内存缓存大小（单位：MB）。应该小于缓存磁盘大小。",
"type": "int",
"attribute": "required",
"default": 0,
"condition": {
"field": "offline-cache",
"value": true
},
"valid": {
"min": 0,
"min": 1,
"max": 1024
}
},
@@ -77,9 +90,12 @@
"description_zh": "当 MQTT 连接异常时，最大的磁盘缓存大小（单位：MB）。应该大于缓存内存大小。如果不为 0，缓存内存大小也应该不为 0。",
"type": "int",
"attribute": "required",
"default": 0,
"condition": {
"field": "offline-cache",
"value": true
},
"valid": {
"min": 0,
"min": 1,
"max": 10240
}
},
plugins/mqtt/mqtt_config.c
+85 -33
View File
@@ -115,6 +115,79 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
return 0;
}
static int parse_cache_params(neu_plugin_t *plugin, const char *setting,
neu_json_elem_t *offline_cache,
neu_json_elem_t *cache_mem_size,
neu_json_elem_t *cache_disk_size)
{
int ret = 0;
char *err_param = NULL;
bool flag_present = true; // whether setting has `offline-cache`
// offline-cache flag, optional
ret = neu_parse_param(setting, NULL, 1, offline_cache);
if (0 != ret) {
plog_info(plugin, "setting no offline cache flag");
flag_present = false; // `offline-cache` not present in setting
}
if (flag_present && !offline_cache->v.val_bool) {
// cache explicitly disabled in setting
cache_mem_size->v.val_int = 0;
cache_disk_size->v.val_int = 0;
return 0;
}
// we are here because one of the following:
// 1. `offline-cache` presents to be true in setting
// 2. no `offline-cache` in setting
// we require `cache_mem_size` and `cache_disk_size` to present in both
// cases, especially in case 2 for backward compatibility.
ret = neu_parse_param(setting, &err_param, 2, cache_mem_size,
cache_disk_size);
if (0 != ret) {
plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
free(err_param);
return -1;
}
if (cache_mem_size->v.val_int > 1024 ||
(flag_present && cache_mem_size->v.val_int < 1)) {
plog_error(plugin, "setting invalid cache memory size: %" PRIi64,
cache_mem_size->v.val_int);
return -1;
}
if (cache_disk_size->v.val_int > 10240 ||
(flag_present && cache_disk_size->v.val_int < 1)) {
plog_error(plugin, "setting invalid cache disk size: %" PRIi64,
cache_disk_size->v.val_int);
return -1;
}
if (cache_mem_size->v.val_int > cache_disk_size->v.val_int) {
plog_error(plugin,
"setting cache memory size %" PRIi64
" larger than cache disk size %" PRIi64,
cache_mem_size->v.val_int, cache_disk_size->v.val_int);
return -1;
}
if (0 == cache_mem_size->v.val_int && 0 != cache_disk_size->v.val_int) {
plog_error(plugin,
"setting cache disk size %" PRIi64 " without memory cache",
cache_disk_size->v.val_int);
return -1;
}
if (!flag_present) {
offline_cache->v.val_bool = cache_mem_size->v.val_int > 0;
}
return 0;
}
int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
mqtt_config_t *config)
{
@@ -130,6 +203,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
};
neu_json_elem_t format = { .name = "format", .t = NEU_JSON_INT };
neu_json_elem_t offline_cache = { .name = "offline-cache",
.t = NEU_JSON_BOOL };
neu_json_elem_t cache_mem_size = { .name = "cache-mem-size",
.t = NEU_JSON_INT };
neu_json_elem_t cache_disk_size = { .name = "cache-disk-size",
@@ -149,8 +224,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
return -1;
}
ret = neu_parse_param(setting, &err_param, 7, &client_id, &format, &qos,
&cache_mem_size, &cache_disk_size, &host, &port);
ret = neu_parse_param(setting, &err_param, 5, &client_id, &format, &qos,
&host, &port);
if (0 != ret) {
plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
goto error;
@@ -176,32 +251,10 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
goto error;
}
// cache memory size, required
if (cache_mem_size.v.val_int > 1024) {
plog_error(plugin, "setting invalid cache memory size: %" PRIi64,
cache_mem_size.v.val_int);
goto error;
}
// cache disk size, required
if (cache_disk_size.v.val_int > 10240) {
plog_error(plugin, "setting invalid cache disk size: %" PRIi64,
cache_disk_size.v.val_int);
goto error;
}
if (cache_mem_size.v.val_int > cache_disk_size.v.val_int) {
plog_error(plugin,
"setting cache memory size %" PRIi64
" larger than cache disk size %" PRIi64,
cache_mem_size.v.val_int, cache_disk_size.v.val_int);
goto error;
}
if (0 == cache_mem_size.v.val_int && 0 != cache_disk_size.v.val_int) {
plog_error(plugin,
"setting cache disk size %" PRIi64 " without memory cache",
cache_disk_size.v.val_int);
// offline cache
ret = parse_cache_params(plugin, setting, &offline_cache, &cache_mem_size,
&cache_disk_size);
if (0 != ret) {
goto error;
}
@@ -234,11 +287,10 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
goto error;
}
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
config->format = format.v.val_int;
config->cache =
0 != cache_mem_size.v.val_int || 0 != cache_disk_size.v.val_int;
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
config->format = format.v.val_int;
config->cache = offline_cache.v.val_bool;
config->cache_mem_size = cache_mem_size.v.val_int * MB;
config->cache_disk_size = cache_disk_size.v.val_int * MB;
config->host = host.v.val_str;