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 9e81b907..289a621c 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -48,6 +48,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) @@ -178,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) { @@ -197,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); @@ -234,20 +250,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)); @@ -272,26 +274,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; } @@ -301,15 +324,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; @@ -329,8 +345,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_MQTT_FAILURE; + } + + plugin->routine = routine; + plugin->running = true; + plugin->common.link_state = NEU_PLUGIN_LINK_STATE_CONNECTING; } return NEU_ERR_SUCCESS; @@ -340,10 +365,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; } @@ -357,21 +379,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; } @@ -387,19 +411,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; } @@ -414,25 +440,51 @@ 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; } +// 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) { @@ -441,6 +493,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: diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c index d6fe65d5..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 { @@ -84,7 +82,8 @@ struct mqtt_c_client { struct reconnect_state state; pthread_t daemon; void * user_data; - TAILQ_HEAD(, subscribe_tuple) head; + state_update state_update_func; + UT_array * array; }; static neu_err_code_e client_destroy(mqtt_c_client_t *client); @@ -402,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); } } @@ -492,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); @@ -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,9 +553,11 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option, return NULL; } - client->option = option; - client->user_data = context; - TAILQ_INIT(&client->head); + client->option = option; + client->user_data = context; + client->state_update_func = client->option->state_update_func; + 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); @@ -700,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; } @@ -744,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; } @@ -779,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); } } @@ -879,6 +889,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);