From 34a3e43c588d433fd706253b62b9b75c0efe9656 Mon Sep 17 00:00:00 2001 From: gc87 Date: Thu, 16 Jun 2022 13:04:40 +0800 Subject: [PATCH 1/4] refactor(mqtt): fix connection state --- include/neuron/connection/mqtt_client_intf.h | 9 ++++++--- plugins/mqtt/mqtt.c | 19 +++++++++++++++++-- src/connection/mqtt_c_client.c | 15 +++++++++++++-- 3 files changed, 36 insertions(+), 7 deletions(-) diff --git a/include/neuron/connection/mqtt_client_intf.h b/include/neuron/connection/mqtt_client_intf.h index 900e0610..427c28a2 100644 --- a/include/neuron/connection/mqtt_client_intf.h +++ b/include/neuron/connection/mqtt_client_intf.h @@ -28,6 +28,8 @@ extern "C" { #include #include +typedef void (*state_update)(void *context, int state); + typedef struct { /* debug app options */ int verbose; @@ -56,9 +58,10 @@ typedef struct { char *keypass; /*application*/ - int format; - char *upload_topic; - char *group_id; + int format; + char * upload_topic; + char * group_id; + state_update state_update_func; } neu_mqtt_option_t; typedef void *neu_mqtt_client_t; diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 7206570d..bc541dcf 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -179,6 +179,20 @@ static void topics_subscribe(UT_array *topics, neu_mqtt_client_t client) } } +static void mqtt_routine_state(void *context, int state) +{ + if (NULL == context) { + return; + } + + neu_plugin_t *plugin = (neu_plugin_t *) context; + if (0 == state) { + plugin->common.link_state = NEU_PLUGIN_LINK_STATE_CONNECTED; + } else { + plugin->common.link_state = NEU_PLUGIN_LINK_STATE_DISCONNECTED; + } +} + static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, const char * config) { @@ -198,8 +212,9 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin, return NULL; } - neu_mqtt_client_t *client = (neu_mqtt_client_t *) &routine->client; - neu_err_code_e error = NEU_ERR_SUCCESS; + routine->option.state_update_func = mqtt_routine_state; + neu_mqtt_client_t *client = (neu_mqtt_client_t *) &routine->client; + neu_err_code_e error = NEU_ERR_SUCCESS; error = neu_mqtt_client_open(client, &routine->option, plugin); if (NEU_ERR_SUCCESS != error) { mqtt_option_uninit(&routine->option); diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c index d6fe65d5..4e723efb 100644 --- a/src/connection/mqtt_c_client.c +++ b/src/connection/mqtt_c_client.c @@ -84,6 +84,7 @@ struct mqtt_c_client { struct reconnect_state state; pthread_t daemon; void * user_data; + state_update state_update_func; TAILQ_HEAD(, subscribe_tuple) head; }; @@ -506,6 +507,8 @@ static void publish_callback(void ** p_reconnect_state, free(topic_name); } +neu_err_code_e mqtt_c_client_is_connected(mqtt_c_client_t *client); + static void *client_refresher(void *context) { assert(NULL != context); @@ -529,6 +532,11 @@ static void *client_refresher(void *context) state->sock_state = error; pthread_mutex_unlock(&state->sock_state_mutex); usleep(INTERVAL); + + if (NULL != client->state_update_func) { + int rc = mqtt_c_client_is_connected(client); + client->state_update_func(client->user_data, rc); + } } reconnect_state_uninit(state); @@ -545,8 +553,9 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option, return NULL; } - client->option = option; - client->user_data = context; + client->option = option; + client->user_data = context; + client->state_update_func = client->option->state_update_func; TAILQ_INIT(&client->head); struct reconnect_state *state = &client->state; @@ -879,6 +888,8 @@ neu_err_code_e mqtt_c_client_close(mqtt_c_client_t *client) mqtt_c_client_suspend(client); client_disconnect(client); + client->state_update_func = NULL; + client->user_data = NULL; pthread_mutex_lock(&client->state.running_mutex); client->state.running = false; pthread_mutex_unlock(&client->state.running_mutex); From 7da3b7b50a9597f1c6878f499c96b4ec69fca68b Mon Sep 17 00:00:00 2001 From: gc87 Date: Thu, 16 Jun 2022 18:13:04 +0800 Subject: [PATCH 2/4] feat(mqtt): stop/start -> disconnect/reconnect --- plugins/mqtt/mqtt.c | 114 ++++++++++++++++++++++++++++---------------- 1 file changed, 74 insertions(+), 40 deletions(-) diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index bc541dcf..7f2b675e 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -49,6 +49,7 @@ struct neu_plugin { neu_plugin_common_t common; bool running; mqtt_routine_t * routine; + char * config; }; static char *topics_format(char *format, char *name) @@ -250,20 +251,6 @@ static void mqtt_routine_stop(mqtt_routine_t *routine) mqtt_option_uninit(&routine->option); } -static void mqtt_routine_continue(mqtt_routine_t *routine) -{ - assert(NULL != routine && NULL != routine->client); - - neu_mqtt_client_continue(routine->client); -} - -static void mqtt_routine_suspend(mqtt_routine_t *routine) -{ - assert(NULL != routine && NULL != routine->client); - - neu_mqtt_client_suspend(routine->client); -} - static neu_plugin_t *mqtt_plugin_open(void) { neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t)); @@ -288,26 +275,47 @@ static int mqtt_plugin_init(neu_plugin_t *plugin) plugin->routine = NULL; plugin->running = false; + plugin->config = NULL; const char *name = neu_plugin_module.module_name; plog_info(plugin, "initialize plugin: %s", name); return NEU_ERR_SUCCESS; } +static void plugin_stop_running(neu_plugin_t *plugin) +{ + if (plugin->running) { + mqtt_routine_stop(plugin->routine); + if (NULL != plugin->routine) { + free(plugin->routine); + plugin->routine = NULL; + } + + plugin->running = false; + plugin->common.link_state = NEU_PLUGIN_LINK_STATE_DISCONNECTED; + } +} + +static void plugin_config_free(neu_plugin_t *plugin) +{ + if (NULL != plugin->config) { + free(plugin->config); + plugin->config = NULL; + } +} + +static void plugin_config_save(neu_plugin_t *plugin, const char *config) +{ + plugin_config_free(plugin); + plugin->config = strdup(config); +} + static int mqtt_plugin_uninit(neu_plugin_t *plugin) { assert(NULL != plugin); - if (plugin->running) { - mqtt_routine_stop(plugin->routine); - plugin->running = false; - if (NULL != plugin->routine) { - free(plugin->routine); - plugin->routine = NULL; - } - } - - plugin->common.link_state = NEU_PLUGIN_LINK_STATE_DISCONNECTED; - const char *name = neu_plugin_module.module_name; + plugin_stop_running(plugin); + plugin_config_free(plugin); + const char *name = neu_plugin_module.module_name; plog_info(plugin, "uninitialize plugin: %s", name); return NEU_ERR_SUCCESS; } @@ -317,15 +325,8 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *config) assert(NULL != plugin); assert(NULL != config); - if (plugin->running) { - mqtt_routine_stop(plugin->routine); - plugin->running = false; - if (NULL != plugin->routine) { - free(plugin->routine); - plugin->routine = NULL; - } - } - + plugin_config_save(plugin, config); + plugin_stop_running(plugin); mqtt_routine_t *routine = mqtt_routine_start(plugin, config); if (NULL == routine) { plugin->routine = NULL; @@ -345,8 +346,17 @@ static int mqtt_plugin_start(neu_plugin_t *plugin) { assert(NULL != plugin); - if (plugin->running) { - mqtt_routine_continue(plugin->routine); + if (!plugin->running && NULL != plugin->config) { + mqtt_routine_t *routine = mqtt_routine_start(plugin, plugin->config); + if (NULL == routine) { + plugin->routine = NULL; + plugin->running = false; + return NEU_ERR_FAILURE; + } + + plugin->routine = routine; + plugin->running = true; + plugin->common.link_state = NEU_PLUGIN_LINK_STATE_CONNECTING; } return NEU_ERR_SUCCESS; @@ -356,10 +366,7 @@ static int mqtt_plugin_stop(neu_plugin_t *plugin) { assert(NULL != plugin); - if (plugin->running) { - mqtt_routine_suspend(plugin->routine); - } - + plugin_stop_running(plugin); return NEU_ERR_SUCCESS; } @@ -449,6 +456,29 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data) return NEU_ERR_SUCCESS; } +// static neu_err_code_e reconnect(neu_plugin_t *plugin, neu_reqresp_head_t +// *head) +// { +// if (NEU_RESP_ERROR == head->type || NEU_RESP_READ_GROUP == head->type || +// NEU_REQRESP_TRANS_DATA == head->type) { +// if (!plugin->running && NULL != plugin->config) { +// mqtt_routine_t *routine = +// mqtt_routine_start(plugin, plugin->config); +// if (NULL == routine) { +// plugin->routine = NULL; +// plugin->running = false; +// return NEU_ERR_FAILURE; +// } + +// plugin->routine = routine; +// plugin->running = true; +// plugin->common.link_state = NEU_PLUGIN_LINK_STATE_CONNECTING; +// } +// } + +// return NEU_ERR_SUCCESS; +// } + static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, void *data) { @@ -457,6 +487,10 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, assert(NULL != data); neu_err_code_e error = NEU_ERR_SUCCESS; + // error = reconnect(plugin, head); + // if (NEU_ERR_SUCCESS != error) { + // return error; + // } switch (head->type) { case NEU_RESP_ERROR: From 6881025f79dc1e1d355262ee45c9aa1de2ec59e2 Mon Sep 17 00:00:00 2001 From: gc87 Date: Thu, 16 Jun 2022 18:15:53 +0800 Subject: [PATCH 3/4] fix(mqtt): string leak --- plugins/mqtt/mqtt.c | 43 +++++++++++++++++++++++++------------------ 1 file changed, 25 insertions(+), 18 deletions(-) diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 7f2b675e..8ce32290 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -351,7 +351,7 @@ static int mqtt_plugin_start(neu_plugin_t *plugin) if (NULL == routine) { plugin->routine = NULL; plugin->running = false; - return NEU_ERR_FAILURE; + return NEU_ERR_MQTT_FAILURE; } plugin->routine = routine; @@ -380,21 +380,23 @@ static neu_err_code_e write_response(neu_plugin_t * plugin, return NEU_ERR_MQTT_FAILURE; } - mqtt_routine_t * routine = plugin->routine; - int type = TOPIC_TYPE_READ; - struct topic_pair *pair = topics_find_type(routine->topics, type); - char * json = command_write_response(head, write_data); - - const char * topic = pair->topic_response; - const int qos = pair->qos_response; - neu_err_code_e error = neu_mqtt_client_publish( - routine->client, topic, qos, (unsigned char *) json, strlen(json)); + mqtt_routine_t * routine = plugin->routine; + int type = TOPIC_TYPE_READ; + struct topic_pair *pair = topics_find_type(routine->topics, type); + char * json_str = command_write_response(head, write_data); + const char * topic = pair->topic_response; + const int qos = pair->qos_response; + neu_err_code_e error = + neu_mqtt_client_publish(routine->client, topic, qos, + (unsigned char *) json_str, strlen(json_str)); if (NEU_ERR_SUCCESS != error) { plog_error(plugin, "trans data publish error code :%d, topoic:%s", error, topic); + free(json_str); return error; } + free(json_str); return NEU_ERR_SUCCESS; } @@ -410,19 +412,21 @@ static neu_err_code_e read_response(neu_plugin_t * plugin, mqtt_routine_t * routine = plugin->routine; int type = TOPIC_TYPE_READ; struct topic_pair *pair = topics_find_type(routine->topics, type); - char * json = + char * json_str = command_read_once_response(head, read_data, routine->option.format); - const char * topic = pair->topic_response; const int qos = pair->qos_response; - neu_err_code_e error = neu_mqtt_client_publish( - routine->client, topic, qos, (unsigned char *) json, strlen(json)); + neu_err_code_e error = + neu_mqtt_client_publish(routine->client, topic, qos, + (unsigned char *) json_str, strlen(json_str)); if (NEU_ERR_SUCCESS != error) { plog_error(plugin, "read response publish error code :%d, topoic:%s", error, topic); + free(json_str); return error; } + free(json_str); return NEU_ERR_SUCCESS; } @@ -437,22 +441,25 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data) mqtt_routine_t * routine = plugin->routine; int type = TOPIC_TYPE_UPLOAD; struct topic_pair *pair = topics_find_type(routine->topics, type); - char * json = + char * json_str = command_read_periodic_response(trans_data, routine->option.format); - if (NULL == json) { + if (NULL == json_str) { return NEU_ERR_MQTT_FAILURE; } const char * topic = pair->topic_response; const int qos = pair->qos_response; - neu_err_code_e error = neu_mqtt_client_publish( - routine->client, topic, qos, (unsigned char *) json, strlen(json)); + neu_err_code_e error = + neu_mqtt_client_publish(routine->client, topic, qos, + (unsigned char *) json_str, strlen(json_str)); if (NEU_ERR_SUCCESS != error) { plog_error(plugin, "trans data publish error code :%d, topoic:%s", error, topic); + free(json_str); return error; } + free(json_str); return NEU_ERR_SUCCESS; } From 24eeb4787fdb03a3c77bc3a272f203ea81d4714d Mon Sep 17 00:00:00 2001 From: gc87 Date: Fri, 17 Jun 2022 11:49:08 +0800 Subject: [PATCH 4/4] refactor(mqtt): remove sys/queue.h deps --- src/connection/mqtt_c_client.c | 55 +++++++++++++++++----------------- 1 file changed, 28 insertions(+), 27 deletions(-) diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c index 4e723efb..6512e37a 100644 --- a/src/connection/mqtt_c_client.c +++ b/src/connection/mqtt_c_client.c @@ -22,7 +22,6 @@ #include #include #include -#include #include #include @@ -75,7 +74,6 @@ struct subscribe_tuple { int qos; neu_subscribe_handle handle; mqtt_c_client_t * client; - TAILQ_ENTRY(subscribe_tuple) entry; }; struct mqtt_c_client { @@ -85,7 +83,7 @@ struct mqtt_c_client { pthread_t daemon; void * user_data; state_update state_update_func; - TAILQ_HEAD(, subscribe_tuple) head; + UT_array * array; }; static neu_err_code_e client_destroy(mqtt_c_client_t *client); @@ -403,12 +401,13 @@ static void subscribe_on_reconnect(mqtt_c_client_t *client) { assert(NULL != client); - struct subscribe_tuple *item = NULL; - TAILQ_FOREACH(item, &client->head, entry) - { - neu_err_code_e error = client_subscribe_send(client, item); + for (struct subscribe_tuple **p = + (struct subscribe_tuple **) utarray_front(client->array); + NULL != p; + p = (struct subscribe_tuple **) utarray_next(client->array, p)) { + neu_err_code_e error = client_subscribe_send(client, (*p)); if (NEU_ERR_SUCCESS != error) { - zlog_error(neuron, "send subscribe %s error:%d", item->topic, + zlog_error(neuron, "send subscribe %s error:%d", (*p)->topic, error); } } @@ -493,11 +492,12 @@ static void publish_callback(void ** p_reconnect_state, zlog_debug(neuron, "received publish('%s'): %s", topic_name, (const char *) published->application_message); - struct subscribe_tuple *item = NULL; - TAILQ_FOREACH(item, &client->head, entry) - { - if (0 == strcmp(item->topic, topic_name) && NULL != item->handle) { - item->handle(topic_name, strlen(topic_name), + for (struct subscribe_tuple **p = + (struct subscribe_tuple **) utarray_front(client->array); + NULL != p; + p = (struct subscribe_tuple **) utarray_next(client->array, p)) { + if (0 == strcmp((*p)->topic, topic_name) && NULL != (*p)->handle) { + (*p)->handle(topic_name, strlen(topic_name), (void *) published->application_message, published->application_message_size, client->user_data); @@ -556,7 +556,8 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option, client->option = option; client->user_data = context; client->state_update_func = client->option->state_update_func; - TAILQ_INIT(&client->head); + UT_icd tuple_icd = { sizeof(struct subscribe_tuple *), NULL, NULL, NULL }; + utarray_new(client->array, &tuple_icd); struct reconnect_state *state = &client->state; state->hostname = strdup(client->option->host); @@ -709,15 +710,14 @@ static neu_err_code_e client_disconnect(mqtt_c_client_t *client) static neu_err_code_e client_destroy(mqtt_c_client_t *client) { - while (!TAILQ_EMPTY(&client->head)) { - struct subscribe_tuple *item = NULL; - item = TAILQ_FIRST(&client->head); - if (NULL != item) { - TAILQ_REMOVE(&client->head, item, entry); - client_subscribe_destroy(item); - } + for (struct subscribe_tuple **p = + (struct subscribe_tuple **) utarray_front(client->array); + NULL != p; + p = (struct subscribe_tuple **) utarray_next(client->array, p)) { + client_subscribe_destroy(*p); } + utarray_free(client->array); free(client); return NEU_ERR_SUCCESS; } @@ -753,7 +753,7 @@ static neu_err_code_e client_subscribe_send(mqtt_c_client_t * client, static neu_err_code_e client_subscribe_add(mqtt_c_client_t * client, struct subscribe_tuple *tuple) { - TAILQ_INSERT_TAIL(&client->head, tuple, entry); + utarray_push_back(client->array, &tuple); return NEU_ERR_SUCCESS; } @@ -788,11 +788,12 @@ neu_err_code_e mqtt_c_client_subscribe(mqtt_c_client_t *client, static struct subscribe_tuple * client_unsubscribe_create(mqtt_c_client_t *client, const char *topic) { - struct subscribe_tuple *item = NULL; - TAILQ_FOREACH(item, &client->head, entry) - { - if (0 == strcmp(item->topic, topic)) { - return item; + for (struct subscribe_tuple **p = + (struct subscribe_tuple **) utarray_front(client->array); + NULL != p; + p = (struct subscribe_tuple **) utarray_next(client->array, p)) { + if (0 == strcmp((*p)->topic, topic)) { + return (*p); } }