From 92f387ae7b62e745e91ea34dd0e82d59e4391916 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Thu, 20 Apr 2023 16:40:16 +0800 Subject: [PATCH] feat(mqtt): make offline cache setting more explicit 1. Add `offline-cache` bool parameter. 2. Require `cache-mem-size` and `cache-disk-size` only when `offline-cache` is enabled. We provide no defaults for `cache-mem-size` and `cache-disk-size`, so the user must provide values explicitly. --- plugins/mqtt/mqtt.json | 24 ++++++-- plugins/mqtt/mqtt_config.c | 118 ++++++++++++++++++++++++++----------- 2 files changed, 105 insertions(+), 37 deletions(-) diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index 495f281e..1cbe961c 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -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 } }, diff --git a/plugins/mqtt/mqtt_config.c b/plugins/mqtt/mqtt_config.c index 1b0e9e6a..4fe1de24 100644 --- a/plugins/mqtt/mqtt_config.c +++ b/plugins/mqtt/mqtt_config.c @@ -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;