GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-29 15:09:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add function neu_mqtt_client_set_cache_sync_interval

Browse Source
This commit is contained in:
eeff
2023-08-15 17:46:20 +08:00
parent 89096f423c
commit db5ac155cd
2 changed files with 36 additions and 0 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);
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))) {