From db5ac155cd6226111e2bc457e4f43c0b6a48cc7b Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 15 Aug 2023 16:32:38 +0800 Subject: [PATCH 1/2] feat: add function neu_mqtt_client_set_cache_sync_interval --- include/neuron/connection/mqtt_client.h | 7 ++++++ src/connection/mqtt_client.c | 29 +++++++++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/include/neuron/connection/mqtt_client.h b/include/neuron/connection/mqtt_client.h index d5bfc048..3ecd0c80 100644 --- a/include/neuron/connection/mqtt_client.h +++ b/include/neuron/connection/mqtt_client.h @@ -35,6 +35,10 @@ extern "C" { #define NEU_METRIC_CACHED_MSGS_NUM_TYPE NEU_METRIC_TYPE_GAUAGE #define NEU_METRIC_CACHED_MSGS_NUM_HELP "Number of messages cached" +#define NEU_MQTT_CACHE_SYNC_INTERVAL_MIN 10 +#define NEU_MQTT_CACHE_SYNC_INTERVAL_MAX 12000 +#define NEU_MQTT_CACHE_SYNC_INTERVAL_DEFAULT 100 + typedef enum { NEU_MQTT_VERSION_V31 = 3, NEU_MQTT_VERSION_V311 = 4, @@ -83,6 +87,9 @@ int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled, 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); +// 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); int neu_mqtt_client_set_zlog_category(neu_mqtt_client_t *client, zlog_category_t * cat); diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c index b8765676..94cee0f7 100644 --- a/src/connection/mqtt_client.c +++ b/src/connection/mqtt_client.c @@ -112,6 +112,7 @@ struct neu_mqtt_client_s { char * url; nng_tls_config * tls_cfg; nng_msg * conn_msg; + nng_duration retry; bool open; bool connected; neu_mqtt_client_connection_cb_t connect_cb; @@ -844,6 +845,7 @@ neu_mqtt_client_t *neu_mqtt_client_new(neu_mqtt_version_e version) } client->version = version; + client->retry = NEU_MQTT_CACHE_SYNC_INTERVAL_DEFAULT; client->task_limit = 1024; return client; @@ -1113,6 +1115,26 @@ end: return rv; } +int neu_mqtt_client_set_cache_sync_interval(neu_mqtt_client_t *client, + uint32_t interval) +{ + int rv = 0; + + nng_mtx_lock(client->mtx); + return_failure_if_open(); + + // sync interval should be within reasonable range + if (NEU_MQTT_CACHE_SYNC_INTERVAL_MIN <= interval && + interval <= NEU_MQTT_CACHE_SYNC_INTERVAL_MAX) { + client->retry = interval; + } else { + rv = -1; + } + + nng_mtx_unlock(client->mtx); + return rv; +} + int neu_mqtt_client_set_zlog_category(neu_mqtt_client_t *client, zlog_category_t * cat) { @@ -1156,6 +1178,13 @@ int neu_mqtt_client_open(neu_mqtt_client_t *client) goto error; } + if ((rv = nng_socket_set_ms(client->sock, NNG_OPT_MQTT_RETRY_INTERVAL, + client->retry)) != 0) { + log(error, "nng_socket_set_ms(%" PRIu32 ") fail: %s", client->retry, + nng_strerror(rv)); + goto error; + } + if (client->sqlite_cfg && (rv = nng_socket_set_ptr(client->sock, NNG_OPT_MQTT_SQLITE, client->sqlite_cfg))) { From 3430e1a14b550debbb4928053cb58698d4140e7b Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 15 Aug 2023 17:34:00 +0800 Subject: [PATCH 2/2] feat(mqtt): add `cache-sync-interval` setting --- plugins/mqtt/mqtt.json | 17 ++++++++ plugins/mqtt/mqtt_config.c | 87 +++++++++++++++++++++++--------------- plugins/mqtt/mqtt_config.h | 35 +++++++-------- plugins/mqtt/mqtt_plugin.c | 8 ++++ 4 files changed, 97 insertions(+), 50 deletions(-) diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index 8333c359..05dab01f 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -123,6 +123,23 @@ "max": 10240 } }, + "cache-sync-interval": { + "name": "Cache Sync Interval (MS)", + "name_zh": "缓存消息重传间隔（MS）", + "description": "When MQTT connection restores, cached messages will be synchronised to the broker. This controls the interval in milliseconds between each message to synchronise.", + "description_zh": "当 MQTT 连接重建后，缓存的消息会被同步到服务器。此配置项控制消息重传时每条消息之间的时间间隔，以毫秒为单位。", + "type": "int", + "attribute": "required", + "condition": { + "field": "offline-cache", + "value": true + }, + "default": 1000, + "valid": { + "min": 10, + "max": 120000 + } + }, "host": { "name": "Broker Host", "name_zh": "服务器地址", diff --git a/plugins/mqtt/mqtt_config.c b/plugins/mqtt/mqtt_config.c index 41630795..457a073b 100644 --- a/plugins/mqtt/mqtt_config.c +++ b/plugins/mqtt/mqtt_config.c @@ -124,7 +124,8 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting, 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) + neu_json_elem_t *cache_disk_size, + neu_json_elem_t *cache_sync_interval) { int ret = 0; char *err_param = NULL; @@ -139,8 +140,9 @@ static int parse_cache_params(neu_plugin_t *plugin, const char *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; + cache_mem_size->v.val_int = 0; + cache_disk_size->v.val_int = 0; + cache_sync_interval->v.val_int = 0; return 0; } @@ -191,6 +193,20 @@ static int parse_cache_params(neu_plugin_t *plugin, const char *setting, offline_cache->v.val_bool = cache_mem_size->v.val_int > 0; } + // cache-sync-interval, optional for backward compatibility + ret = neu_parse_param(setting, NULL, 1, cache_sync_interval); + if (0 == ret) { + if (cache_sync_interval->v.val_int < NEU_MQTT_CACHE_SYNC_INTERVAL_MIN || + NEU_MQTT_CACHE_SYNC_INTERVAL_MAX < cache_sync_interval->v.val_int) { + plog_error(plugin, "setting invalid cache sync interval: %" PRIi64, + cache_sync_interval->v.val_int); + return -1; + } + } else { + plog_notice(plugin, "setting no cache sync interval"); + cache_sync_interval->v.val_int = NEU_MQTT_CACHE_SYNC_INTERVAL_DEFAULT; + } + return 0; } @@ -221,21 +237,23 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting, .v.val_str = NULL, .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility }; - neu_json_elem_t offline_cache = { .name = "offline-cache", + neu_json_elem_t offline_cache = { .name = "offline-cache", .t = NEU_JSON_BOOL }; - neu_json_elem_t cache_mem_size = { .name = "cache-mem-size", + 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", + neu_json_elem_t cache_disk_size = { .name = "cache-disk-size", .t = NEU_JSON_INT }; - neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR }; - neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT }; - neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR }; - neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR }; - neu_json_elem_t ssl = { .name = "ssl", .t = NEU_JSON_BOOL }; - neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR }; - neu_json_elem_t cert = { .name = "cert", .t = NEU_JSON_STR }; - neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR }; - neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR }; + neu_json_elem_t cache_sync_interval = { .name = "cache-sync-interval", + .t = NEU_JSON_INT }; + neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR }; + neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT }; + neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR }; + neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR }; + neu_json_elem_t ssl = { .name = "ssl", .t = NEU_JSON_BOOL }; + neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR }; + neu_json_elem_t cert = { .name = "cert", .t = NEU_JSON_STR }; + neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR }; + neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR }; if (NULL == setting || NULL == config) { plog_error(plugin, "invalid argument, null pointer"); @@ -287,7 +305,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting, // offline cache ret = parse_cache_params(plugin, setting, &offline_cache, &cache_mem_size, - &cache_disk_size); + &cache_disk_size, &cache_sync_interval); if (0 != ret) { goto error; } @@ -321,23 +339,24 @@ 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->write_req_topic = write_req_topic.v.val_str; - config->write_resp_topic = write_resp_topic.v.val_str; - 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; - config->port = port.v.val_int; - config->username = username.v.val_str; - config->password = password.v.val_str; - config->ssl = ssl.v.val_bool; - config->ca = ca.v.val_str; - config->cert = cert.v.val_str; - config->key = key.v.val_str; - config->keypass = keypass.v.val_str; + config->client_id = client_id.v.val_str; + config->qos = qos.v.val_int; + config->format = format.v.val_int; + config->write_req_topic = write_req_topic.v.val_str; + config->write_resp_topic = write_resp_topic.v.val_str; + 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->cache_sync_interval = cache_sync_interval.v.val_int; + config->host = host.v.val_str; + config->port = port.v.val_int; + config->username = username.v.val_str; + config->password = password.v.val_str; + config->ssl = ssl.v.val_bool; + config->ca = ca.v.val_str; + config->cert = cert.v.val_str; + config->key = key.v.val_str; + config->keypass = keypass.v.val_str; plog_notice(plugin, "config client-id : %s", config->client_id); plog_notice(plugin, "config qos : %d", config->qos); @@ -350,6 +369,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting, plog_notice(plugin, "config cache-mem-size : %zu", config->cache_mem_size); plog_notice(plugin, "config cache-disk-size : %zu", config->cache_disk_size); + plog_notice(plugin, "config cache-sync-interval : %zu", + config->cache_sync_interval); plog_notice(plugin, "config host : %s", config->host); plog_notice(plugin, "config port : %" PRIu16, config->port); diff --git a/plugins/mqtt/mqtt_config.h b/plugins/mqtt/mqtt_config.h index 81270597..ec6e68e2 100644 --- a/plugins/mqtt/mqtt_config.h +++ b/plugins/mqtt/mqtt_config.h @@ -48,23 +48,24 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f) } typedef struct { - char * client_id; // client id - neu_mqtt_qos_e qos; // message QoS - mqtt_upload_format_e format; // upload format - char * write_req_topic; // write request topic - char * write_resp_topic; // write response topic - size_t cache; // cache enable flag - size_t cache_mem_size; // cache memory size in bytes - size_t cache_disk_size; // cache disk size in bytes - char * host; // broker host - uint16_t port; // broker port - char * username; // user name - char * password; // user password - bool ssl; // ssl flag - char * ca; // CA - char * cert; // client cert - char * key; // client key - char * keypass; // client key password + char * client_id; // client id + neu_mqtt_qos_e qos; // message QoS + mqtt_upload_format_e format; // upload format + char * write_req_topic; // write request topic + char * write_resp_topic; // write response topic + size_t cache; // cache enable flag + size_t cache_mem_size; // cache memory size in bytes + size_t cache_disk_size; // cache disk size in bytes + size_t cache_sync_interval; // cache sync interval + char * host; // broker host + uint16_t port; // broker port + char * username; // user name + char * password; // user password + bool ssl; // ssl flag + char * ca; // CA + char * cert; // client cert + char * key; // client key + char * keypass; // client key password } mqtt_config_t; int mqtt_config_parse(neu_plugin_t *plugin, const char *setting, diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index a164b6b7..5ad1ca5c 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -137,6 +137,14 @@ static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client, return -1; } + rv = neu_mqtt_client_set_cache_sync_interval(client, + config->cache_sync_interval); + if (0 != rv) { + plog_error(plugin, + "neu_mqtt_client_set_msg_cache_upload_interval fail"); + return -1; + } + if (NULL != config->username) { rv = neu_mqtt_client_set_user(client, config->username, config->password);