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);