Merge pull request #1263 from fengzeroz/v2.4

version: 2.4.6

.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.6
+  UI_VERSION: 2.4.4
+  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

.gitignore

@@ -16,4 +16,7 @@ tmp
 *.deb
 *.rpm
 neuron-sdk
-.vscode
+.vscode
+log.html
+output.xml
+report.html

cmake/cross.cmake

@@ -17,3 +17,7 @@ set(CMAKE_FIND_ROOT_PATH_MODE_PROGRAM NEVER)
 set(CMAKE_FIND_ROOT_PATH_MODE_LIBRARY ONLY)
 set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
 link_directories(${CMAKE_STAGING_PREFIX})
+
+if(CMAKE_SYSTEM_PROCESSOR MATCHES "mips32")
+	link_libraries(atomic)
+endif()

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

ft/resource/error.resource

@@ -71,4 +71,7 @@ ${NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED} 	       	3008
 ${NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE}	3009
 ${NEU_ERR_PLUGIN_NOT_RUNNING}			3010
 ${NEU_ERR_PLUGIN_TAG_NOT_READY}           	3011
-${NEU_ERR_PLUGIN_PACKET_OUT_OF_ORDER}     	3012
+${NEU_ERR_PLUGIN_PACKET_OUT_OF_ORDER}     	3012
+${NEU_ERR_PLUGIN_NAME_TOO_LONG}         	3013
+${NEU_ERR_PLUGIN_NOT_FOUND}             	3014
+${NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE}     	3015

include/neuron/connection/mqtt_client.h

@@ -114,6 +114,8 @@ int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
                               const char *topic, void *data,
                               neu_mqtt_client_subscribe_cb_t cb);
 
+int neu_mqtt_client_unsubscribe(neu_mqtt_client_t *client, const char *topic);
+
 #ifdef __cplusplus
 }
 #endif

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 {

include/neuron/errcodes.h

@@ -113,6 +113,7 @@ typedef enum {
     NEU_ERR_PLUGIN_PACKET_OUT_OF_ORDER     = 3012,
     NEU_ERR_PLUGIN_NAME_TOO_LONG           = 3013,
     NEU_ERR_PLUGIN_NOT_FOUND               = 3014,
+    NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE     = 3015,
 
     NEU_ERR_MQTT_FAILURE                        = 4000,
     NEU_ERR_MQTT_NO_CERTFILESET                 = 4001,

include/neuron/metrics.h

@@ -154,6 +154,7 @@ typedef struct {
     unsigned            cpu_cores;
     size_t              mem_total_bytes;
     size_t              mem_used_bytes;
+    size_t              mem_cache_bytes;
     size_t              disk_size_gibibytes;
     size_t              disk_used_gibibytes;
     size_t              disk_avail_gibibytes;

include/neuron/utils/protocol_buf.h

@@ -192,4 +192,16 @@ neu_protocol_unpack_buf_used_size(neu_protocol_unpack_buf_t *buf)
     return buf->offset;
 }
 
+inline static void
+neu_protocol_unpack_buf_use_all(neu_protocol_unpack_buf_t *buf)
+{
+    buf->offset = buf->size;
+}
+
+inline static uint16_t
+neu_protocol_unpack_buf_size(neu_protocol_unpack_buf_t *buf)
+{
+    return buf->size;
+}
+
 #endif

include/neuron/utils/time.h

@@ -26,6 +26,7 @@ extern "C" {
 #include <stdint.h>
 #include <stdlib.h>
 #include <sys/time.h>
+#include <time.h>
 
 static inline int64_t neu_time_ms()
 {
@@ -34,6 +35,15 @@ static inline int64_t neu_time_ms()
     return (int64_t) tv.tv_sec * 1000 + (int64_t) tv.tv_usec / 1000;
 }
 
+static inline void neu_msleep(unsigned msec)
+{
+    struct timespec tv = {
+        .tv_sec  = msec / 1000,
+        .tv_nsec = (msec % 1000) * 1000000,
+    };
+    nanosleep(&tv, NULL);
+}
+
 #ifdef __cplusplus
 }
 #endif

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;

persistence/0032_2.4.4_plugin_name.sql

@@ -0,0 +1,110 @@
+/**
+ * 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 nodes
+SET plugin_name = 'MQTT'
+WHERE plugin_name = 'mqtt';
+
+UPDATE nodes
+SET plugin_name = 'eKuiper'
+WHERE plugin_name = 'ekuiper';
+
+UPDATE nodes
+SET plugin_name = 'SparkPlugB'
+WHERE plugin_name = 'sparkplugb';
+
+UPDATE nodes
+SET plugin_name = 'Beckhoff ADS'
+WHERE plugin_name = 'ads';
+
+UPDATE nodes
+SET plugin_name = 'Modbus TCP community'
+WHERE plugin_name = 'modbus-tcp';
+
+UPDATE nodes
+SET plugin_name = 'Modbus RTU'
+WHERE plugin_name = 'modbus-rtu';
+
+UPDATE nodes
+SET plugin_name = 'Modbus TCP'
+WHERE plugin_name = 'modbus-plus-tcp';
+
+UPDATE nodes
+SET plugin_name = 'Modbus TCP QH'
+WHERE plugin_name = 'modbus-qh-tcp';
+
+UPDATE nodes
+SET plugin_name = 'Siemens S7 ISOTCP'
+WHERE plugin_name = 's7comm';
+
+UPDATE nodes
+SET plugin_name = 'Siemens S7 ISOTCP for 300/400'
+WHERE plugin_name = 's7comm_for_300';
+
+UPDATE nodes
+SET plugin_name = 'OPC UA'
+WHERE plugin_name = 'opcua';
+
+UPDATE nodes
+SET plugin_name = 'Omron FINS TCP'
+WHERE plugin_name = 'fins';
+
+UPDATE nodes
+SET plugin_name = 'Mitsubishi 3E'
+WHERE plugin_name = 'qna3e';
+
+UPDATE nodes
+SET plugin_name = 'Mitsubishi 1E'
+WHERE plugin_name = 'a1e';
+
+UPDATE nodes
+SET plugin_name = 'IEC60870-5-104'
+WHERE plugin_name = 'iec104';
+
+UPDATE nodes
+SET plugin_name = 'BACnet/IP'
+WHERE plugin_name = 'bacnet';
+
+UPDATE nodes
+SET plugin_name = 'KNXnet/IP'
+WHERE plugin_name = 'knx';
+
+UPDATE nodes
+SET plugin_name = 'DLT645-2007'
+WHERE plugin_name = 'dlt645';
+
+UPDATE nodes
+SET plugin_name = 'NON A11'
+WHERE plugin_name = 'nona11';
+
+UPDATE nodes
+SET plugin_name = 'EtherNet/IP(CIP)'
+WHERE plugin_name = 'EtherNet-IP';
+
+UPDATE nodes
+SET plugin_name = 'Fanuc Focas Ethernet'
+WHERE plugin_name = 'focas';
+
+UPDATE nodes
+SET plugin_name = 'File'
+WHERE plugin_name = 'file';
+
+COMMIT;

persistence/0033_2.4.5_mqtt_upload_topic.sql

@@ -0,0 +1,59 @@
+/**
+ * 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;
+
+-- for backward compatibility with version 2.3,
+-- migrate MQTT plugin upload-topic to subscription parameters.
+WITH
+  topics AS (
+    SELECT
+      name AS app,
+      json_extract(json(setting), '$.params.upload-topic') AS topic
+    FROM
+      nodes
+      JOIN settings
+    WHERE
+      name = node_name
+      AND plugin_name = 'MQTT'
+  )
+UPDATE
+  subscriptions
+SET
+  params = (
+    SELECT
+      json_object('params', json_object('topic', topic)) AS params
+    FROM
+      topics
+    WHERE
+      topics.app = subscriptions.app_name
+      AND topics.topic NOT NULL
+  )
+WHERE
+  EXISTS (
+    SELECT
+      topic
+    FROM
+      topics
+    WHERE
+      topics.app = subscriptions.app_name
+      AND topics.topic NOT NULL
+      AND subscriptions.params IS NULL
+  );
+
+COMMIT;

plugins/file/file_plugin.c

@@ -81,8 +81,7 @@ static int driver_close(neu_plugin_t *plugin)
 
 static int driver_init(neu_plugin_t *plugin)
 {
-    plugin->events            = neu_event_new();
-    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
+    plugin->events = neu_event_new();
 
     return 0;
 }
@@ -97,14 +96,14 @@ static int driver_uninit(neu_plugin_t *plugin)
 
 static int driver_start(neu_plugin_t *plugin)
 {
-    (void) plugin;
+    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
 
     return 0;
 }
 
 static int driver_stop(neu_plugin_t *plugin)
 {
-    (void) plugin;
+    plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
 
     return 0;
 }
@@ -121,7 +120,8 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
         return -1;
     }
 
-    plugin->file_length = file_length.v.val_int;
+    plugin->file_length       = file_length.v.val_int;
+    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
 
     return 0;
 }

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;
 
@@ -56,6 +60,7 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
             dvalue.value.i32 = NEU_ERR_FILE_NOT_EXIST;
             goto dvalue_result;
         }
+
         if ((f = fopen(tag->address, "rb+")) == NULL) {
             nlog_error("open file:%s", tag->address);
             dvalue.type      = NEU_TYPE_ERROR;
@@ -63,24 +68,30 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
             goto dvalue_result;
         }
 
+        fseek(f, 0, SEEK_END);
+        int size = ftell(f);
+        if (size >= length) {
+            dvalue.type      = NEU_TYPE_ERROR;
+            dvalue.value.i32 = NEU_ERR_FILE_TOO_LONG;
+
+            fclose(f);
+            goto dvalue_result;
+        }
+
+        fseek(f, 0, SEEK_SET);
         int ret = fread(buf, sizeof(char), length, f);
         if (ret == 0) {
             nlog_error("read file failed:%s", tag->address);
             dvalue.type      = NEU_TYPE_ERROR;
             dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
-            goto dvalue_result;
-        }
 
-        fseek(f, 0, SEEK_END);
-        int size = ftell(f);
-        if (size > length) {
-            dvalue.type      = NEU_TYPE_ERROR;
-            dvalue.value.i32 = NEU_ERR_FILE_TOO_LONG;
+            fclose(f);
             goto dvalue_result;
         }
 
         dvalue.type = NEU_TYPE_STRING;
         strncpy(dvalue.value.str, buf, strlen(buf));
+        rtt = 1;
         fclose(f);
 
     dvalue_result:
@@ -90,6 +101,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/monitor/metric_handle.c

@@ -31,10 +31,10 @@
 
 // clang-format off
 #define METRIC_GLOBAL_TMPL                                                       \
-    "# HELP distro OS distribution\n"                                            \
-    "distro %s\n"                                                                \
-    "# HELP kernel OS kernel version\n"                                          \
-    "kernel %s\n"                                                                \
+    "# HELP os_info OS distro and kernel version\n"                              \
+    "# TYPE os_info gauge\n"                                                     \
+    "os_info{version=\"%s\"} 0\n"                                                \
+    "os_info{kernel=\"%s\"} 0\n"                                                 \
     "# HELP cpu_percent Total CPU utilisation percentage\n"                      \
     "# TYPE cpu_percent gauge\n"                                                 \
     "cpu_percent %u\n"                                                           \
@@ -47,6 +47,12 @@
     "# HELP mem_used_bytes Used memory in bytes\n"                               \
     "# TYPE mem_used_bytes gauge\n"                                              \
     "mem_used_bytes %zu\n"                                                       \
+    "# HELP mem_cache Memory buffer/cache size in bytes\n"                       \
+    "# TYPE mem_cache gauge\n"                                                   \
+    "mem_cache_bytes %zu\n"                                                      \
+    "# HELP rss_bytes RSS (Resident Set Size) in bytes\n"                        \
+    "# TYPE rss_bytes gauge\n"                                                   \
+    "rss_bytes %zu\n"                                                            \
     "# HELP disk_size_gibibytes Disk size in gibibytes\n"                        \
     "# TYPE disk_size_gibibytes counter\n"                                       \
     "disk_size_gibibytes %zu\n"                                                  \
@@ -141,6 +147,7 @@ static inline void gen_global_metrics(const neu_metrics_t *metrics,
 {
     fprintf(stream, METRIC_GLOBAL_TMPL, metrics->distro, metrics->kernel,
             metrics->cpu_percent, metrics->cpu_cores, metrics->mem_total_bytes,
+            metrics->mem_used_bytes, metrics->mem_cache_bytes,
             metrics->mem_used_bytes, metrics->disk_size_gibibytes,
             metrics->disk_used_gibibytes, metrics->disk_avail_gibibytes,
             metrics->core_dumped, metrics->uptime_seconds, metrics->north_nodes,

plugins/monitor/monitor.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}",
@@ -27,7 +27,7 @@
     "name": "Heartbeat Interval",
     "name_zh": "心跳间隔",
     "description": "Interval in seconds between two heartbeat messages. Setting to zero will disable heartbeat monitoring.(seconds)",
-    "description_zh": "两个心跳消息之间的间隔时间，设置为0将禁用心跳监控。(秒)",
+    "description_zh": "每个心跳消息之间的间隔时间（秒），设置为 0 将禁用心跳消息。",
     "attribute": "required",
     "type": "int",
     "default": 0,
@@ -51,7 +51,7 @@
     "name": "Broker Host",
     "name_zh": "服务器地址",
     "description": "MQTT broker host",
-    "description_zh": "MQTT服务器地址",
+    "description_zh": "MQTT 服务器地址",
     "attribute": "required",
     "type": "string",
     "default": "broker.emqx.io",
@@ -63,7 +63,7 @@
     "name": "Broker Port",
     "name_zh": "服务器端口",
     "description": "MQTT broker port",
-    "description_zh": "MQTT服务器端口",
+    "description_zh": "MQTT 服务器端口",
     "attribute": "required",
     "type": "int",
     "default": 1883,
@@ -75,8 +75,8 @@
   "username": {
     "name": "Username",
     "name_zh": "用户名",
-    "description": "User name",
-    "description_zh": "用户名",
+    "description": "MQTT user name",
+    "description_zh": "MQTT 用户名",
     "attribute": "optional",
     "type": "string",
     "default": "",
@@ -87,8 +87,8 @@
   "password": {
     "name": "Password",
     "name_zh": "密码",
-    "description": "Password",
-    "description_zh": "密码",
+    "description": "MQTT user password",
+    "description_zh": "MQTT 用户密码",
     "attribute": "optional",
     "type": "string",
     "default": "",
@@ -100,7 +100,7 @@
     "name": "SSL",
     "name_zh": "SSL",
     "description": "Enable SSL connection",
-    "description_zh": "是否启用SSL连接",
+    "description_zh": "是否启用 SSL 连接",
     "attribute": "optional",
     "type": "bool",
     "default": false,
@@ -108,9 +108,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": {
@@ -124,8 +124,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": {
@@ -139,8 +139,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": {
@@ -152,10 +152,10 @@
     }
   },
   "keypass": {
-    "name": "Keypass",
-    "name_zh": "私钥密码",
-    "description": "Private key password",
-    "description_zh": "私钥密码",
+    "name": "Client Private Key Password",
+    "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/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,46 +15,95 @@
     "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
         }
       ]
     }
   },
+  "write-req-topic": {
+    "name": "Write Request Topic",
+    "name_zh": "写请求主题",
+    "description": "MQTT topic to receive write request messages.",
+    "description_zh": "接收点位写入请求的 MQTT 主题。",
+    "attribute": "required",
+    "type": "string",
+    "default": "/neuron/${node-name}/write/req",
+    "valid": {
+      "length": 255
+    }
+  },
+  "write-resp-topic": {
+    "name": "Write Response Topic",
+    "name_zh": "写响应主题",
+    "description": "MQTT topic to send write response messages.",
+    "description_zh": "发送点位写入响应的 MQTT 主题。",
+    "attribute": "required",
+    "type": "string",
+    "default": "/neuron/${node-name}/write/resp",
+    "valid": {
+      "length": 255
+    }
+  },
+  "offline-cache": {
+    "name": "Offline Data Caching",
+    "name_zh": "离线缓存",
+    "description": "Offline caching switch. Cache MQTT messages when offline, and sync cached messages when back online.",
+    "description_zh": "离线缓存开关。连接断开时缓存 MQTT 消息，连接重建时同步缓存的 MQTT 消息到服务器。",
+    "attribute": "optional",
+    "type": "bool",
+    "default": false,
+    "valid": {}
+  },
   "cache-mem-size": {
     "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,
+    "condition": {
+      "field": "offline-cache",
+      "value": true
+    },
     "valid": {
-      "min": 0,
+      "min": 1,
       "max": 1024
     }
   },
@@ -62,12 +111,15 @@
     "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,
+    "condition": {
+      "field": "offline-cache",
+      "value": true
+    },
     "valid": {
-      "min": 0,
+      "min": 1,
       "max": 10240
     }
   },
@@ -75,7 +127,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 +139,7 @@
     "name": "Broker Port",
     "name_zh": "服务器端口",
     "description": "MQTT broker port",
-    "description_zh": "MQTT服务器端口",
+    "description_zh": "MQTT 服务器端口",
     "attribute": "required",
     "type": "int",
     "default": 1883,
@@ -99,8 +151,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 +163,8 @@
   "password": {
     "name": "Password",
     "name_zh": "密码",
-    "description": "Password",
-    "description_zh": "密码",
+    "description": "MQTT user password",
+    "description_zh": "MQTT 用户密码",
     "attribute": "optional",
     "type": "string",
     "default": "",
@@ -124,7 +176,7 @@
     "name": "SSL",
     "name_zh": "SSL",
     "description": "Enable SSL connection",
-    "description_zh": "是否启用SSL连接",
+    "description_zh": "是否启用 SSL 连接",
     "attribute": "optional",
     "type": "bool",
     "default": false,
@@ -132,9 +184,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 +200,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 +215,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 +229,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": {
@@ -190,4 +242,4 @@
       "length": 256
     }
   }
-}
+}

plugins/mqtt/mqtt_config.c

@@ -17,6 +17,7 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
+#include "utils/asprintf.h"
 #include "json/json.h"
 #include "json/neu_json_param.h"
 
@@ -115,6 +116,79 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
     return 0;
 }
 
+static int parse_cache_params(neu_plugin_t *plugin, const char *setting,
+                              neu_json_elem_t *offline_cache,
+                              neu_json_elem_t *cache_mem_size,
+                              neu_json_elem_t *cache_disk_size)
+{
+    int   ret          = 0;
+    char *err_param    = NULL;
+    bool  flag_present = true; // whether setting has `offline-cache`
+
+    // offline-cache flag, optional
+    ret = neu_parse_param(setting, NULL, 1, offline_cache);
+    if (0 != ret) {
+        plog_info(plugin, "setting no offline cache flag");
+        flag_present = false; // `offline-cache` not present in setting
+    }
+
+    if (flag_present && !offline_cache->v.val_bool) {
+        // cache explicitly disabled in setting
+        cache_mem_size->v.val_int  = 0;
+        cache_disk_size->v.val_int = 0;
+        return 0;
+    }
+
+    // we are here because one of the following:
+    //    1. `offline-cache` presents to be true in setting
+    //    2. no `offline-cache` in setting
+    // we require `cache_mem_size` and `cache_disk_size` to present in both
+    // cases, especially in case 2 for backward compatibility.
+
+    ret = neu_parse_param(setting, &err_param, 2, cache_mem_size,
+                          cache_disk_size);
+    if (0 != ret) {
+        plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
+        free(err_param);
+        return -1;
+    }
+
+    if (cache_mem_size->v.val_int > 1024 ||
+        (flag_present && cache_mem_size->v.val_int < 1)) {
+        plog_error(plugin, "setting invalid cache memory size: %" PRIi64,
+                   cache_mem_size->v.val_int);
+        return -1;
+    }
+
+    if (cache_disk_size->v.val_int > 10240 ||
+        (flag_present && cache_disk_size->v.val_int < 1)) {
+        plog_error(plugin, "setting invalid cache disk size: %" PRIi64,
+                   cache_disk_size->v.val_int);
+        return -1;
+    }
+
+    if (cache_mem_size->v.val_int > cache_disk_size->v.val_int) {
+        plog_error(plugin,
+                   "setting cache memory size %" PRIi64
+                   " larger than cache disk size %" PRIi64,
+                   cache_mem_size->v.val_int, cache_disk_size->v.val_int);
+        return -1;
+    }
+
+    if (0 == cache_mem_size->v.val_int && 0 != cache_disk_size->v.val_int) {
+        plog_error(plugin,
+                   "setting cache disk size %" PRIi64 " without memory cache",
+                   cache_disk_size->v.val_int);
+        return -1;
+    }
+
+    if (!flag_present) {
+        offline_cache->v.val_bool = cache_mem_size->v.val_int > 0;
+    }
+
+    return 0;
+}
+
 int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
                       mqtt_config_t *config)
 {
@@ -130,6 +204,20 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
     };
     neu_json_elem_t format          = { .name = "format", .t = NEU_JSON_INT };
+    neu_json_elem_t write_req_topic = {
+        .name      = "write-req-topic",
+        .t         = NEU_JSON_STR,
+        .v.val_str = NULL,
+        .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
+    };
+    neu_json_elem_t write_resp_topic = {
+        .name      = "write-resp-topic",
+        .t         = NEU_JSON_STR,
+        .v.val_str = NULL,
+        .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
+    };
+    neu_json_elem_t offline_cache   = { .name = "offline-cache",
+                                      .t    = NEU_JSON_BOOL };
     neu_json_elem_t cache_mem_size  = { .name = "cache-mem-size",
                                        .t    = NEU_JSON_INT };
     neu_json_elem_t cache_disk_size = { .name = "cache-disk-size",
@@ -149,8 +237,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         return -1;
     }
 
-    ret = neu_parse_param(setting, &err_param, 7, &client_id, &format, &qos,
-                          &cache_mem_size, &cache_disk_size, &host, &port);
+    ret = neu_parse_param(setting, &err_param, 7, &client_id, &qos, &format,
+                          &write_req_topic, &write_resp_topic, &host, &port);
     if (0 != ret) {
         plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
         goto error;
@@ -176,32 +264,26 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
-    // cache memory size, required
-    if (cache_mem_size.v.val_int > 1024) {
-        plog_error(plugin, "setting invalid cache memory size: %" PRIi64,
-                   cache_mem_size.v.val_int);
+    // write request topic
+    if (NULL == write_req_topic.v.val_str &&
+        0 > neu_asprintf(&write_req_topic.v.val_str, "/neuron/%s/write/req",
+                         plugin->common.name)) {
+        plog_error(plugin, "setting write request topic error");
         goto error;
     }
 
-    // cache disk size, required
-    if (cache_disk_size.v.val_int > 10240) {
-        plog_error(plugin, "setting invalid cache disk size: %" PRIi64,
-                   cache_disk_size.v.val_int);
+    // write response topic
+    if (NULL == write_resp_topic.v.val_str &&
+        0 > neu_asprintf(&write_resp_topic.v.val_str, "/neuron/%s/write/resp",
+                         plugin->common.name)) {
+        plog_error(plugin, "setting write response topic error");
         goto error;
     }
 
-    if (cache_mem_size.v.val_int > cache_disk_size.v.val_int) {
-        plog_error(plugin,
-                   "setting cache memory size %" PRIi64
-                   " larger than cache disk size %" PRIi64,
-                   cache_mem_size.v.val_int, cache_disk_size.v.val_int);
-        goto error;
-    }
-
-    if (0 == cache_mem_size.v.val_int && 0 != cache_disk_size.v.val_int) {
-        plog_error(plugin,
-                   "setting cache disk size %" PRIi64 " without memory cache",
-                   cache_disk_size.v.val_int);
+    // offline cache
+    ret = parse_cache_params(plugin, setting, &offline_cache, &cache_mem_size,
+                             &cache_disk_size);
+    if (0 != ret) {
         goto error;
     }
 
@@ -234,56 +316,61 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
-    config->client_id = client_id.v.val_str;
-    config->qos       = qos.v.val_int;
-    config->format    = format.v.val_int;
-    config->cache =
-        0 != cache_mem_size.v.val_int || 0 != cache_disk_size.v.val_int;
-    config->cache_mem_size  = cache_mem_size.v.val_int * MB;
-    config->cache_disk_size = cache_disk_size.v.val_int * MB;
-    config->host            = host.v.val_str;
-    config->port            = port.v.val_int;
-    config->username        = username.v.val_str;
-    config->password        = password.v.val_str;
-    config->ca              = ca.v.val_str;
-    config->cert            = cert.v.val_str;
-    config->key             = key.v.val_str;
-    config->keypass         = keypass.v.val_str;
+    config->client_id        = client_id.v.val_str;
+    config->qos              = qos.v.val_int;
+    config->format           = format.v.val_int;
+    config->write_req_topic  = write_req_topic.v.val_str;
+    config->write_resp_topic = write_resp_topic.v.val_str;
+    config->cache            = offline_cache.v.val_bool;
+    config->cache_mem_size   = cache_mem_size.v.val_int * MB;
+    config->cache_disk_size  = cache_disk_size.v.val_int * MB;
+    config->host             = host.v.val_str;
+    config->port             = port.v.val_int;
+    config->username         = username.v.val_str;
+    config->password         = password.v.val_str;
+    config->ca               = ca.v.val_str;
+    config->cert             = cert.v.val_str;
+    config->key              = key.v.val_str;
+    config->keypass          = keypass.v.val_str;
 
-    plog_info(plugin, "config client-id      : %s", config->client_id);
-    plog_info(plugin, "config qos            : %d", config->qos);
-    plog_info(plugin, "config format         : %s",
+    plog_info(plugin, "config client-id       : %s", config->client_id);
+    plog_info(plugin, "config qos             : %d", config->qos);
+    plog_info(plugin, "config format          : %s",
               mqtt_upload_format_str(config->format));
-    plog_info(plugin, "config cache          : %zu", config->cache);
-    plog_info(plugin, "config cache-mem-size : %zu", config->cache_mem_size);
-    plog_info(plugin, "config cache-disk-size: %zu", config->cache_disk_size);
-    plog_info(plugin, "config host           : %s", config->host);
-    plog_info(plugin, "config port           : %" PRIu16, config->port);
+    plog_info(plugin, "config write-req-topic : %s", config->write_req_topic);
+    plog_info(plugin, "config write-resp-topic: %s", config->write_resp_topic);
+    plog_info(plugin, "config cache           : %zu", config->cache);
+    plog_info(plugin, "config cache-mem-size  : %zu", config->cache_mem_size);
+    plog_info(plugin, "config cache-disk-size : %zu", config->cache_disk_size);
+    plog_info(plugin, "config host            : %s", config->host);
+    plog_info(plugin, "config port            : %" PRIu16, config->port);
 
     if (config->username) {
-        plog_info(plugin, "config username       : %s", config->username);
+        plog_info(plugin, "config username        : %s", config->username);
     }
     if (config->password) {
-        plog_info(plugin, "config password       : %s",
+        plog_info(plugin, "config password        : %s",
                   0 == strlen(config->password) ? "" : placeholder);
     }
     if (config->ca) {
-        plog_info(plugin, "config ca             : %s", placeholder);
+        plog_info(plugin, "config ca              : %s", placeholder);
     }
     if (config->cert) {
-        plog_info(plugin, "config cert           : %s", placeholder);
+        plog_info(plugin, "config cert            : %s", placeholder);
     }
     if (config->key) {
-        plog_info(plugin, "config key            : %s", placeholder);
+        plog_info(plugin, "config key             : %s", placeholder);
     }
     if (config->keypass) {
-        plog_info(plugin, "config keypass        : %s", placeholder);
+        plog_info(plugin, "config keypass         : %s", placeholder);
     }
 
     return 0;
 
 error:
     free(err_param);
+    free(write_req_topic.v.val_str);
+    free(write_resp_topic.v.val_str);
     free(host.v.val_str);
     free(username.v.val_str);
     free(password.v.val_str);
@@ -297,6 +384,8 @@ error:
 void mqtt_config_fini(mqtt_config_t *config)
 {
     free(config->client_id);
+    free(config->write_req_topic);
+    free(config->write_resp_topic);
     free(config->host);
     free(config->username);
     free(config->password);

plugins/mqtt/mqtt_config.h

@@ -48,20 +48,22 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
 }
 
 typedef struct {
-    char *               client_id;       // client id
-    mqtt_upload_format_e format;          // upload format
-    neu_mqtt_qos_e       qos;             // message QoS
-    size_t               cache;           // cache enable flag
-    size_t               cache_mem_size;  // cache memory size in bytes
-    size_t               cache_disk_size; // cache disk size in bytes
-    char *               host;            // broker host
-    uint16_t             port;            // broker port
-    char *               username;        // user name
-    char *               password;        // user password
-    char *               ca;              // CA
-    char *               cert;            // client cert
-    char *               key;             // client key
-    char *               keypass;         // client key password
+    char *               client_id;        // client id
+    neu_mqtt_qos_e       qos;              // message QoS
+    mqtt_upload_format_e format;           // upload format
+    char *               write_req_topic;  // write request topic
+    char *               write_resp_topic; // write response topic
+    size_t               cache;            // cache enable flag
+    size_t               cache_mem_size;   // cache memory size in bytes
+    size_t               cache_disk_size;  // cache disk size in bytes
+    char *               host;             // broker host
+    uint16_t             port;             // broker port
+    char *               username;         // user name
+    char *               password;         // user password
+    char *               ca;               // CA
+    char *               cert;             // client cert
+    char *               key;              // client key
+    char *               keypass;          // client key password
 } mqtt_config_t;
 
 int  mqtt_config_parse(neu_plugin_t *plugin, const char *setting,

plugins/mqtt/mqtt_handle.c

@@ -373,7 +373,7 @@ int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
         goto end;
     }
 
-    char *         topic = plugin->write_resp_topic;
+    char *         topic = plugin->config.write_resp_topic;
     neu_mqtt_qos_e qos   = plugin->config.qos;
     rv       = publish(plugin, qos, topic, json_str, strlen(json_str));
     json_str = NULL;

plugins/mqtt/mqtt_plugin.c

@@ -18,6 +18,7 @@
  **/
 
 #include "utils/asprintf.h"
+#include "utils/time.h"
 
 #include "mqtt_config.h"
 #include "mqtt_handle.h"
@@ -81,10 +82,6 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin)
     plugin->read_req_topic = NULL;
     free(plugin->read_resp_topic);
     plugin->read_resp_topic = NULL;
-    free(plugin->write_req_topic);
-    plugin->write_req_topic = NULL;
-    free(plugin->write_resp_topic);
-    plugin->write_resp_topic = NULL;
 
     route_tbl_free(plugin->route_tbl);
 
@@ -159,6 +156,65 @@ static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client,
     return rv;
 }
 
+static int create_topic(neu_plugin_t *plugin)
+{
+    if (plugin->read_req_topic) {
+        // topics already created
+        return 0;
+    }
+
+    neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
+                 plugin->common.name);
+    if (NULL == plugin->read_req_topic) {
+        return -1;
+    }
+
+    neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
+                 plugin->common.name);
+    if (NULL == plugin->read_resp_topic) {
+        free(plugin->read_req_topic);
+        plugin->read_req_topic = NULL;
+        return -1;
+    }
+
+    return 0;
+}
+
+static int subscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
+{
+    if (0 != create_topic(plugin)) {
+        plog_error(plugin, "create topics fail");
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (0 !=
+        neu_mqtt_client_subscribe(plugin->client, config->qos,
+                                  plugin->read_req_topic, plugin,
+                                  handle_read_req)) {
+        plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic);
+        return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+    }
+
+    if (0 !=
+        neu_mqtt_client_subscribe(plugin->client, config->qos,
+                                  config->write_req_topic, plugin,
+                                  handle_write_req)) {
+        plog_error(plugin, "subscribe [%s] fail",
+                   plugin->config.write_req_topic);
+        return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+    }
+
+    return 0;
+}
+
+static int unsubscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
+{
+    neu_mqtt_client_unsubscribe(plugin->client, plugin->read_req_topic);
+    neu_mqtt_client_unsubscribe(plugin->client, config->write_req_topic);
+    neu_msleep(100); // wait for message completion
+    return 0;
+}
+
 static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
 {
     int           rv          = 0;
@@ -181,7 +237,8 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
         }
     } else if (neu_mqtt_client_is_open(plugin->client)) {
         started = true;
-        rv      = neu_mqtt_client_close(plugin->client);
+        unsubscribe(plugin, &plugin->config);
+        rv = neu_mqtt_client_close(plugin->client);
         if (0 != rv) {
             plog_error(plugin, "neu_mqtt_client_close fail");
             rv = NEU_ERR_EINTERNAL;
@@ -195,10 +252,15 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
         goto error;
     }
 
-    if (started && 0 != neu_mqtt_client_open(plugin->client)) {
-        plog_error(plugin, "neu_mqtt_client_open fail");
-        rv = NEU_ERR_MQTT_CONNECT_FAILURE;
-        goto error;
+    if (started) {
+        if (0 != neu_mqtt_client_open(plugin->client)) {
+            plog_error(plugin, "neu_mqtt_client_open fail");
+            rv = NEU_ERR_MQTT_CONNECT_FAILURE;
+            goto error;
+        }
+        if (0 != (rv = subscribe(plugin, &config))) {
+            goto error;
+        }
     }
 
     if (plugin->config.host) {
@@ -216,52 +278,6 @@ error:
     return rv;
 }
 
-static int create_topic(neu_plugin_t *plugin)
-{
-    if (plugin->read_req_topic) {
-        // topics already created
-        return 0;
-    }
-
-    neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
-                 plugin->common.name);
-    if (NULL == plugin->read_req_topic) {
-        return -1;
-    }
-
-    neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
-                 plugin->common.name);
-    if (NULL == plugin->read_resp_topic) {
-        free(plugin->read_req_topic);
-        plugin->read_req_topic = NULL;
-        return -1;
-    }
-
-    neu_asprintf(&plugin->write_req_topic, "/neuron/%s/write/req",
-                 plugin->common.name);
-    if (NULL == plugin->write_req_topic) {
-        free(plugin->read_req_topic);
-        plugin->read_req_topic = NULL;
-        free(plugin->read_resp_topic);
-        plugin->read_resp_topic = NULL;
-        return -1;
-    }
-
-    neu_asprintf(&plugin->write_resp_topic, "/neuron/%s/write/resp",
-                 plugin->common.name);
-    if (NULL == plugin->write_resp_topic) {
-        free(plugin->read_req_topic);
-        plugin->read_req_topic = NULL;
-        free(plugin->read_resp_topic);
-        plugin->read_resp_topic = NULL;
-        free(plugin->write_req_topic);
-        plugin->write_req_topic = NULL;
-        return -1;
-    }
-
-    return 0;
-}
-
 static int mqtt_plugin_start(neu_plugin_t *plugin)
 {
     int         rv          = 0;
@@ -279,29 +295,7 @@ static int mqtt_plugin_start(neu_plugin_t *plugin)
         goto end;
     }
 
-    if (0 != create_topic(plugin)) {
-        plog_error(plugin, "create topics fail");
-        rv = NEU_ERR_EINTERNAL;
-        goto end;
-    }
-
-    if (0 !=
-        neu_mqtt_client_subscribe(plugin->client, plugin->config.qos,
-                                  plugin->read_req_topic, plugin,
-                                  handle_read_req)) {
-        plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic);
-        rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
-        goto end;
-    }
-
-    if (0 !=
-        neu_mqtt_client_subscribe(plugin->client, plugin->config.qos,
-                                  plugin->write_req_topic, plugin,
-                                  handle_write_req)) {
-        plog_error(plugin, "subscribe [%s] fail", plugin->write_req_topic);
-        rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
-        goto end;
-    }
+    rv = subscribe(plugin, &plugin->config);
 
 end:
     if (0 == rv) {
@@ -317,6 +311,7 @@ end:
 static int mqtt_plugin_stop(neu_plugin_t *plugin)
 {
     if (plugin->client) {
+        unsubscribe(plugin, &plugin->config);
         neu_mqtt_client_close(plugin->client);
         plog_info(plugin, "mqtt client closed");
     }

plugins/mqtt/mqtt_plugin.h

@@ -49,8 +49,6 @@ struct neu_plugin {
     int64_t             cache_metric_update_ts;
     char *              read_req_topic;
     char *              read_resp_topic;
-    char *              write_req_topic;
-    char *              write_resp_topic;
     route_entry_t *     route_tbl;
 };
 

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/adapter.c

@@ -124,8 +124,8 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
 
     rv = nng_pair1_open(&adapter->sock);
     assert(rv == 0);
-    nng_socket_set_int(adapter->sock, NNG_OPT_RECVBUF, 128);
-    nng_socket_set_int(adapter->sock, NNG_OPT_SENDBUF, 128);
+    nng_socket_set_int(adapter->sock, NNG_OPT_RECVBUF, 8192);
+    nng_socket_set_int(adapter->sock, NNG_OPT_SENDBUF, 8192);
     nng_socket_set_ms(adapter->sock, NNG_OPT_SENDTIMEO, 1000);
     nng_socket_get_int(adapter->sock, NNG_OPT_RECVFD, &adapter->recv_fd);
 

src/adapter/driver/driver.c

@@ -19,6 +19,7 @@
 #include <assert.h>
 #include <math.h>
 #include <netinet/in.h>
+#include <pthread.h>
 #include <stdlib.h>
 
 #include <nng/nng.h>
@@ -37,12 +38,20 @@
 #include "errcodes.h"
 #include "tag.h"
 
+typedef struct to_be_write_tag {
+    neu_datatag_t *tag;
+    neu_value_u    value;
+    void *         req;
+} to_be_write_tag_t;
+
 typedef struct group {
     char *name;
 
-    int64_t      timestamp;
-    neu_group_t *group;
-    UT_array *   static_tags;
+    int64_t         timestamp;
+    neu_group_t *   group;
+    UT_array *      static_tags;
+    UT_array *      wt_tags;
+    pthread_mutex_t wt_mtx;
 
     neu_event_timer_t *report;
     neu_event_timer_t *read;
@@ -74,6 +83,7 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
                    neu_dvalue_t value);
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
 static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
+static void     store_write_tag(group_t *group, to_be_write_tag_t *tag);
 
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
 {
@@ -182,7 +192,13 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
         free(el->name);
         utarray_free(el->grp.tags);
 
+        utarray_foreach(el->wt_tags, to_be_write_tag_t *, tag)
+        {
+            neu_tag_free(tag->tag);
+        }
+
         utarray_free(el->static_tags);
+        utarray_free(el->wt_tags);
         neu_group_destroy(el->group);
         free(el);
     }
@@ -302,6 +318,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 +378,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);
@@ -409,8 +430,12 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             driver->adapter.cb_funs.response(&driver->adapter, req, &error);
             free(req);
         } else {
-            driver->adapter.module->intf_funs->driver.write_tag(
-                driver->adapter.plugin, (void *) req, tag, cmd->value.value);
+            to_be_write_tag_t wtag = { 0 };
+            wtag.req               = (void *) req;
+            wtag.value             = cmd->value.value;
+            wtag.tag               = neu_tag_dup(tag);
+
+            store_write_tag(g, &wtag);
         }
 
         neu_tag_free(tag);
@@ -424,6 +449,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
 int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
                                  uint32_t interval)
 {
+    UT_icd   icd  = { sizeof(to_be_write_tag_t), NULL, NULL, NULL };
     group_t *find = NULL;
     int      ret  = NEU_ERR_GROUP_EXIST;
 
@@ -438,6 +464,10 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
             .type        = NEU_EVENT_TIMER_NOBLOCK,
         };
 
+        pthread_mutex_init(&find->wt_mtx, NULL);
+
+        utarray_new(find->wt_tags, &icd);
+
         find->driver         = driver;
         find->name           = strdup(name);
         find->group          = neu_group_new(name, interval);
@@ -525,11 +555,18 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
             neu_driver_cache_del(driver->cache, name, tag->name);
         }
 
+        utarray_foreach(find->wt_tags, to_be_write_tag_t *, tag)
+        {
+            neu_tag_free(tag->tag);
+        }
+
         utarray_free(find->static_tags);
         utarray_free(find->grp.tags);
+        utarray_free(find->wt_tags);
         neu_group_destroy(find->group);
         free(find);
 
+        pthread_mutex_destroy(&find->wt_mtx);
         neu_adapter_del_group_metrics(&driver->adapter, name);
         ret = NEU_ERR_SUCCESS;
     }
@@ -887,6 +924,17 @@ static int read_callback(void *usr_data)
         return 0;
     }
 
+    pthread_mutex_lock(&group->wt_mtx);
+    utarray_foreach(group->wt_tags, to_be_write_tag_t *, wtag)
+    {
+        group->driver->adapter.module->intf_funs->driver.write_tag(
+            group->driver->adapter.plugin, (void *) wtag->req, wtag->tag,
+            wtag->value);
+        neu_tag_free(wtag->tag);
+    }
+    utarray_clear(group->wt_tags);
+    pthread_mutex_unlock(&group->wt_mtx);
+
     if (neu_group_is_change(group->group, group->timestamp)) {
         neu_group_change_test(group->group, group->timestamp, (void *) group,
                               group_change);
@@ -998,7 +1046,8 @@ static int read_report_group(int64_t timestamp, int64_t timeout,
             break;
         }
 
-        if ((timestamp - value.timestamp) > timeout) {
+        if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) &&
+            (timestamp - value.timestamp) > timeout) {
             datas[index].value.type      = NEU_TYPE_ERROR;
             datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED;
         } else {
@@ -1193,3 +1242,10 @@ static void read_group(int64_t timestamp, int64_t timeout,
         }
     }
 }
+
+static void store_write_tag(group_t *group, to_be_write_tag_t *tag)
+{
+    pthread_mutex_lock(&group->wt_mtx);
+    utarray_push_back(group->wt_tags, tag);
+    pthread_mutex_unlock(&group->wt_mtx);
+}

src/base/metrics.c

@@ -101,6 +101,11 @@ static inline size_t memory_used()
     return parse_memory_fields(3);
 }
 
+static inline size_t memory_cache()
+{
+    return parse_memory_fields(6);
+}
+
 static inline int disk_usage(size_t *size_p, size_t *used_p, size_t *avail_p)
 {
     struct statvfs buf = {};
@@ -276,6 +281,7 @@ void neu_metrics_visist(neu_metrics_cb_t cb, void *data)
 {
     unsigned cpu       = cpu_usage();
     size_t   mem_used  = memory_used();
+    size_t   mem_cache = memory_cache();
     size_t   disk_size = 0, disk_used = 0, disk_avail = 0;
     disk_usage(&disk_size, &disk_used, &disk_avail);
     bool     core_dumped    = has_core_dumps();
@@ -284,6 +290,7 @@ void neu_metrics_visist(neu_metrics_cb_t cb, void *data)
     g_metrics_.cpu_percent          = cpu;
     g_metrics_.cpu_cores            = get_nprocs();
     g_metrics_.mem_used_bytes       = mem_used;
+    g_metrics_.mem_cache_bytes      = mem_cache;
     g_metrics_.disk_size_gibibytes  = disk_size;
     g_metrics_.disk_used_gibibytes  = disk_used;
     g_metrics_.disk_avail_gibibytes = disk_avail;

src/connection/connection.c

@@ -33,6 +33,10 @@
 
 #include "connection/neu_connection.h"
 
+#ifndef CMSPAR
+#define CMSPAR 010000000000 /* mark or space (stick) parity */
+#endif
+
 struct tcp_client {
     int                fd;
     struct sockaddr_in client;
@@ -69,6 +73,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 +186,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 +195,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);
@@ -389,7 +401,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                    conn->fd, len, ret, strerror(errno), errno);
     }
 
-    if (errno == EPIPE || ret == -1 || ret == 0) {
+    if (errno == EPIPE || ret == -1 || (ret == 0 && errno != 0)) {
         conn_disconnect(conn);
     }
 
@@ -728,6 +740,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 +903,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

@@ -33,6 +33,7 @@
 #include "connection/mqtt_client.h"
 #include "event/event.h"
 #include "utils/asprintf.h"
+#include "utils/time.h"
 #include "utils/utextend.h"
 #include "utils/uthash.h"
 #include "utils/utlist.h"
@@ -68,6 +69,7 @@ typedef struct {
 typedef enum {
     TASK_PUB,
     TASK_SUB,
+    TASK_UNSUB,
     TASK_RECV,
 } task_kind_e;
 
@@ -133,7 +135,8 @@ static inline void    tasks_free(task_t *tasks);
 static void           task_cb(void *arg);
 static void           task_handle_pub(task_t *task, neu_mqtt_client_t *client);
 static void           task_handle_sub(task_t *task, neu_mqtt_client_t *client);
-static void           task_handle_recv(task_t *task, neu_mqtt_client_t *client);
+static void task_handle_unsub(task_t *task, neu_mqtt_client_t *client);
+static void task_handle_recv(task_t *task, neu_mqtt_client_t *client);
 
 static subscription_t *       subscription_new(neu_mqtt_client_t *client,
                                                neu_mqtt_qos_e qos, const char *topic,
@@ -150,6 +153,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,
@@ -220,6 +224,8 @@ static void task_cb(void *arg)
         task_handle_pub(task, client);
     } else if (TASK_SUB == task->kind) {
         task_handle_sub(task, client);
+    } else if (TASK_UNSUB == task->kind) {
+        task_handle_unsub(task, client);
     } else if (TASK_RECV == task->kind) {
         task_handle_recv(task, client);
     } else {
@@ -294,6 +300,41 @@ end:
     return;
 }
 
+static void task_handle_unsub(task_t *task, neu_mqtt_client_t *client)
+{
+    int      rv  = 0;
+    nng_msg *msg = NULL;
+    nng_aio *aio = task->aio;
+
+    if (0 != (rv = nng_aio_result(aio))) {
+        log(error, "send UNSUBSCRIBE error: %s", nng_strerror(rv));
+        goto end;
+    }
+
+    msg = nng_aio_get_msg(aio);
+    if (NULL == msg) {
+        rv = -1;
+        goto end;
+    }
+
+    nng_mqtt_packet_type pt = nng_mqtt_msg_get_packet_type(msg);
+    if (NNG_MQTT_UNSUBACK != pt) {
+        log(error, "recv packet type:%d, expect UNSUBACK packet", pt);
+        rv = -1;
+        goto end;
+    }
+
+end:
+    if (task->sub) {
+        log(info, "unsub [%s, QoS%d], %s", task->sub->topic, task->sub->qos,
+            (0 == rv) ? "OK" : "FAIL");
+    } else {
+        log(info, "unsub %s", (0 == rv) ? "OK" : "FAIL");
+    }
+    nng_msg_free(msg);
+    return;
+}
+
 static void task_handle_recv(task_t *task, neu_mqtt_client_t *client)
 {
     (void) client;
@@ -534,7 +575,7 @@ static inline void client_free_task(neu_mqtt_client_t *client, task_t *task)
 {
     nng_aio_set_msg(task->aio, NULL);
 
-    if (TASK_SUB == task->kind) {
+    if (TASK_SUB == task->kind || TASK_UNSUB == task->kind) {
         subscription_free(task->sub);
     } else if (TASK_RECV == task->kind) {
         subscription_free(task->recv.sub);
@@ -545,6 +586,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)
 {
@@ -558,6 +609,16 @@ static inline void client_add_subscription(neu_mqtt_client_t *client,
     HASH_ADD_STR(client->subscriptions, topic, sub);
 }
 
+static inline void client_del_subscription(neu_mqtt_client_t *client,
+                                           subscription_t *   sub)
+{
+    HASH_DEL(client->subscriptions, sub);
+    if (sub->ack) {
+        client->suback_count -= 1;
+    }
+    subscription_free(sub);
+}
+
 static int client_send_sub_msg(neu_mqtt_client_t *client,
                                subscription_t *   subscription)
 {
@@ -595,6 +656,38 @@ static int client_send_sub_msg(neu_mqtt_client_t *client,
     return 0;
 }
 
+static int client_send_unsub_msg(neu_mqtt_client_t *client,
+                                 subscription_t *   subscription)
+{
+    int      rv      = 0;
+    nng_msg *sub_msg = NULL;
+
+    if ((rv = nng_mqtt_msg_alloc(&sub_msg, 0)) != 0) {
+        log(error, "nng_mqtt_msg_alloc fail: %s", nng_strerror(rv));
+        return -1;
+    }
+
+    nng_mqtt_topic mqtt_topic = { .buf    = (uint8_t *) subscription->topic,
+                                  .length = strlen(subscription->topic) };
+
+    nng_mqtt_msg_set_packet_type(sub_msg, NNG_MQTT_UNSUBSCRIBE);
+    nng_mqtt_msg_set_unsubscribe_topics(sub_msg, &mqtt_topic, 1);
+
+    task_t *task = client_alloc_task(client);
+    if (NULL == task) {
+        log(error, "client_alloc_task fail");
+        nng_msg_free(sub_msg);
+        return -1;
+    }
+
+    task->kind = TASK_UNSUB;
+    task->sub  = subscription ? subscription_ref(subscription) : NULL;
+    nng_aio_set_msg(task->aio, sub_msg);
+    nng_send_aio(client->sock, task->aio);
+
+    return 0;
+}
+
 static inline void client_start_recv(neu_mqtt_client_t *client)
 {
     if (client->recv_aio && !client->receiving) {
@@ -1149,6 +1242,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);
+        neu_msleep(100);
+        nng_mtx_lock(client->mtx);
+    }
     client->open = false;
     nng_mtx_unlock(client->mtx);
 
@@ -1244,3 +1343,43 @@ error:
     subscription_free(subscription);
     return -1;
 }
+
+int neu_mqtt_client_unsubscribe(neu_mqtt_client_t *client, const char *topic)
+{
+    int             rv           = 0;
+    subscription_t *subscription = NULL;
+
+    nng_mtx_lock(client->mtx);
+    if (NULL == client->recv_aio) {
+        if ((rv = nng_aio_alloc(&client->recv_aio, recv_cb, client)) != 0) {
+            log(error, "nng_aio_alloc fail: %s", nng_strerror(rv));
+            goto error;
+        }
+        if (client->connected) {
+            // connect_cb already fired, then start receiving
+            client_start_recv(client);
+        }
+    }
+
+    HASH_FIND_STR(client->subscriptions, topic, subscription);
+
+    if (client->connected) {
+        // send a unsubscribe message even subscription is NULL
+        if (client_send_unsub_msg(client, subscription) < 0) {
+            log(error, "client_send_unsub_msg fail");
+            goto error;
+        }
+    }
+
+    if (subscription) {
+        client_del_subscription(client, subscription);
+    }
+
+    nng_mtx_unlock(client->mtx);
+
+    return 0;
+
+error:
+    nng_mtx_unlock(client->mtx);
+    return -1;
+}

src/core/manager.c

@@ -97,8 +97,9 @@ neu_manager_t *neu_manager_create()
     rv = nng_listen(manager->socket, url, NULL, 0);
     assert(rv == 0);
 
-    nng_socket_set_int(manager->socket, NNG_OPT_RECVBUF, 2048);
-    nng_socket_set_int(manager->socket, NNG_OPT_SENDBUF, 2048);
+    nng_socket_set_int(manager->socket, NNG_OPT_RECVBUF, 8192);
+    nng_socket_set_int(manager->socket, NNG_OPT_SENDBUF, 8292);
+    nng_socket_set_ms(manager->socket, NNG_OPT_SENDTIMEO, 1000);
     nng_socket_get_int(manager->socket, NNG_OPT_RECVFD, &param.fd);
     manager->loop = neu_event_add_io(manager->events, param);
 
@@ -482,6 +483,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             forward_msg(manager, msg, cmd->app);
             manager_storage_subscribe(manager, cmd->app, cmd->driver,
                                       cmd->group, cmd->params);
+        } else {
+            free(cmd->params);
         }
 
         header->type = NEU_RESP_ERROR;
@@ -729,6 +732,7 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg,
     if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
         nlog_debug("forward msg to pipe %d", pipe.id);
     } else {
+        nlog_warn("forward msg to pipe %d fail", pipe.id);
         nng_msg_free(out_msg);
     }
 }
@@ -744,6 +748,7 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg,
     if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
         nlog_debug("forward msg to %s", node);
     } else {
+        nlog_warn("forward msg to pipe (%s)%d fail", node, pipe.id);
         nng_msg_free(out_msg);
     }
 }

src/core/plugin_manager.c

@@ -50,11 +50,6 @@ typedef struct plugin_entity {
     UT_hash_handle hh;
 } plugin_entity_t;
 
-typedef struct {
-    char *name_old;
-    char *name_new;
-} plugin_name_t;
-
 struct neu_plugin_manager {
     plugin_entity_t *plugins;
 };
@@ -239,41 +234,6 @@ int neu_plugin_manager_find(neu_plugin_manager_t *mgr, const char *plugin_name,
     plugin_entity_t *plugin = NULL;
     int              ret    = -1;
 
-    static const plugin_name_t name[] = {
-        { "mqtt", "MQTT" },
-        { "ekuiper", "eKuiper" },
-        { "modbus-tcp", "Modbus TCP community" },
-        { "a1e", "Mitsubishi 1E" },
-        { "ads", "Beckhoff ADS" },
-        { "bacnet", "BACnet/IP" },
-        { "dlt645", "DLT645-2007" },
-        { "EtherNet-IP", "EtherNet/IP(CIP)" },
-        { "fins", "Omron FINS TCP" },
-        { "focas", "Fanuc Focas Ethernet" },
-        { "iec104", "IEC60870-5-104" },
-        { "knx", "KNXnet/IP" },
-        { "modbus-qh-tcp", "Modbus TCP QH" },
-        { "modbus-rtu", "Modbus RTU" },
-        { "modbus-plus-tcp", "Modbus TCP" },
-        { "nona11", "NON A11" },
-        { "opcua", "OPC UA" },
-        { "qna3e", "Mitsubishi 3E" },
-        { "s7comm", "Siemens S7 ISOTCP" },
-        { "s7comm_for_300", "Siemens S7 ISOTCP for 300/400" },
-        { "sparkplugb", "SparkPlugB" },
-        { NULL, NULL },
-    };
-
-    int i = 0;
-    while (name[i].name_old != NULL) {
-        if (strcmp(plugin_name, name[i].name_old) == 0) {
-            plugin_name = name[i].name_new;
-            break;
-        }
-
-        i++;
-    }
-
     HASH_FIND_STR(mgr->plugins, plugin_name, plugin);
     if (plugin != NULL) {
         ret           = 0;

src/parser/neu_json_global_config.c

@@ -105,9 +105,6 @@ int neu_json_decode_global_config_req_tags(
     }
 
     req->n_tag = json_array_size(tag_array);
-    if (req->n_tag <= 0) {
-        goto decode_fail;
-    }
 
     req->tags = calloc(req->n_tag, sizeof(*req->tags));
     if (NULL == req->tags) {

src/parser/neu_json_group_config.c

@@ -311,9 +311,6 @@ int neu_json_decode_get_driver_group_resp_json(
     }
 
     resp->n_group = json_array_size(grp_array);
-    if (resp->n_group <= 0) {
-        goto decode_fail;
-    }
 
     resp->groups = calloc(resp->n_group, sizeof(*resp->groups));
     if (NULL == resp->groups) {

version

@@ -1 +1 @@
-2.4.0
+2.4.6

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