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

Merge pull request #1455 from eeff/eeff/dev

Browse Source 
feat(mqtt): add cache-sync-interval setting
This commit is contained in:
fengzeroz
2023-08-16 10:19:23 +08:00
committed by GitHub
parent 42d25570f9 3430e1a14b
commit 7db7a47c10
6 changed files with 133 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/connection/mqtt_client.h
+7
View File
@@ -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);
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);
src/connection/mqtt_client.c
+29
View File
@@ -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))) {