Merge pull request #767 from fengzeroz/v2.1.0

version: 2.1.6

.github/workflows/ci.yml

@@ -96,11 +96,16 @@ jobs:
     needs: [cross-build, build-debug-sdk]
 
     steps:
+      - name: Extract branch name
+        shell: bash
+        run: echo "##[set-output name=branch;]$(echo ${GITHUB_REF#refs/heads/})"
+        id: extract_branch
+
       - name: delete old daily-pre-release
         uses: dev-drprasad/delete-tag-and-release@v0.2.0
         with:
           delete_release: true
-          tag_name: daily
+          tag_name: ${{ steps.extract_branch.outputs.branch }}-daily
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
@@ -110,8 +115,8 @@ jobs:
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:
-          tag_name: daily
-          release_name: Daily Release
+          tag_name: ${{ steps.extract_branch.outputs.branch }}-daily
+          release_name: Daily Release(${{ steps.extract_branch.outputs.branch }})
           draft: false
           prerelease: true
 
@@ -170,7 +175,7 @@ jobs:
           asset_name: ${{ matrix.type }}-${{ steps.get_version.outputs.version }}-daily-linux-${{ matrix.arch }}.${{ matrix.package }}
           asset_content_type: application/octet-stream
 
-  upload_daily_release_debug-asset:
+  upload_daily_release_debug_asset:
     runs-on: ubuntu-20.04
     if: github.event_name == 'push'
     needs: create_daily_release
@@ -218,7 +223,7 @@ jobs:
 
   clean:
     runs-on: ubuntu-20.04
-    needs: upload_daily_release_asset
+    needs: [upload_daily_release_asset, upload_daily_release_debug_asset]
     steps:
       - uses: geekyeggo/delete-artifact@v1
         with:

include/neuron/connection/neu_connection.h

@@ -38,6 +38,8 @@ typedef enum neu_conn_tty_baud {
     NEU_CONN_TTY_BAUD_38400,
     NEU_CONN_TTY_BAUD_19200,
     NEU_CONN_TTY_BAUD_9600,
+    NEU_CONN_TTY_BAUD_4800,
+    NEU_CONN_TTY_BAUD_2400,
 } neu_conn_tty_baud_e;
 
 typedef enum neu_conn_tty_parity {
@@ -60,6 +62,11 @@ typedef enum neu_conn_tty_data {
     NEU_CONN_TTY_DATA_8,
 } neu_conn_tty_data_e;
 
+typedef enum neu_conn_tty_flow {
+    NEU_CONN_TTYP_FLOW_DISABLE,
+    NEU_CONN_TTYP_FLOW_ENABLE,
+} neu_conn_tty_flow_e;
+
 typedef void (*neu_conn_callback)(void *data, int fd);
 
 typedef struct neu_conn_param {
@@ -96,6 +103,7 @@ typedef struct neu_conn_param {
             neu_conn_tty_stop_e   stop;
             neu_conn_tty_baud_e   baud;
             neu_conn_tty_parity_e parity;
+            neu_conn_tty_flow_e   flow;
             uint16_t              timeout; // millisecond
         } tty_client;
     } params;
@@ -109,6 +117,8 @@ neu_conn_t *neu_conn_new(neu_conn_param_t *param, void *data,
 neu_conn_t *neu_conn_reconfig(neu_conn_t *conn, neu_conn_param_t *param);
 void        neu_conn_destory(neu_conn_t *conn);
 void        neu_conn_disconnect(neu_conn_t *conn);
+void        neu_conn_stop(neu_conn_t *conn);
+void        neu_conn_start(neu_conn_t *conn);
 int         neu_conn_tcp_server_accept(neu_conn_t *conn);
 int         neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd);
 

include/neuron/utils/async_queue.h

@@ -41,5 +41,6 @@ int  neu_async_queue_pop(neu_async_queue_t *q, uint64_t key, void **elem);
 
 void neu_async_queue_remove(neu_async_queue_t *q, neu_async_queue_filter filter,
                             void *filter_elem);
+void neu_async_queue_clean(neu_async_queue_t *q);
 
 #endif

package-sdk.sh

@@ -50,10 +50,12 @@ cp neuron.pem ${package_name}/config/
 
 cp build/plugins/schema/ekuiper.json \
     build/plugins/schema/mqtt.json \
+    build/plugins/schema/modbus-tcp.json \
     ${package_name}/plugins/schema/
 
 cp build/plugins/libplugin-ekuiper.so \
     build/plugins/libplugin-mqtt.so \
+    build/plugins/libplugin-modbus-tcp.so \
     ${package_name}/plugins/
 
 tar czf ${package_name}-${arch}.tar.gz ${package_name}/

plugins/modbus/modbus_req.c

@@ -79,16 +79,16 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
     } else {
         struct modbus_group_data *gd =
             (struct modbus_group_data *) group->user_data;
+        plugin->plugin_group_data = gd;
 
         for (uint16_t i = 0; i < gd->cmd_sort->n_cmd; i++) {
             uint16_t response_size = 0;
-            int      ret           = modbus_stack_read(
+            plugin->cmd_idx        = i;
+            int ret                = modbus_stack_read(
                 plugin->stack, gd->cmd_sort->cmd[i].slave_id,
                 gd->cmd_sort->cmd[i].area, gd->cmd_sort->cmd[i].start_address,
                 gd->cmd_sort->cmd[i].n_register, &response_size);
             if (ret > 0) {
-                plugin->plugin_group_data = gd;
-                plugin->cmd_idx           = i;
                 process_protocol_buf(plugin, response_size);
             }
         }
@@ -106,6 +106,16 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
     uint16_t start_address = gd->cmd_sort->cmd[plugin->cmd_idx].start_address;
     uint16_t n_register    = gd->cmd_sort->cmd[plugin->cmd_idx].n_register;
 
+    if (bytes == NULL) {
+        neu_dvalue_t dvalue = { 0 };
+
+        dvalue.type      = NEU_TYPE_ERROR;
+        dvalue.value.i32 = NEU_ERR_PLUGIN_DISCONNECTED;
+        plugin->common.adapter_callbacks->driver.update(
+            plugin->common.adapter, gd->group, NULL, dvalue);
+        return 0;
+    }
+
     utarray_foreach(gd->cmd_sort->cmd[plugin->cmd_idx].tags, modbus_point_t **,
                     p_tag)
     {

plugins/modbus/modbus_stack.c

@@ -113,6 +113,7 @@ int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
 {
     static __thread uint8_t                 buf[32] = { 0 };
     static __thread neu_protocol_pack_buf_t pbuf    = { 0 };
+    int                                     ret     = 0;
 
     neu_protocol_pack_buf_init(&pbuf, buf, sizeof(buf));
 
@@ -143,8 +144,12 @@ int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
     modbus_header_wrap(&pbuf, stack->seq++);
     *response_size += sizeof(struct modbus_header);
 
-    return stack->send_fn(stack->ctx, neu_protocol_pack_buf_used_size(&pbuf),
-                          neu_protocol_pack_buf_get(&pbuf));
+    ret = stack->send_fn(stack->ctx, neu_protocol_pack_buf_used_size(&pbuf),
+                         neu_protocol_pack_buf_get(&pbuf));
+    if (ret <= 0) {
+        stack->value_fn(stack->ctx, 0, 0, NULL);
+    }
+    return ret;
 }
 
 int modbus_stack_write(modbus_stack_t *stack, void *req, uint8_t slave_id,

plugins/modbus/modbus_tcp.c

@@ -107,14 +107,13 @@ static int driver_uninit(neu_plugin_t *plugin)
 
 static int driver_start(neu_plugin_t *plugin)
 {
-    (void) plugin;
-
+    neu_conn_start(plugin->conn);
     return 0;
 }
 
 static int driver_stop(neu_plugin_t *plugin)
 {
-    neu_conn_disconnect(plugin->conn);
+    neu_conn_stop(plugin->conn);
     return 0;
 }
 

plugins/mqtt/mqtt_plugin.c

@@ -62,8 +62,6 @@ struct mqtt_routine {
     neu_event_timer_t *send;
 };
 
-static neu_plugin_t *plugin_log = NULL;
-
 static char *topics_format(char *format, char *name)
 {
     char temp[256] = { '\0' };
@@ -76,8 +74,9 @@ static char *topics_format(char *format, char *name)
     return NULL;
 }
 
-static void topics_add(UT_array *topics, char *request, int qos_request,
-                       char *response, int qos_response, int type)
+static void topics_add(UT_array *topics, neu_plugin_t *plugin, char *request,
+                       int qos_request, char *response, int qos_response,
+                       int type)
 {
     struct topic_pair *pair = calloc(1, sizeof(struct topic_pair));
     pair->topic_request     = request;
@@ -86,19 +85,21 @@ static void topics_add(UT_array *topics, char *request, int qos_request,
     pair->qos_response      = qos_response;
     pair->type              = type;
     utarray_push_back(topics, &pair);
-    plog_info(plugin_log, "add topic-req:%s, topic-res:%s, count:%d",
+
+    plog_info(plugin, "add topic-req:%s, topic-res:%s, count:%d",
               pair->topic_request, pair->topic_response, utarray_len(topics));
 }
 
-static void topics_cleanup(UT_array *topics)
+static void topics_cleanup(UT_array *topics, neu_plugin_t *plugin)
 {
     for (struct topic_pair **pair =
              (struct topic_pair **) utarray_front(topics);
          NULL != pair;
          pair = (struct topic_pair **) utarray_next(topics, pair)) {
 
-        plog_info(plugin_log, "remove topic-req:%s, topic-res:%s",
+        plog_info(plugin, "remove topic-req:%s, topic-res:%s",
                   (*pair)->topic_request, (*pair)->topic_response);
+
         if (NULL != (*pair)->topic_request) {
             free((*pair)->topic_request);
         }
@@ -111,39 +112,46 @@ static void topics_cleanup(UT_array *topics)
     }
 }
 
-static void topics_generate(UT_array *topics, char *name, char *upload_topic,
-                            char *heartbeat_topic)
+static void topics_generate(UT_array *topics, neu_plugin_t *plugin,
+                            neu_mqtt_option_t *option)
 {
+    char *name            = option->clientid;
+    char *upload_topic    = option->upload_topic;
+    char *heartbeat_topic = option->heartbeat_topic;
+
     char *read_req = topics_format(TOPIC_READ_REQ, name);
     char *read_res = topics_format(TOPIC_READ_RES, name);
-    topics_add(topics, read_req, QOS0, read_res, QOS0, TOPIC_TYPE_READ);
+    topics_add(topics, plugin, read_req, QOS0, read_res, QOS0, TOPIC_TYPE_READ);
 
     char *write_req = topics_format(TOPIC_WRITE_REQ, name);
     char *write_res = topics_format(TOPIC_WRITE_RES, name);
-    topics_add(topics, write_req, QOS0, write_res, QOS0, TOPIC_TYPE_WRITE);
+    topics_add(topics, plugin, write_req, QOS0, write_res, QOS0,
+               TOPIC_TYPE_WRITE);
 
     /// UPLOAD TOPIC SETTING
     char *upload_req = NULL;
     char *upload_res = NULL;
     if (NULL != upload_topic && 0 < strlen(upload_topic)) {
+        plog_debug(plugin, "user defined upload topic:%s", upload_topic);
         upload_res = strdup(upload_topic);
     } else {
         upload_res = topics_format(TOPIC_UPLOAD_RES, name);
     }
 
-    topics_add(topics, upload_req, QOS0, upload_res, QOS0, TOPIC_TYPE_UPLOAD);
+    topics_add(topics, plugin, upload_req, QOS0, upload_res, QOS0,
+               TOPIC_TYPE_UPLOAD);
 
     // HEARTBEAT TOPIC SETTING
     char *heartbeat_req = NULL;
     char *heartbeat_res = NULL;
     if (NULL != heartbeat_topic && 0 < strlen(heartbeat_topic)) {
-        nlog_debug("user defined heartbeat topic:%s", heartbeat_topic);
+        plog_debug(plugin, "user defined heartbeat topic:%s", heartbeat_topic);
         heartbeat_res = strdup(heartbeat_topic);
     } else {
         heartbeat_res = topics_format(TOPIC_HEARTBEAT_RES, name);
     }
 
-    topics_add(topics, heartbeat_req, QOS0, heartbeat_res, QOS0,
+    topics_add(topics, plugin, heartbeat_req, QOS0, heartbeat_res, QOS0,
                TOPIC_TYPE_HEARTBEAT);
 }
 
@@ -258,8 +266,8 @@ static int mqtt_routine_heartbeat(void *data)
         neu_mqtt_client_publish(routine->client, topic, qos,
                                 (unsigned char *) json_str, strlen(json_str));
     if (NEU_ERR_SUCCESS != error) {
-        plog_error(plugin_log, "heartbeat publish error code :%d, topoic:%s",
-                   error, topic);
+        plog_error(routine->plugin,
+                   "heartbeat publish error code :%d, topoic:%s", error, topic);
     }
 
     free(json_str);
@@ -306,9 +314,7 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin,
     routine->plugin      = plugin;
     UT_icd topic_ptr_icd = { sizeof(struct topic_pair *), NULL, NULL, NULL };
     utarray_new(routine->topics, &topic_ptr_icd);
-    topics_generate(routine->topics, routine->option.clientid,
-                    routine->option.upload_topic,
-                    routine->option.heartbeat_topic);
+    topics_generate(routine->topics, plugin, &routine->option);
     topics_subscribe(routine->topics, routine->client);
 
     UT_icd state_ptr_icd = { sizeof(struct node_state *), NULL, NULL, NULL };
@@ -356,7 +362,7 @@ static void mqtt_routine_stop(mqtt_routine_t *routine)
     plog_info(routine->plugin, "close mqtt client:%s:%s", routine->option.host,
               routine->option.port);
 
-    topics_cleanup(routine->topics);
+    topics_cleanup(routine->topics, routine->plugin);
     utarray_free(routine->topics);
     mqtt_option_uninit(routine->plugin, &routine->option);
 }
@@ -365,7 +371,6 @@ static neu_plugin_t *mqtt_plugin_open(void)
 {
     neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t));
     neu_plugin_common_init(&plugin->common);
-    plugin_log = plugin; // for plog
     return plugin;
 }
 
@@ -376,7 +381,6 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
     const char *name = neu_plugin_module.module_name;
     plog_info(plugin, "success to free plugin:%s", name);
 
-    plugin_log = NULL;
     free(plugin);
     return NEU_ERR_SUCCESS;
 }

plugins/restful/handle.h

@@ -32,8 +32,6 @@
         size_t    req_data_size = 0;                                          \
         req_type *req           = NULL;                                       \
                                                                               \
-        nng_aio_set_timeout(aio, 3000);                                       \
-        nng_aio_defer(aio, http_timeout, NULL);                               \
         if (http_get_body((aio), (void **) &req_data, &req_data_size) == 0 && \
             decode_fun(req_data, &req) == 0) {                                \
             { func };                                                         \

plugins/restful/http.c

@@ -408,6 +408,7 @@ void http_timeout(nng_aio *aio, void *arg, int err)
         http_internal_error(aio, "{\"error\": \"canceled\"}");
         break;
     default:
+        nlog_error("aio error: %d", err);
         http_internal_error(aio, "{\"error\": \"unknown\"}");
         break;
     }

plugins/restful/http.h

@@ -80,7 +80,6 @@ int         http_unauthorized(nng_aio *aio, char *content);
 int         http_not_found(nng_aio *aio, char *content);
 int         http_conflict(nng_aio *aio, char *content);
 int         http_internal_error(nng_aio *aio, char *content);
-void        http_timeout(nng_aio *aio, void *arg, int err);
 
 #ifdef __cplusplus
 }

src/adapter/driver/driver.c

@@ -290,6 +290,12 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             free(tag);
             return;
         }
+        if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
+            if (cmd->value.type == NEU_TYPE_INT64) {
+                cmd->value.value.d64 = (double) cmd->value.value.u64;
+            }
+        }
+
         switch (tag->type) {
         case NEU_TYPE_BOOL:
         case NEU_TYPE_DOUBLE:

src/base/neu_plugin_common.c

@@ -20,6 +20,7 @@
 #include <assert.h>
 #include <stdlib.h>
 #include <string.h>
+#include <time.h>
 
 #include "errcodes.h"
 #include "plugin.h"
@@ -31,6 +32,7 @@ void neu_plugin_common_init(neu_plugin_common_t *common)
     common->magic      = NEU_PLUGIN_MAGIC_NUMBER;
     common->link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
     common->tag_size   = 0;
+    common->timestamp  = time(NULL) * 1000; // don't need much precision at init
 }
 
 bool neu_plugin_common_check(neu_plugin_t *plugin)

src/connection/connection.c

@@ -42,6 +42,7 @@ struct neu_conn {
     neu_conn_param_t param;
     void *           data;
     bool             is_connected;
+    bool             stop;
 
     neu_conn_callback connected;
     neu_conn_callback disconnected;
@@ -84,6 +85,7 @@ neu_conn_t *neu_conn_new(neu_conn_param_t *param, void *data,
     conn->data             = data;
     conn->disconnected     = disconnected;
     conn->connected        = connected;
+    conn->stop             = false;
 
     conn_tcp_server_listen(conn);
 
@@ -92,6 +94,21 @@ neu_conn_t *neu_conn_new(neu_conn_param_t *param, void *data,
     return conn;
 }
 
+void neu_conn_stop(neu_conn_t *conn)
+{
+    pthread_mutex_lock(&conn->mtx);
+    conn->stop = true;
+    conn_disconnect(conn);
+    pthread_mutex_unlock(&conn->mtx);
+}
+
+void neu_conn_start(neu_conn_t *conn)
+{
+    pthread_mutex_lock(&conn->mtx);
+    conn->stop = false;
+    pthread_mutex_unlock(&conn->mtx);
+}
+
 neu_conn_t *neu_conn_reconfig(neu_conn_t *conn, neu_conn_param_t *param)
 {
     pthread_mutex_lock(&conn->mtx);
@@ -198,8 +215,20 @@ int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd)
 ssize_t neu_conn_tcp_server_send(neu_conn_t *conn, int fd, uint8_t *buf,
                                  ssize_t len)
 {
+    ssize_t ret = 0;
+
+    pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
+
     conn_tcp_server_listen(conn);
-    return send(fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT);
+    ret = send(fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT);
+
+    pthread_mutex_unlock(&conn->mtx);
+
+    return ret;
 }
 
 ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
@@ -207,11 +236,19 @@ ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
 {
     ssize_t ret = 0;
 
+    pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
+
     if (conn->block) {
         ret = recv(fd, buf, len, MSG_WAITALL);
     } else {
         ret = recv(fd, buf, len, 0);
     }
+    pthread_mutex_unlock(&conn->mtx);
+
     return ret;
 }
 
@@ -220,6 +257,10 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
     ssize_t ret = 0;
 
     pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
 
     if (conn->is_connected) {
         switch (conn->param.type) {
@@ -268,6 +309,10 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
     ssize_t ret = 0;
 
     pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
 
     switch (conn->param.type) {
     case NEU_CONN_TCP_SERVER:
@@ -416,6 +461,7 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         conn->param.params.tty_client.stop   = param->params.tty_client.stop;
         conn->param.params.tty_client.baud   = param->params.tty_client.baud;
         conn->param.params.tty_client.parity = param->params.tty_client.parity;
+        conn->param.params.tty_client.flow   = param->params.tty_client.flow;
         conn->param.params.tty_client.timeout =
             param->params.tty_client.timeout;
         conn->block = conn->param.params.tty_client.timeout > 0;
@@ -606,6 +652,15 @@ static void conn_connect(neu_conn_t *conn)
         tty_opt.c_cc[VTIME] = conn->param.params.tty_client.timeout / 100;
         tty_opt.c_cc[VMIN]  = 0;
 
+        switch (conn->param.params.tty_client.flow) {
+        case NEU_CONN_TTYP_FLOW_DISABLE:
+            tty_opt.c_cflag &= ~CRTSCTS;
+            break;
+        case NEU_CONN_TTYP_FLOW_ENABLE:
+            tty_opt.c_cflag |= CRTSCTS;
+            break;
+        }
+
         switch (conn->param.params.tty_client.baud) {
         case NEU_CONN_TTY_BAUD_115200:
             cfsetospeed(&tty_opt, B115200);
@@ -627,6 +682,14 @@ static void conn_connect(neu_conn_t *conn)
             cfsetospeed(&tty_opt, B9600);
             cfsetispeed(&tty_opt, B9600);
             break;
+        case NEU_CONN_TTY_BAUD_4800:
+            cfsetospeed(&tty_opt, B4800);
+            cfsetispeed(&tty_opt, B4800);
+            break;
+        case NEU_CONN_TTY_BAUD_2400:
+            cfsetospeed(&tty_opt, B2400);
+            cfsetispeed(&tty_opt, B2400);
+            break;
         }
 
         switch (conn->param.params.tty_client.parity) {
@@ -682,10 +745,18 @@ static void conn_connect(neu_conn_t *conn)
             break;
         }
 
-        tty_opt.c_oflag = 0;
-        tty_opt.c_lflag = 0;
-        tty_opt.c_iflag |= IGNBRK;
-        tty_opt.c_cflag |= CREAD | HUPCL | CLOCAL;
+        tty_opt.c_cflag |= CREAD | CLOCAL;
+        tty_opt.c_lflag &= ~ICANON;
+        tty_opt.c_lflag &= ~ECHO;
+        tty_opt.c_lflag &= ~ECHOE;
+        tty_opt.c_lflag &= ~ECHONL;
+        tty_opt.c_lflag &= ~ISIG;
+
+        tty_opt.c_iflag &= ~(IXON | IXOFF | IXANY);
+        tty_opt.c_iflag &=
+            ~(IGNBRK | BRKINT | PARMRK | ISTRIP | INLCR | IGNCR | ICRNL);
+        tty_opt.c_oflag &= ~OPOST;
+        tty_opt.c_oflag &= ~ONLCR;
 
         tcflush(fd, TCIFLUSH);
         tcsetattr(fd, TCSANOW, &tty_opt);

src/connection/mqtt_c_client.c

@@ -61,9 +61,10 @@ struct reconnect_state {
     EVP_PKEY *       key;
     SSL_CTX *        ssl_ctx;
     BIO *            sock_fd;
-    int              sock_state; // sock state, value same as mqtt.error
-    bool             running;    // refresher stop flag
-    bool             working;    // publish & publish callback working flag
+    int              sock_state;        // sock state, value same as mqtt.error
+    int              sock_do_connected; // sock do connect state
+    bool             running;           // refresher stop flag
+    bool             working; // publish & publish callback working flag
     pthread_mutex_t  sock_state_mutex;
     pthread_mutex_t  running_mutex;
     pthread_mutex_t  working_mutex;
@@ -341,11 +342,16 @@ static int bio_connect(struct reconnect_state *state)
     return 0;
 }
 
-static void bio_update_connection(struct reconnect_state *state)
+static int bio_update_connection(struct reconnect_state *state)
 {
     bio_uninit(state);
     bio_init(state);
-    bio_connect(state);
+
+    if (NULL == state->sock_fd) {
+        return -1;
+    }
+
+    return bio_connect(state);
 }
 
 static void reconnect_state_uninit(struct reconnect_state *state)
@@ -427,22 +433,44 @@ static void subscribe_on_reconnect(mqtt_c_client_t *client)
                            (*p)->topic, error);
             }
         }
-
-        if (NULL != client->log) {
-            zlog_error(client->log, "send subscribe %s success:%d", (*p)->topic,
-                       error);
-        }
     }
 }
 
 enum MQTTErrors inspector_callback(struct mqtt_client *mqtt)
 {
     struct reconnect_state *state = mqtt->reconnect_state;
+
+    if (NULL == state->sock_fd) {
+        mqtt->error = MQTT_ERROR_SOCKET_ERROR;
+
+        if (NULL != state->client->log) {
+            zlog_error(state->client->log, "bio is null");
+        }
+
+        return mqtt->error;
+    }
+
+    if (0 != state->sock_do_connected) {
+        mqtt->error = MQTT_ERROR_SOCKET_ERROR;
+
+        if (NULL != state->client->log) {
+            zlog_error(state->client->log, "bio do connect failed");
+        }
+
+        return mqtt->error;
+    }
+
     if (0 == strcmp(state->connection, "ssl://")) {
         SSL *ssl = NULL;
         BIO_get_ssl(state->sock_fd, &ssl);
         if (X509_V_OK != SSL_get_verify_result(ssl)) {
             mqtt->error = MQTT_ERROR_SOCKET_ERROR;
+
+            if (NULL != state->client->log) {
+                zlog_error(state->client->log, "bio ssl verify failed");
+            }
+
+            return mqtt->error;
         }
     }
 
@@ -458,8 +486,7 @@ static void client_reconnect(struct mqtt_client *mqtt, void **p_reconnect_state)
         *((struct reconnect_state **) p_reconnect_state);
     mqtt_c_client_t *client  = state->client;
     mqtt->inspector_callback = inspector_callback;
-
-    bio_update_connection(state);
+    state->sock_do_connected = bio_update_connection(state);
 
     mqtt_reinit(mqtt, state->sock_fd, state->send_buf, state->send_buf_sz,
                 state->recv_buf, state->recv_buf_sz);
@@ -513,8 +540,10 @@ static void publish_callback(void **                       p_reconnect_state,
     memcpy(topic_name, published->topic_name, published->topic_name_size);
     topic_name[published->topic_name_size] = '\0';
 
-    nlog_debug("received publish('%s'): %s", topic_name,
-               (const char *) published->application_message);
+    if (NULL != client->log) {
+        zlog_debug(client->log, "received publish('%s'): %s", topic_name,
+                   (const char *) published->application_message);
+    }
 
     for (struct subscribe_tuple **p =
              (struct subscribe_tuple **) utarray_front(client->array);
@@ -563,8 +592,10 @@ static void *client_refresher(void *context)
             client->state_update_func(client->user_data, rc);
 
             if (0 != rc && error_count % 100 == 0) {
-                zlog_error(client->log, "connect to %s:%s erorr, retry...",
-                           client->state.hostname, client->state.port);
+                if (NULL != client->log) {
+                    zlog_error(client->log, "connect to %s:%s erorr, retry...",
+                               client->state.hostname, client->state.port);
+                }
             }
         }
         error_count++;
@@ -580,7 +611,10 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option,
 {
     mqtt_c_client_t *client = calloc(1, sizeof(mqtt_c_client_t));
     if (NULL == client) {
-        zlog_error(option->log, "mqtt c client alloc error");
+        if (NULL != option->log) {
+            zlog_error(option->log, "mqtt c client alloc error");
+        }
+
         return NULL;
     }
 

src/utils/async_queue.c

@@ -160,4 +160,18 @@ void neu_async_queue_remove(neu_async_queue_t *q, neu_async_queue_filter filter,
         }
     }
     nng_mtx_unlock(q->mtx);
+}
+
+void neu_async_queue_clean(neu_async_queue_t *q)
+{
+    element *el = NULL, *tmp = NULL;
+
+    nng_mtx_lock(q->mtx);
+    DL_FOREACH_SAFE(q->list, el, tmp)
+    {
+        DL_DELETE(q->list, el);
+        q->free_fn(el->data);
+        free(el);
+    }
+    nng_mtx_unlock(q->mtx);
 }

version

@@ -1 +1 @@
-2.1.1
+2.1.6

zlog.conf

@@ -9,4 +9,4 @@ simple = "%d [%V] %f:%L %m%n"
 
 [rules]
 
-*.INFO     "./logs/%c.log", 10MB * 3 ~ "./logs/%c-%d(%Y%m%d).#2s.log"; simple
+*.INFO     "./logs/%c.log", 10MB * 3 ~ "./logs/%c.#2s.log"; simple