GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(mqtt): add cache-sync-interval setting

Browse Source
This commit is contained in:
eeff
2023-08-15 17:34:00 +08:00
parent db5ac155cd
commit 3430e1a14b
4 changed files with 97 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/mqtt.json
+17
View File
@@ -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": "服务器地址",
plugins/mqtt/mqtt_config.c
+54 -33
View File
@@ -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);
plugins/mqtt/mqtt_config.h
+18 -17
View File
@@ -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,
plugins/mqtt/mqtt_plugin.c
+8
View File
@@ -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);