Merge pull request #1196 from fengzeroz/v2.4

version: 2.4.2

.github/workflows/check.yml

@@ -7,6 +7,11 @@ on:
     types:
       - published
 
+
+env:
+  BRANCH: v2.4
+
+
 jobs:
   clang_format_check:
     runs-on: ubuntu-20.04
@@ -40,7 +45,7 @@ jobs:
 
   ut:
     runs-on: ubuntu-20.04
-    container: neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.4
 
     steps:
       - uses: actions/checkout@v2
@@ -61,7 +66,7 @@ jobs:
 
   ft:
     runs-on: ubuntu-20.04
-    container: neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.4
 
     steps:
       - uses: actions/checkout@v2

.github/workflows/ci.yml

@@ -8,9 +8,9 @@ on:
       - published
 
 env:
-  VERSION: 2.4.0
-  UI_VERSION: 2.4.0
-  BRANCH: main
+  VERSION: 2.4.2
+  UI_VERSION: 2.4.2
+  BRANCH: v2.4
 
 jobs:
   cross-build:
@@ -21,7 +21,7 @@ jobs:
           - [aarch64-linux-gnu, aarch64, arm64]
           - [arm-linux-gnueabihf, armv4, armhf]
           - [x86_64-linux-gnu, x86_64, amd64]
-    container: neugates/build:${{ matrix.arch[1] }}-main
+    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.4
 
     steps:
       - uses: actions/checkout@v3
@@ -64,7 +64,7 @@ jobs:
       matrix:
         arch:
           - [x86_64-linux-gnu, x86_64, amd64]
-    container: neugates/build:${{ matrix.arch[1] }}-main
+    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.4
 
     steps:
       - uses: actions/checkout@v3

dev.conf

@@ -9,5 +9,5 @@ file_format = "%d [%V] %f:%L %m%n"
 
 [rules]
 
-*.*     "./logs/%c.log", 100MB * 10 ~ "./logs/%c.#2s.log"; file_format
+*.*     "./logs/%c.log", 50MB * 1 ~ "./logs/%c.#2s.log"; file_format
 *.*    	>stdout; stdout_format

include/neuron/connection/neu_connection.h

@@ -40,6 +40,12 @@ typedef enum neu_conn_tty_baud {
     NEU_CONN_TTY_BAUD_9600,
     NEU_CONN_TTY_BAUD_4800,
     NEU_CONN_TTY_BAUD_2400,
+    NEU_CONN_TTY_BAUD_1800,
+    NEU_CONN_TTY_BAUD_1200,
+    NEU_CONN_TTY_BAUD_600,
+    NEU_CONN_TTY_BAUD_300,
+    NEU_CONN_TTY_BAUD_200,
+    NEU_CONN_TTY_BAUD_150,
 } neu_conn_tty_baud_e;
 
 typedef enum neu_conn_tty_parity {

persistence/0031_2.4.2_ekuiper_node.sql.ex

@@ -0,0 +1,45 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2023 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+BEGIN TRANSACTION;
+
+-- Update default data-stream-processing node setting when
+-- the data-stream-processing node exists and is at init state.
+-- The init state implies the user has not provide any setting yet.
+INSERT INTO
+  settings (node_name, setting)
+SELECT
+  name,
+  '{"params":{"host":"0.0.0.0","port":7081}}'
+FROM
+  nodes
+WHERE
+  name = 'data-stream-processing'
+  and state = 1;
+
+-- Update default data-stream-processing node to running state when
+-- the data-stream-processing node exists and is at init state.
+UPDATE
+  nodes
+SET
+  state = 3
+where
+  name = 'data-stream-processing'
+  and state = 1;
+
+COMMIT;

plugins/file/file_req.c

@@ -32,6 +32,10 @@ static void plugin_group_free(neu_plugin_group_t *pgp);
 
 int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
 {
+    neu_adapter_update_metric_cb_t update_metric =
+        plugin->common.adapter_callbacks->update_metric;
+    int64_t rtt = NEU_METRIC_LAST_RTT_MS_MAX;
+
     struct file_group_data *gd     = NULL;
     int                     length = plugin->file_length;
 
@@ -83,6 +87,8 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         strncpy(dvalue.value.str, buf, strlen(buf));
         fclose(f);
 
+        rtt = 1;
+
     dvalue_result:
         plugin->common.adapter_callbacks->driver.update(
             plugin->common.adapter, group->group_name, tag->name, dvalue);
@@ -90,6 +96,8 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         free(buf);
     }
 
+    update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, rtt, NULL);
+
     return 0;
 }
 

plugins/mqtt/mqtt.json

@@ -1,9 +1,9 @@
 {
   "client-id": {
     "name": "Client ID",
-    "name_zh": "客户端ID",
-    "description": "Neuron MQTT client id",
-    "description_zh": "Neuron MQTT客户端ID",
+    "name_zh": "客户端 ID",
+    "description": "MQTT client id",
+    "description_zh": "MQTT 客户端 ID",
     "type": "string",
     "attribute": "required",
     "default": "${node-name}",
@@ -15,31 +15,43 @@
     "name": "QoS Level",
     "name_zh": "QoS 等级",
     "description": "MQTT QoS level for message delivery",
-    "description_zh": "MQTT消息传输使用的服务质量等级",
-    "type": "int",
+    "description_zh": "MQTT 消息传输使用的服务质量等级",
+    "type": "map",
     "attribute": "optional",
     "default": 0,
     "valid": {
-      "min": 0,
-      "max": 2
+      "map": [
+        {
+          "key": "QoS 0",
+          "value": 0
+        },
+        {
+          "key": "QoS 1",
+          "value": 1
+        },
+        {
+          "key": "QoS 2",
+          "value": 2
+        }
+      ]
     }
   },
   "format": {
     "name": "Upload Format",
     "name_zh": "上报数据格式",
-    "description": "JSON format of the data reported in the upload topic. In values mode, data are split into `values` and `errors` sub objects. In tags mode, tag data are put in a single array.",
-    "description_zh": "上报数据的JSON格式。在values模式下，数据被分为`values`和`errors`两个子对象。在tags模式下，tag数据被放在一个数组中。",
+    "description": "JSON format of the data reported. In values-mode, data are split into `values` and `errors` sub objects. In tags-mode, tag data are put in a single array.",
+    "description_zh": "上报数据的 JSON 格式。在 values-format 格式下，数据被分为 `values` 和 `errors` 两个子对象。在 tags-format 格式下，数据被放在一个数组中。",
     "attribute": "required",
     "type": "map",
     "default": 0,
     "valid": {
       "map": [
         {
-          "key": "format-values",
+          "key": "values-format",
           "value": 0
         },
         {
-          "key": "format-tags",
+          "key": "tags-format",
           "value": 1
         }
       ]
@@ -49,7 +61,7 @@
     "name": "Cache Memory Size (MB)",
     "name_zh": "缓存内存大小（MB）",
     "description": "Max in-memory cache size in megabytes when MQTT connection exception occurs. Should be smaller than cache disk size.",
-    "description_zh": "当MQTT连接异常时，最大的内存缓存大小（单位：MB）。应该小于缓存磁盘大小。",
+    "description_zh": "当 MQTT 连接异常时，最大的内存缓存大小（单位：MB）。应该小于缓存磁盘大小。",
     "type": "int",
     "attribute": "required",
     "default": 0,
@@ -62,7 +74,7 @@
     "name": "Cache Disk Size (MB)",
     "name_zh": "缓存磁盘大小（MB）",
     "description": "Max in-disk cache size in megabytes when MQTT connection exception occurs. Should be larger than cache memory size. If nonzero, cache memory size should also be nonzero.",
-    "description_zh": "当MQTT连接异常时，最大的磁盘缓存大小（单位：MB）。应该大于缓存内存大小。如果不为0，缓存内存大小也不应该为0。",
+    "description_zh": "当 MQTT 连接异常时，最大的磁盘缓存大小（单位：MB）。应该大于缓存内存大小。如果不为 0，缓存内存大小也应该不为 0。",
     "type": "int",
     "attribute": "required",
     "default": 0,
@@ -75,7 +87,7 @@
     "name": "Broker Host",
     "name_zh": "服务器地址",
     "description": "MQTT broker host",
-    "description_zh": "MQTT服务器地址",
+    "description_zh": "MQTT 服务器地址",
     "attribute": "required",
     "type": "string",
     "default": "broker.emqx.io",
@@ -87,7 +99,7 @@
     "name": "Broker Port",
     "name_zh": "服务器端口",
     "description": "MQTT broker port",
-    "description_zh": "MQTT服务器端口",
+    "description_zh": "MQTT 服务器端口",
     "attribute": "required",
     "type": "int",
     "default": 1883,
@@ -99,8 +111,8 @@
   "username": {
     "name": "Username",
     "name_zh": "用户名",
-    "description": "User name",
-    "description_zh": "用户名",
+    "description": "MQTT user name",
+    "description_zh": "MQTT 用户名",
     "attribute": "optional",
     "type": "string",
     "default": "",
@@ -111,8 +123,8 @@
   "password": {
     "name": "Password",
     "name_zh": "密码",
-    "description": "Password",
-    "description_zh": "密码",
+    "description": "MQTT user password",
+    "description_zh": "MQTT 用户密码",
     "attribute": "optional",
     "type": "string",
     "default": "",
@@ -124,7 +136,7 @@
     "name": "SSL",
     "name_zh": "SSL",
     "description": "Enable SSL connection",
-    "description_zh": "是否启用SSL连接",
+    "description_zh": "是否启用 SSL 连接",
     "attribute": "optional",
     "type": "bool",
     "default": false,
@@ -132,9 +144,9 @@
   },
   "ca": {
     "name": "CA",
-    "name_zh": "CA证书",
-    "description": "CA certificate file",
-    "description_zh": "CA证书文件",
+    "name_zh": "CA 证书",
+    "description": "CA certificate which signs the server certificate",
+    "description_zh": "签发服务器证书的 CA 证书",
     "attribute": "required",
     "type": "file",
     "condition": {
@@ -148,8 +160,8 @@
   "cert": {
     "name": "Client Cert",
     "name_zh": "客户端证书",
-    "description": "Client x509 certificate file",
-    "description_zh": "客户端证书文件",
+    "description": "Client x509 certificate when using two way authentication",
+    "description_zh": "使用双向认证时，客户端的 x509 证书",
     "attribute": "optional",
     "type": "file",
     "condition": {
@@ -163,8 +175,8 @@
   "key": {
     "name": "Client Private Key",
     "name_zh": "客户端私钥",
-    "description": "Client private key file",
-    "description_zh": "客户端私钥文件",
+    "description": "Client private key when using two way authentication",
+    "description_zh": "使用双向认证时，客户端的私钥",
     "attribute": "optional",
     "type": "file",
     "condition": {
@@ -177,9 +189,9 @@
   },
   "keypass": {
     "name": "Client Private Key Password",
-    "name_zh": "私钥密码",
-    "description": "Client private key password",
-    "description_zh": "私钥密码",
+    "name_zh": "客户端私钥密码",
+    "description": "Client private key password when using two way authentication with an encrypted client private key",
+    "description_zh": "使用加密的客户端私钥进行双向认证时，客户端私钥的密码",
     "attribute": "optional",
     "type": "string",
     "condition": {

plugins/restful/log_handle.c

@@ -42,8 +42,6 @@
 #include "log_handle.h"
 #include "utils/utarray.h"
 
-UT_array *collect_log_files();
-
 void handle_logs_files(nng_aio *aio)
 {
     void * data = NULL;
@@ -114,94 +112,43 @@ void handle_log_level(nng_aio *aio)
     NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_update_log_level_req_t,
         neu_json_decode_update_log_level_req, {
-            char node_name[NEU_NODE_NAME_LEN] = { 0 };
+            int                        ret    = 0;
+            neu_reqresp_head_t         header = { 0 };
+            neu_req_update_log_level_t cmd    = { 0 };
+            zlog_category_t *          ct = zlog_get_category(req->node_name);
+            zlog_category_t *          neuron = zlog_get_category("neuron");
 
-            UT_array *log_files = collect_log_files();
-
-            utarray_foreach(log_files, char **, log_file)
-            {
-                if (strncmp(*log_file, req->node_name, strlen(*log_file) - 4) ==
-                    0) {
-                    strcpy(node_name, req->node_name);
-                }
-            }
-
-            if (strlen(node_name) == 0) {
-                NEU_JSON_RESPONSE_ERROR(NEU_ERR_NODE_NOT_EXIST, {
+            ret = zlog_level_switch(ct, ZLOG_LEVEL_DEBUG);
+            if (ret != 0) {
+                nlog_error("Modify node log level fail, node_name:%s, "
+                           "ret: %d,",
+                           req->node_name, ret);
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
                     neu_http_response(aio, error_code.error, result_error);
                 });
-            } else {
-                int                        ret    = 0;
-                neu_reqresp_head_t         header = { 0 };
-                neu_req_update_log_level_t cmd    = { 0 };
-                zlog_category_t *          ct = zlog_get_category(node_name);
-                zlog_category_t *          neuron = zlog_get_category("neuron");
-
-                ret = zlog_level_switch(ct, ZLOG_LEVEL_DEBUG);
-                if (ret != 0) {
-                    nlog_error("Modify node log level fail, node_name:%s, "
-                               "ret: %d,",
-                               node_name, ret);
-                    NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
-                        neu_http_response(aio, error_code.error, result_error);
-                    });
-                }
-
-                ret = zlog_level_switch(neuron, ZLOG_LEVEL_DEBUG);
-                if (ret != 0) {
-                    nlog_error("Modify neuron log level fail, ret: %d", ret);
-                    NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
-                        neu_http_response(aio, error_code.error, result_error);
-                    });
-                }
-
-                header.ctx  = aio;
-                header.type = NEU_REQ_UPDATE_LOG_LEVEL;
-                strcpy(cmd.node, req->node_name);
-
-                ret = neu_plugin_op(plugin, header, &cmd);
-                if (ret != 0) {
-                    NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
-                        neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
-                    });
-                }
             }
 
-            utarray_free(log_files);
+            ret = zlog_level_switch(neuron, ZLOG_LEVEL_DEBUG);
+            if (ret != 0) {
+                nlog_error("Modify neuron log level fail, ret: %d", ret);
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
+                    neu_http_response(aio, error_code.error, result_error);
+                });
+            }
+
+            header.ctx  = aio;
+            header.type = NEU_REQ_UPDATE_LOG_LEVEL;
+            strcpy(cmd.node, req->node_name);
+
+            ret = neu_plugin_op(plugin, header, &cmd);
+            if (ret != 0) {
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
+                    neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
+                });
+            }
         })
 }
 
-static inline bool ends_with(const char *str, const char *suffix)
-{
-    size_t m = strlen(str);
-    size_t n = strlen(suffix);
-    return m >= n && !strcmp(str + m - n, suffix);
-}
-
-UT_array *collect_log_files()
-{
-    DIR *dp = opendir("./logs");
-    if (NULL == dp) {
-        nlog_error("open logs directory fail");
-        return NULL;
-    }
-
-    UT_array *log_files = NULL;
-    utarray_new(log_files, &ut_str_icd);
-
-    struct dirent *de = NULL;
-    while (NULL != (de = readdir(dp))) {
-        char *name = de->d_name;
-        if (ends_with(de->d_name, ".log")) {
-            utarray_push_back(log_files, &name);
-        }
-    }
-
-    closedir(dp);
-
-    return log_files;
-}
-
 int read_file(const char *file_name, void **datap, size_t *lenp)
 {
     int         rv = 0;

src/adapter/driver/driver.c

@@ -302,6 +302,8 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
                         (double) (cmd->value.value.d64 / tag->decimal);
                 } else if (cmd->value.type == NEU_TYPE_DOUBLE) {
                     cmd->value.value.d64 = cmd->value.value.d64 / tag->decimal;
+                } else if (cmd->value.type == NEU_TYPE_FLOAT) {
+                    cmd->value.value.f32 = cmd->value.value.f32 / tag->decimal;
                 }
                 break;
             default:
@@ -360,8 +362,11 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             }
             break;
         case NEU_TYPE_FLOAT:
-            cmd->value.type      = NEU_TYPE_FLOAT;
-            cmd->value.value.f32 = (float) cmd->value.value.d64;
+            if (NEU_TYPE_FLOAT != cmd->value.type) {
+                cmd->value.value.f32 = (float) cmd->value.value.d64;
+                cmd->value.type      = NEU_TYPE_FLOAT;
+            }
+
             switch (tag->option.value32.endian) {
             case NEU_DATATAG_ENDIAN_LB32:
                 neu_ntohs_p((uint16_t *) cmd->value.value.bytes);

src/connection/connection.c

@@ -69,6 +69,8 @@ struct neu_conn {
 static void conn_tcp_server_add_client(neu_conn_t *conn, int fd,
                                        struct sockaddr_in client);
 static void conn_tcp_server_del_client(neu_conn_t *conn, int fd);
+static int  conn_tcp_server_replace_client(neu_conn_t *conn, int fd,
+                                           struct sockaddr_in client);
 
 static void conn_tcp_server_listen(neu_conn_t *conn);
 static void conn_tcp_server_stop(neu_conn_t *conn);
@@ -180,16 +182,6 @@ int neu_conn_tcp_server_accept(neu_conn_t *conn)
         return -1;
     }
 
-    if (conn->tcp_server.n_client >= conn->param.params.tcp_server.max_link) {
-        close(fd);
-        zlog_warn(conn->param.log, "%s:%d accpet max link: %d, reject",
-                  conn->param.params.tcp_server.ip,
-                  conn->param.params.tcp_server.port,
-                  conn->param.params.tcp_server.max_link);
-        pthread_mutex_unlock(&conn->mtx);
-        return -1;
-    }
-
     if (conn->block) {
         struct timeval tv = {
             .tv_sec  = conn->param.params.tcp_server.timeout / 1000,
@@ -199,7 +191,23 @@ int neu_conn_tcp_server_accept(neu_conn_t *conn)
         setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
     }
 
-    conn_tcp_server_add_client(conn, fd, client);
+    if (conn->tcp_server.n_client >= conn->param.params.tcp_server.max_link) {
+        int free_fd = conn_tcp_server_replace_client(conn, fd, client);
+        if (free_fd > 0) {
+            zlog_warn(conn->param.log, "replace old client %d with %d", free_fd,
+                      fd);
+        } else {
+            close(fd);
+            zlog_warn(conn->param.log, "%s:%d accpet max link: %d, reject",
+                      conn->param.params.tcp_server.ip,
+                      conn->param.params.tcp_server.port,
+                      conn->param.params.tcp_server.max_link);
+            pthread_mutex_unlock(&conn->mtx);
+            return -1;
+        }
+    } else {
+        conn_tcp_server_add_client(conn, fd, client);
+    }
 
     conn->is_connected = true;
     conn->connected(conn->data, fd);
@@ -728,6 +736,30 @@ static void conn_connect(neu_conn_t *conn)
             cfsetospeed(&tty_opt, B2400);
             cfsetispeed(&tty_opt, B2400);
             break;
+        case NEU_CONN_TTY_BAUD_1800:
+            cfsetospeed(&tty_opt, B1800);
+            cfsetispeed(&tty_opt, B1800);
+            break;
+        case NEU_CONN_TTY_BAUD_1200:
+            cfsetospeed(&tty_opt, B1200);
+            cfsetispeed(&tty_opt, B1200);
+            break;
+        case NEU_CONN_TTY_BAUD_600:
+            cfsetospeed(&tty_opt, B600);
+            cfsetispeed(&tty_opt, B600);
+            break;
+        case NEU_CONN_TTY_BAUD_300:
+            cfsetospeed(&tty_opt, B300);
+            cfsetispeed(&tty_opt, B300);
+            break;
+        case NEU_CONN_TTY_BAUD_200:
+            cfsetospeed(&tty_opt, B200);
+            cfsetispeed(&tty_opt, B200);
+            break;
+        case NEU_CONN_TTY_BAUD_150:
+            cfsetospeed(&tty_opt, B150);
+            cfsetispeed(&tty_opt, B150);
+            break;
         }
 
         switch (conn->param.params.tty_client.parity) {
@@ -867,6 +899,25 @@ static void conn_tcp_server_del_client(neu_conn_t *conn, int fd)
     }
 }
 
+static int conn_tcp_server_replace_client(neu_conn_t *conn, int fd,
+                                          struct sockaddr_in client)
+{
+    for (int i = 0; i < conn->param.params.tcp_server.max_link; i++) {
+        if (conn->tcp_server.clients[i].fd > 0) {
+            int ret = conn->tcp_server.clients[i].fd;
+
+            close(conn->tcp_server.clients[i].fd);
+            conn->tcp_server.clients[i].fd = 0;
+
+            conn->tcp_server.clients[i].fd     = fd;
+            conn->tcp_server.clients[i].client = client;
+            return ret;
+        }
+    }
+
+    return 0;
+}
+
 int neu_conn_stream_consume(neu_conn_t *conn, void *context,
                             neu_conn_stream_consume_fn fn)
 {

src/connection/mqtt_client.c

@@ -150,6 +150,7 @@ static void connect_cb(nng_pipe p, nng_pipe_ev ev, void *arg);
 
 static inline task_t *client_alloc_task(neu_mqtt_client_t *client);
 static inline void    client_free_task(neu_mqtt_client_t *client, task_t *task);
+static inline size_t  client_task_free_list_len(neu_mqtt_client_t *client);
 static inline void    client_add_subscription(neu_mqtt_client_t *client,
                                               subscription_t *   sub);
 static int            client_send_sub_msg(neu_mqtt_client_t *client,
@@ -545,6 +546,16 @@ static inline void client_free_task(neu_mqtt_client_t *client, task_t *task)
     DL_PREPEND(client->task_free_list, task);
 }
 
+static inline size_t client_task_free_list_len(neu_mqtt_client_t *client)
+{
+    size_t  count = 0;
+    task_t *elt   = NULL;
+    // count by iterating the list,
+    // this won't be a problem as a short free list is expected
+    DL_COUNT(client->task_free_list, elt, count);
+    return count;
+}
+
 static inline void client_add_subscription(neu_mqtt_client_t *client,
                                            subscription_t *   sub)
 {
@@ -1149,6 +1160,12 @@ int neu_mqtt_client_close(neu_mqtt_client_t *client)
     }
 
     nng_mtx_lock(client->mtx);
+    // wait for all tasks
+    while (client_task_free_list_len(client) != client->task_count) {
+        nng_mtx_unlock(client->mtx);
+        nng_msleep(100);
+        nng_mtx_lock(client->mtx);
+    }
     client->open = false;
     nng_mtx_unlock(client->mtx);
 

version

@@ -1 +1 @@
-2.4.0
+2.4.2

zlog.conf

@@ -8,4 +8,4 @@ simple = "%d [%V] %f:%L %m%n"
 
 [rules]
 
-*.*     "./logs/%c.log", 100MB * 10 ~ "./logs/%c.#2s.log"; simple
+*.*     "./logs/%c.log", 50MB * 1 ~ "./logs/%c.#2s.log"; simple