diff --git a/include/neuron/connection/mqtt_client.h b/include/neuron/connection/mqtt_client.h index dfa2f1f4..dd55f2c7 100644 --- a/include/neuron/connection/mqtt_client.h +++ b/include/neuron/connection/mqtt_client.h @@ -30,6 +30,11 @@ extern "C" { #include "utils/zlog.h" +// number of messages cached +#define NEU_METRIC_CACHED_MSGS_NUM "cached_msgs" +#define NEU_METRIC_CACHED_MSGS_NUM_TYPE NEU_METRIC_TYPE_GAUAGE +#define NEU_METRIC_CACHED_MSGS_NUM_HELP "Number of messages cached" + typedef enum { NEU_MQTT_VERSION_V31 = 3, NEU_MQTT_VERSION_V311 = 4, @@ -58,8 +63,9 @@ neu_mqtt_client_t *neu_mqtt_client_from_addr(const char *host, uint16_t port, neu_mqtt_version_e version); void neu_mqtt_client_free(neu_mqtt_client_t *client); -bool neu_mqtt_client_is_opened(neu_mqtt_client_t *client); -bool neu_mqtt_client_is_connected(neu_mqtt_client_t *client); +bool neu_mqtt_client_is_opened(neu_mqtt_client_t *client); +bool neu_mqtt_client_is_connected(neu_mqtt_client_t *client); +size_t neu_mqtt_client_get_cached_msgs_num(neu_mqtt_client_t *client); int neu_mqtt_client_set_addr(neu_mqtt_client_t *client, const char *host, uint16_t port); diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c index 834dd52f..eb4ffe57 100644 --- a/src/connection/mqtt_client.c +++ b/src/connection/mqtt_client.c @@ -638,7 +638,7 @@ alloc_sqlite_config(neu_mqtt_client_t *client) } nng_mqtt_set_sqlite_enable(cfg, true); - nng_mqtt_set_sqlite_flush_threshold(cfg, 256); + nng_mqtt_set_sqlite_flush_threshold(cfg, 1); nng_mqtt_set_sqlite_max_rows(cfg, 256); nng_mqtt_set_sqlite_db_dir(cfg, "persistence/"); nng_mqtt_sqlite_db_init(cfg, db, client->version); @@ -710,6 +710,15 @@ bool neu_mqtt_client_is_connected(neu_mqtt_client_t *client) return client->connected; } +size_t neu_mqtt_client_get_cached_msgs_num(neu_mqtt_client_t *client) +{ + if (NULL == client->sqlite_cfg) { + return 0; + } + + return nng_mqtt_sqlite_db_get_cached_size(client->sqlite_cfg); +} + int neu_mqtt_client_set_addr(neu_mqtt_client_t *client, const char *host, uint16_t port) {