Merge pull request #553 from gc87/mqtt-queue-remove

Disconnect on user send stop command

include/neuron/connection/mqtt_client_intf.h

@@ -28,6 +28,8 @@ extern "C" {
 #include <stddef.h>
 #include <stdint.h>
 
+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;

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:

src/connection/mqtt_c_client.c

@@ -22,7 +22,6 @@
 #include <openssl/ssl.h>
 #include <stdio.h>
 #include <stdlib.h>
-#include <sys/queue.h>
 #include <unistd.h>
 
 #include <mqtt.h>
@@ -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);