Merge pull request #2259 from fengzeroz/v2.9

version: 2.9.4

.github/workflows/check.yml

@@ -18,7 +18,7 @@ jobs:
           submodules: recursive
 
       - name: Run clang-format style check for C/C++ programs.
-        uses: DoozyX/clang-format-lint-action@v0.12
+        uses: DoozyX/clang-format-lint-action@v0.18
         with:
           source: 'src plugins include tests'
           exclude: 'include/neuron/utils/uthash.h include/neuron/utils/utarray.h include/neuron/utils/utlist.h include/neuron/utils/zlog.h'
@@ -74,9 +74,7 @@ jobs:
 
       - name: setup dependencies
         run: |
-          if [ "${{ matrix.plugin }}" = "mqtt" ] || [ "${{ matrix.plugin }}" = "azure" ]; then
-            sudo apt-get install -y mosquitto
-          fi
+          sudo apt-get install -y mosquitto
           pip install -U pytest requests prometheus-client gmqtt pynng sniffio
           sudo apt-get install -y socat
 

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.9.0
+  VERSION: 2.9.4
   UI_VERSION: 2.6.1
   BRANCH: v2.9 
 

include/neuron/connection/mqtt_client.h

@@ -85,25 +85,26 @@ bool   neu_mqtt_client_is_open(neu_mqtt_client_t *client);
 bool   neu_mqtt_client_is_connected(neu_mqtt_client_t *client);
 size_t neu_mqtt_client_get_cached_msgs_num(neu_mqtt_client_t *client);
 
-int neu_mqtt_client_set_addr(neu_mqtt_client_t *client, const char *host,
-                             uint16_t port);
-int neu_mqtt_client_set_id(neu_mqtt_client_t *client, const char *id);
-int neu_mqtt_client_set_will_msg(neu_mqtt_client_t *client, const char *topic,
-                                 uint8_t *msg, uint32_t len, bool retain,
-                                 uint8_t qos);
-int neu_mqtt_client_set_user(neu_mqtt_client_t *client, const char *username,
-                             const char *password);
-int neu_mqtt_client_set_connect_cb(neu_mqtt_client_t *             client,
-                                   neu_mqtt_client_connection_cb_t cb,
-                                   void *                          data);
-int neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t *             client,
-                                      neu_mqtt_client_connection_cb_t cb,
-                                      void *                          data);
-int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled,
-                            const char *ca, const char *cert, const char *key,
-                            const char *keypass);
-int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
-                                   size_t mem_size_bytes, size_t db_size_bytes);
+int  neu_mqtt_client_set_addr(neu_mqtt_client_t *client, const char *host,
+                              uint16_t port);
+int  neu_mqtt_client_set_id(neu_mqtt_client_t *client, const char *id);
+int  neu_mqtt_client_set_will_msg(neu_mqtt_client_t *client, const char *topic,
+                                  uint8_t *msg, uint32_t len, bool retain,
+                                  uint8_t qos);
+int  neu_mqtt_client_set_user(neu_mqtt_client_t *client, const char *username,
+                              const char *password);
+int  neu_mqtt_client_set_connect_cb(neu_mqtt_client_t *             client,
+                                    neu_mqtt_client_connection_cb_t cb,
+                                    void *                          data);
+int  neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t *             client,
+                                       neu_mqtt_client_connection_cb_t cb,
+                                       void *                          data);
+int  neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled,
+                             const char *ca, const char *cert, const char *key,
+                             const char *keypass);
+void neu_mqtt_client_remove_cache_db(neu_mqtt_client_t *client);
+int  neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
+                                    size_t mem_size_bytes, size_t db_size_bytes);
 // default to NEU_MQTT_CACHE_SYNC_INTERVAL_DEFAULT if not set
 int neu_mqtt_client_set_cache_sync_interval(neu_mqtt_client_t *client,
                                             uint32_t           interval);

include/neuron/define.h

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 9
-#define NEU_VERSION_FIX 0
+#define NEU_VERSION_FIX 3
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)
@@ -94,6 +94,7 @@
 extern int  default_log_level;
 extern bool disable_jwt;
 extern char host_port[32];
+extern char g_status[32];
 
 typedef enum neu_plugin_kind {
     NEU_PLUGIN_KIND_STATIC = 0,

include/neuron/errcodes.h

@@ -47,6 +47,7 @@ typedef enum {
     NEU_ERR_COMMAND_EXECUTION_FAILED = 1014,
     NEU_ERR_IP_ADDRESS_INVALID       = 1015,
     NEU_ERR_IP_ADDRESS_IN_USE        = 1016,
+    NEU_ERR_BODY_TOO_BIG             = 1017,
 
     NEU_ERR_NODE_EXIST               = 2002,
     NEU_ERR_NODE_NOT_EXIST           = 2003,

include/neuron/metrics.h

@@ -248,6 +248,7 @@ typedef struct {
     const char *           name;  // NOTE: should points to string literal
     const char *           help;  // NOTE: should points to string literal
     neu_metric_type_e      type;  //
+    uint64_t               init;  //
     uint64_t               value; //
     neu_rolling_counter_t *rcnt;  //
     UT_hash_handle         hh;    // ordered by name
@@ -489,7 +490,7 @@ static inline void neu_node_metrics_reset(neu_node_metrics_t *node_metrics)
     HASH_LOOP(hh, node_metrics->entries, entry)
     {
         if (!neu_metric_type_no_reset(entry->type)) {
-            entry->value = 0;
+            entry->value = entry->init;
             if (neu_metric_type_is_rolling_counter(entry->type)) {
                 neu_rolling_counter_reset(entry->rcnt);
             }
@@ -502,7 +503,7 @@ static inline void neu_node_metrics_reset(neu_node_metrics_t *node_metrics)
         HASH_LOOP(hh, g->entries, entry)
         {
             if (!neu_metric_type_no_reset(entry->type)) {
-                entry->value = 0;
+                entry->value = entry->init;
                 if (neu_metric_type_is_rolling_counter(entry->type)) {
                     neu_rolling_counter_reset(entry->rcnt);
                 }

include/neuron/msg.h

@@ -750,6 +750,7 @@ typedef struct {
     int   current_page;
     int   page_size;
     bool  is_error;
+    int   total_count; // tags count without pagination
 
     UT_array *tags; // neu_resp_tag_value_meta_paginate_t
 } neu_resp_read_group_paginate_t;

include/neuron/type.h

@@ -474,6 +474,238 @@ static inline void neu_value16_set_bit(neu_value16_u *value, uint8_t index,
     }
 }
 
+static inline uint16_t neu_get_u16(uint8_t *bytes)
+{
+    uint16_t ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+
+    return ret;
+}
+
+static inline void neu_set_u16(uint8_t *bytes, uint16_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+}
+
+static inline int16_t neu_get_i16(uint8_t *bytes)
+{
+    int16_t  ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+
+    return ret;
+}
+
+static inline void neu_set_i16(uint8_t *bytes, int16_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+}
+
+static inline uint32_t neu_get_u32(uint8_t *bytes)
+{
+    uint32_t ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+
+    return ret;
+}
+
+static inline void neu_set_u32(uint8_t *bytes, uint32_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+}
+
+static inline int32_t neu_get_i32(uint8_t *bytes)
+{
+    int32_t  ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+
+    return ret;
+}
+
+static inline void neu_set_i32(uint8_t *bytes, int32_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+}
+
+static inline float neu_get_f32(uint8_t *bytes)
+{
+    float    ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+
+    return ret;
+}
+
+static inline void neu_set_f32(uint8_t *bytes, float val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+}
+
+static inline uint64_t neu_get_u64(uint8_t *bytes)
+{
+    uint64_t ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+    t[4] = bytes[4];
+    t[5] = bytes[5];
+    t[6] = bytes[6];
+    t[7] = bytes[7];
+
+    return ret;
+}
+
+static inline void neu_set_u64(uint8_t *bytes, uint64_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+    bytes[4] = t[4];
+    bytes[5] = t[5];
+    bytes[6] = t[6];
+    bytes[7] = t[7];
+}
+
+static inline int64_t neu_get_i64(uint8_t *bytes)
+{
+    int64_t  ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+    t[4] = bytes[4];
+    t[5] = bytes[5];
+    t[6] = bytes[6];
+    t[7] = bytes[7];
+
+    return ret;
+}
+
+static inline void neu_set_i64(uint8_t *bytes, int64_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+    bytes[4] = t[4];
+    bytes[5] = t[5];
+    bytes[6] = t[6];
+    bytes[7] = t[7];
+}
+
+static inline double neu_get_f64(uint8_t *bytes)
+{
+    double   ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+    t[4] = bytes[4];
+    t[5] = bytes[5];
+    t[6] = bytes[6];
+    t[7] = bytes[7];
+
+    return ret;
+}
+
+static inline void neu_set_f64(uint8_t *bytes, double val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+    bytes[4] = t[4];
+    bytes[5] = t[5];
+    bytes[6] = t[6];
+    bytes[7] = t[7];
+}
+
 static inline void neu_ntohs24_p(uint8_t *value)
 {
     uint8_t t = 0;
@@ -513,7 +745,8 @@ static inline uint64_t neu_ntohll(uint64_t u)
 
 static inline void neu_ntonll_p(uint64_t *pu)
 {
-    *pu = neu_htonll(*pu);
+    uint64_t ret = neu_htonll(neu_get_u64((uint8_t *) pu));
+    neu_set_u64((uint8_t *) pu, ret);
 }
 
 static inline void neu_htonll_p(uint64_t *pu)
@@ -523,7 +756,8 @@ static inline void neu_htonll_p(uint64_t *pu)
 
 static inline void neu_ntohl_p(uint32_t *pu)
 {
-    *pu = ntohl(*pu);
+    uint32_t ret = ntohl(neu_get_u32((uint8_t *) pu));
+    neu_set_u32((uint8_t *) pu, ret);
 }
 
 static inline void neu_htonl_p(uint32_t *pu)
@@ -533,7 +767,8 @@ static inline void neu_htonl_p(uint32_t *pu)
 
 static inline void neu_ntohs_p(uint16_t *pu)
 {
-    *pu = ntohs(*pu);
+    uint16_t ret = ntohs(neu_get_u16((uint8_t *) pu));
+    neu_set_u16((uint8_t *) pu, ret);
 }
 
 static inline void neu_htons_p(uint16_t *pu)
@@ -541,134 +776,6 @@ static inline void neu_htons_p(uint16_t *pu)
     neu_ntohs_p(pu);
 }
 
-static inline uint16_t neu_get_u16(uint8_t *bytes)
-{
-    uint16_t ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-
-    return ret;
-}
-
-static inline int16_t neu_get_i16(uint8_t *bytes)
-{
-    int16_t  ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-
-    return ret;
-}
-
-static inline uint32_t neu_get_u32(uint8_t *bytes)
-{
-    uint32_t ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-
-    return ret;
-}
-
-static inline int32_t neu_get_i32(uint8_t *bytes)
-{
-    int32_t  ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-
-    return ret;
-}
-
-static inline float neu_get_f32(uint8_t *bytes)
-{
-    float    ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-
-    return ret;
-}
-
-static inline uint64_t neu_get_u64(uint8_t *bytes)
-{
-    uint64_t ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-    t[4] = bytes[4];
-    t[5] = bytes[5];
-    t[6] = bytes[6];
-    t[7] = bytes[7];
-
-    return ret;
-}
-
-static inline int64_t neu_get_i64(uint8_t *bytes)
-{
-    int64_t  ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-    t[4] = bytes[4];
-    t[5] = bytes[5];
-    t[6] = bytes[6];
-    t[7] = bytes[7];
-
-    return ret;
-}
-
-static inline double neu_get_f64(uint8_t *bytes)
-{
-    double   ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-    t[4] = bytes[4];
-    t[5] = bytes[5];
-    t[6] = bytes[6];
-    t[7] = bytes[7];
-
-    return ret;
-}
-
 #ifdef __cplusplus
 }
 #endif

neuron.json

@@ -3,5 +3,6 @@
 	"port": 7000,
 	"disable_auth": 0,
 	"syslog_host": "",
-	"syslog_port": 541
+	"syslog_port": 541,
+	"sub_filter_error": 0
 }

plugins/modbus/modbus-tcp.json

@@ -67,6 +67,25 @@
 			]
 		}
 	},
+	"check_header": {
+		"name": "Check Header",
+		"name_zh": "校验报文头",
+		"attribute": "required",
+		"type": "map",
+		"default": 0,
+		"valid": {
+			"map": [
+				{
+					"key": "False",
+					"value": 0
+				},
+				{
+					"key": "True",
+					"value": 1
+				}
+			]
+		}
+	},
 	"max_retries": {
 		"name": "Maximum Retry Times",
 		"name_zh": "最大重试次数",

plugins/modbus/modbus_req.c

@@ -206,6 +206,8 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
                                     "modbus message error, skip, %hhu!%hu",
                                     gd->cmd_sort->cmd[i].slave_id,
                                     gd->cmd_sort->cmd[i].start_address);
+                                rtt = NEU_METRIC_LAST_RTT_MS_MAX;
+                                neu_conn_disconnect(plugin->conn);
                             } else if (ret_buf == -2) {
                                 modbus_value_handle(
                                     plugin, gd->cmd_sort->cmd[i].slave_id, 0,
@@ -216,8 +218,8 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
                                     "%hhu!%hu",
                                     gd->cmd_sort->cmd[i].slave_id,
                                     gd->cmd_sort->cmd[i].start_address);
+                                rtt = neu_time_ms() - read_tms;
                             }
-                            rtt = neu_time_ms() - read_tms;
                             break;
                         }
                     } else {
@@ -246,6 +248,8 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
                     plog_error(plugin, "modbus message error, skip, %hhu!%hu",
                                gd->cmd_sort->cmd[i].slave_id,
                                gd->cmd_sort->cmd[i].start_address);
+                    rtt = NEU_METRIC_LAST_RTT_MS_MAX;
+                    neu_conn_disconnect(plugin->conn);
                 } else if (ret_buf == -2) {
                     modbus_value_handle(plugin, gd->cmd_sort->cmd[i].slave_id,
                                         0, NULL, NEU_ERR_PLUGIN_READ_FAILURE);
@@ -253,8 +257,8 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
                                "modbus device response error, skip, %hhu!%hu",
                                gd->cmd_sort->cmd[i].slave_id,
                                gd->cmd_sort->cmd[i].start_address);
+                    rtt = neu_time_ms() - read_tms;
                 }
-                rtt = neu_time_ms() - read_tms;
             }
         } else {
             for (uint16_t j = 0; j < plugin->max_retries; j++) {
@@ -275,6 +279,8 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
                                        "modbus message error, skip, %hhu!%hu",
                                        gd->cmd_sort->cmd[i].slave_id,
                                        gd->cmd_sort->cmd[i].start_address);
+                            rtt = NEU_METRIC_LAST_RTT_MS_MAX;
+                            neu_conn_disconnect(plugin->conn);
                         } else if (ret_buf == -2) {
                             modbus_value_handle(
                                 plugin, gd->cmd_sort->cmd[i].slave_id, 0, NULL,
@@ -284,8 +290,8 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
                                        "%hhu!%hu",
                                        gd->cmd_sort->cmd[i].slave_id,
                                        gd->cmd_sort->cmd[i].start_address);
+                            rtt = neu_time_ms() - read_tms;
                         }
-                        rtt = neu_time_ms() - read_tms;
                         break;
                     } else if (ret_buf == 0) {
                         modbus_value_handle(

plugins/modbus/modbus_req.h

@@ -43,6 +43,7 @@ struct neu_plugin {
     uint16_t interval;
     uint16_t retry_interval;
     uint16_t max_retries;
+    uint16_t check_header;
 };
 
 void modbus_conn_connected(void *data, int fd);

plugins/modbus/modbus_stack.c

@@ -77,6 +77,12 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
         if (ret <= 0) {
             return -1;
         }
+
+        neu_plugin_t *plugin = (neu_plugin_t *) stack->ctx;
+        if (plugin->check_header && header.seq + 1 != stack->read_seq &&
+            header.seq + 1 != stack->write_seq) {
+            return -1;
+        }
     }
 
     ret = modbus_code_unwrap(buf, &code);

plugins/modbus/modbus_tcp.c

@@ -161,6 +161,8 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
     neu_json_elem_t  max_retries = { .name = "max_retries", .t = NEU_JSON_INT };
     neu_json_elem_t  retry_interval = { .name = "retry_interval",
                                        .t    = NEU_JSON_INT };
+    neu_json_elem_t  check_header   = { .name = "check_header",
+                                     .t    = NEU_JSON_INT };
 
     ret = neu_parse_param((char *) config, &err_param, 5, &port, &host, &mode,
                           &timeout, &interval);
@@ -189,10 +191,17 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
         retry_interval.v.val_int = 0;
     }
 
+    ret = neu_parse_param((char *) config, &err_param, 1, &check_header);
+    if (ret != 0) {
+        free(err_param);
+        check_header.v.val_int = 0;
+    }
+
     param.log              = plugin->common.log;
     plugin->interval       = interval.v.val_int;
     plugin->max_retries    = max_retries.v.val_int;
     plugin->retry_interval = retry_interval.v.val_int;
+    plugin->check_header   = check_header.v.val_int;
 
     if (mode.v.val_int == 1) {
         param.type                           = NEU_CONN_TCP_SERVER;

plugins/mqtt/mqtt_plugin_intf.c

@@ -402,9 +402,17 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     case NEU_REQ_UPDATE_NODE:
         error = handle_update_driver(plugin, data);
         break;
-    case NEU_REQRESP_NODE_DELETED:
-        error = handle_del_driver(plugin, data);
+    case NEU_REQRESP_NODE_DELETED: {
+        neu_reqresp_node_deleted_t *req = data;
+        if (0 != strcmp(plugin->common.name, req->node)) {
+            error = handle_del_driver(plugin, data);
+        } else {
+            if (plugin->client) {
+                neu_mqtt_client_remove_cache_db(plugin->client);
+            }
+        }
         break;
+    }
     default:
         error = NEU_ERR_MQTT_FAILURE;
         break;

plugins/restful/handle.c

@@ -407,6 +407,12 @@ static struct neu_http_handler rest_handlers[] = {
         .url           = "/api/v2/scan/tags",
         .value.handler = handle_scan_tags,
     },
+    {
+        .method        = NEU_HTTP_METHOD_GET,
+        .type          = NEU_HTTP_HANDLER_FUNCTION,
+        .url           = "/api/v2/status",
+        .value.handler = handle_status,
+    },
 };
 
 void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size)

plugins/restful/normal_handle.c

@@ -328,3 +328,22 @@ void handle_get_plugin_schema(nng_aio *aio)
     free(buf);
     free(schema_path);
 }
+
+void handle_status(nng_aio *aio)
+{
+    char *result = NULL;
+
+    NEU_VALIDATE_JWT(aio);
+
+    int ret = neu_asprintf(&result, "{\"status\":\"%s\"}", g_status);
+
+    if (ret < 0) {
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
+            neu_http_response(aio, error_code.error, result_error);
+        });
+        return;
+    }
+
+    neu_http_response(aio, 0, result);
+    free(result);
+}

plugins/restful/normal_handle.h

@@ -26,5 +26,6 @@ void handle_ping(nng_aio *aio);
 void handle_login(nng_aio *aio);
 void handle_password(nng_aio *aio);
 void handle_get_plugin_schema(nng_aio *aio);
+void handle_status(nng_aio *aio);
 
 #endif

plugins/restful/rw_handle.c

@@ -499,8 +499,9 @@ void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp)
 void handle_read_paginate_resp(nng_aio *                       aio,
                                neu_resp_read_group_paginate_t *resp)
 {
-    neu_json_read_paginate_resp_t api_res = { 0 };
-    char *                        result  = NULL;
+    neu_json_read_paginate_resp_t api_res    = { 0 };
+    char *                        result     = NULL;
+    unsigned int                  total_tags = resp->total_count;
 
     if (resp->is_error) {
         UT_array *filtered_tags;
@@ -538,51 +539,79 @@ void handle_read_paginate_resp(nng_aio *                       aio,
 
         utarray_free(resp->tags);
         resp->tags = filtered_tags;
-    }
+        total_tags = utarray_len(resp->tags);
 
-    unsigned int total_tags = utarray_len(resp->tags);
+        if (resp->current_page > 0 && resp->page_size > 0) {
+            unsigned int start_index =
+                (resp->current_page - 1) * resp->page_size;
+            unsigned int end_index = start_index + resp->page_size;
+
+            if (start_index >= total_tags) {
+                api_res.n_tag = 0;
+                api_res.tags =
+                    calloc(1, sizeof(neu_json_read_paginate_resp_tag_t));
+            } else {
+                if (end_index > total_tags) {
+                    end_index = total_tags;
+                }
+
+                api_res.n_tag = end_index - start_index;
+                api_res.tags  = calloc(
+                    api_res.n_tag, sizeof(neu_json_read_paginate_resp_tag_t));
+
+                for (unsigned int i = start_index; i < end_index; i++) {
+                    neu_resp_tag_value_meta_paginate_t *tag_value =
+                        (neu_resp_tag_value_meta_paginate_t *) utarray_eltptr(
+                            resp->tags, i);
+                    neu_tag_value_to_json_paginate(
+                        tag_value, &api_res.tags[i - start_index]);
+                }
+            }
+        } else {
+            api_res.n_tag = total_tags;
+            api_res.tags  = calloc(api_res.n_tag,
+                                  sizeof(neu_json_read_paginate_resp_tag_t));
+            int index     = 0;
+
+            utarray_foreach(resp->tags, neu_resp_tag_value_meta_paginate_t *,
+                            tag_value)
+            {
+                neu_tag_value_to_json_paginate(tag_value,
+                                               &api_res.tags[index++]);
+            }
+        }
+    } else {
+
+        if (resp->current_page > 0 && resp->page_size > 0) {
+            api_res.n_tag = utarray_len(resp->tags);
+            api_res.tags  = calloc(api_res.n_tag,
+                                  sizeof(neu_json_read_paginate_resp_tag_t));
+            int index     = 0;
+            utarray_foreach(resp->tags, neu_resp_tag_value_meta_paginate_t *,
+                            tag_value)
+            {
+                neu_tag_value_to_json_paginate(tag_value,
+                                               &api_res.tags[index++]);
+            }
+        } else {
+            api_res.n_tag = total_tags;
+            api_res.tags  = calloc(api_res.n_tag,
+                                  sizeof(neu_json_read_paginate_resp_tag_t));
+            int index     = 0;
+
+            utarray_foreach(resp->tags, neu_resp_tag_value_meta_paginate_t *,
+                            tag_value)
+            {
+                neu_tag_value_to_json_paginate(tag_value,
+                                               &api_res.tags[index++]);
+            }
+        }
+    }
 
     api_res.meta.current_page = resp->current_page;
     api_res.meta.page_size    = resp->page_size;
     api_res.meta.total        = total_tags;
 
-    if (resp->current_page > 0 && resp->page_size > 0) {
-        unsigned int start_index = (resp->current_page - 1) * resp->page_size;
-        unsigned int end_index   = start_index + resp->page_size;
-
-        if (start_index >= total_tags) {
-            api_res.n_tag = 0;
-            api_res.tags = calloc(1, sizeof(neu_json_read_paginate_resp_tag_t));
-        } else {
-            if (end_index > total_tags) {
-                end_index = total_tags;
-            }
-
-            api_res.n_tag = end_index - start_index;
-            api_res.tags  = calloc(api_res.n_tag,
-                                  sizeof(neu_json_read_paginate_resp_tag_t));
-
-            for (unsigned int i = start_index; i < end_index; i++) {
-                neu_resp_tag_value_meta_paginate_t *tag_value =
-                    (neu_resp_tag_value_meta_paginate_t *) utarray_eltptr(
-                        resp->tags, i);
-                neu_tag_value_to_json_paginate(tag_value,
-                                               &api_res.tags[i - start_index]);
-            }
-        }
-    } else {
-        api_res.n_tag = total_tags;
-        api_res.tags =
-            calloc(api_res.n_tag, sizeof(neu_json_read_paginate_resp_tag_t));
-        int index = 0;
-
-        utarray_foreach(resp->tags, neu_resp_tag_value_meta_paginate_t *,
-                        tag_value)
-        {
-            neu_tag_value_to_json_paginate(tag_value, &api_res.tags[index++]);
-        }
-    }
-
     neu_json_encode_by_fn(&api_res, neu_json_encode_read_paginate_resp,
                           &result);
 

src/adapter/adapter.c

@@ -816,7 +816,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_NODE_SETTING: {
         neu_req_node_setting_t *cmd   = (neu_req_node_setting_t *) &header[1];
         neu_resp_error_t        error = { 0 };
-
+        nlog_notice("setting node:%s params:%s", cmd->node, cmd->setting);
         error.error = neu_adapter_set_setting(adapter, cmd->setting);
         if (error.error == NEU_ERR_SUCCESS) {
             adapter_storage_setting(adapter->name, cmd->setting);
@@ -909,7 +909,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_ADD_GROUP: {
         neu_req_add_group_t *cmd   = (neu_req_add_group_t *) &header[1];
         neu_resp_error_t     error = { 0 };
-
+        nlog_notice("add group node:%s group:%s interval:%d", cmd->driver,
+                    cmd->group, cmd->interval);
         if (cmd->interval < NEU_GROUP_INTERVAL_LIMIT) {
             error.error = NEU_ERR_GROUP_PARAMETER_INVALID;
         } else {
@@ -934,7 +935,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_UPDATE_DRIVER_GROUP: {
         neu_req_update_group_t *cmd  = (neu_req_update_group_t *) &header[1];
         neu_resp_update_group_t resp = { 0 };
-
+        nlog_notice("update group node:%s old_name:%s new_name:%s interval:%d",
+                    header->receiver, cmd->group, cmd->new_name, cmd->interval);
         if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
             resp.error = neu_adapter_driver_update_group(
                 (neu_adapter_driver_t *) adapter, cmd->group, cmd->new_name,
@@ -959,7 +961,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_DEL_GROUP: {
         neu_req_del_group_t *cmd   = (neu_req_del_group_t *) &header[1];
         neu_resp_error_t     error = { 0 };
-
+        nlog_notice("del group node:%s group:%s", cmd->driver, cmd->group);
         if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
             error.error = neu_adapter_driver_del_group(
                 (neu_adapter_driver_t *) adapter, cmd->group);
@@ -1015,6 +1017,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
             for (int i = 0; i < cmd->n_tag; i++) {
+                nlog_notice("del tag node:%s group:%s tag:%s", cmd->driver,
+                            cmd->group, cmd->tags[i]);
                 int ret = neu_adapter_driver_del_tag(
                     (neu_adapter_driver_t *) adapter, cmd->group, cmd->tags[i]);
                 if (0 == ret) {

src/adapter/driver/cache.c

@@ -29,6 +29,8 @@
 
 #include "cache.h"
 
+extern bool sub_filter_err;
+
 typedef struct {
     char group[NEU_GROUP_NAME_LEN];
     char tag[NEU_TAG_NAME_LEN];
@@ -39,6 +41,7 @@ struct elem {
     bool    changed;
 
     neu_dvalue_t value;
+    neu_dvalue_t value_old;
 
     neu_tag_meta_t metas[NEU_TAG_META_SIZE];
 
@@ -142,8 +145,91 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
     HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
     if (elem != NULL) {
         elem->timestamp = timestamp;
-        if (elem->value.type != value.type) {
+
+        if (sub_filter_err && value.type == NEU_TYPE_ERROR) {
+            goto error_not_report;
+        }
+
+        if ((!sub_filter_err && elem->value.type != value.type) ||
+            (sub_filter_err && elem->value.type != value.type &&
+             elem->value.type != NEU_TYPE_ERROR)) {
             elem->changed = true;
+        } else if (sub_filter_err && elem->value.type != value.type &&
+                   elem->value.type == NEU_TYPE_ERROR) {
+            switch (value.type) {
+            case NEU_TYPE_INT8:
+            case NEU_TYPE_UINT8:
+            case NEU_TYPE_INT16:
+            case NEU_TYPE_UINT16:
+            case NEU_TYPE_INT32:
+            case NEU_TYPE_UINT32:
+            case NEU_TYPE_INT64:
+            case NEU_TYPE_UINT64:
+            case NEU_TYPE_BIT:
+            case NEU_TYPE_BOOL:
+            case NEU_TYPE_STRING:
+            case NEU_TYPE_TIME:
+            case NEU_TYPE_DATA_AND_TIME:
+            case NEU_TYPE_WORD:
+            case NEU_TYPE_DWORD:
+            case NEU_TYPE_LWORD:
+                if (memcmp(&elem->value_old.value, &value.value,
+                           sizeof(value.value)) != 0) {
+                    elem->changed = true;
+                }
+                break;
+            case NEU_TYPE_BYTES:
+                if (elem->value_old.value.bytes.length !=
+                    value.value.bytes.length) {
+                    elem->changed = true;
+                } else {
+                    if (memcpy(elem->value_old.value.bytes.bytes,
+                               value.value.bytes.bytes,
+                               value.value.bytes.length) != 0) {
+                        elem->changed = true;
+                    }
+                }
+                break;
+            case NEU_TYPE_PTR: {
+                if (elem->value_old.value.ptr.length !=
+                    value.value.ptr.length) {
+                    elem->changed = true;
+                } else {
+                    if (memcmp(elem->value_old.value.ptr.ptr,
+                               value.value.ptr.ptr,
+                               value.value.ptr.length) != 0) {
+                        elem->changed = true;
+                    }
+                }
+
+                break;
+            }
+            case NEU_TYPE_FLOAT:
+                if (elem->value_old.precision == 0) {
+                    elem->changed =
+                        elem->value_old.value.f32 != value.value.f32;
+                } else {
+                    if (fabs(elem->value_old.value.f32 - value.value.f32) >
+                        pow(0.1, elem->value_old.precision)) {
+                        elem->changed = true;
+                    }
+                }
+                break;
+            case NEU_TYPE_DOUBLE:
+                if (elem->value_old.precision == 0) {
+                    elem->changed =
+                        elem->value_old.value.d64 != value.value.d64;
+                } else {
+                    if (fabs(elem->value_old.value.d64 - value.value.d64) >
+                        pow(0.1, elem->value_old.precision)) {
+                        elem->changed = true;
+                    }
+                }
+
+                break;
+            case NEU_TYPE_ERROR:
+                break;
+            }
         } else {
             switch (value.type) {
             case NEU_TYPE_INT8:
@@ -218,6 +304,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
             }
         }
 
+        if (sub_filter_err && value.type != NEU_TYPE_ERROR) {
+            elem->value_old.type      = value.type;
+            elem->value_old.value     = value.value;
+            elem->value_old.precision = value.precision;
+        }
+
+    error_not_report:
+
         if (change) {
             elem->changed = true;
         }

src/adapter/driver/driver.c

@@ -96,7 +96,7 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
                                 neu_tag_cache_type_e cache_type,
                                 neu_driver_cache_t *cache, const char *group,
                                 UT_array *tags, UT_array *tag_values);
-static void read_report_group(int64_t timestamp, int64_t timeout,
+static void read_report_group(bool sub, int64_t timestamp, int64_t timeout,
                               neu_tag_cache_type_e cache_type,
                               neu_driver_cache_t *cache, const char *group,
                               UT_array *tags, UT_array *tag_values);
@@ -264,7 +264,7 @@ static void update_im(neu_adapter_t *adapter, const char *group,
     data->group  = strdup(group);
     utarray_new(data->tags, neu_resp_tag_value_meta_icd());
 
-    read_report_group(global_timestamp, 0,
+    read_report_group(true, global_timestamp, 0,
                       neu_adapter_get_tag_cache_type(&driver->adapter),
                       driver->cache, group, tags, data->tags);
 
@@ -582,7 +582,16 @@ void neu_adapter_driver_read_group_paginate(neu_adapter_driver_t *driver,
 
     neu_resp_read_group_paginate_t resp  = { 0 };
     neu_group_t *                  group = g->group;
-    UT_array *tags = neu_group_query_read_tag(group, cmd->name, cmd->desc);
+    UT_array *                     tags;
+
+    if (cmd->is_error != true && cmd->current_page > 0 && cmd->page_size > 0) {
+        tags = neu_group_query_read_tag_paginate(
+            group, cmd->name, cmd->desc, cmd->current_page, cmd->page_size,
+            &resp.total_count);
+    } else {
+        tags = neu_group_query_read_tag(group, cmd->name, cmd->desc);
+        resp.total_count = utarray_len(tags);
+    }
 
     utarray_new(resp.tags, neu_resp_tag_value_meta_paginate_icd());
 
@@ -1901,7 +1910,7 @@ static int report_callback(void *usr_data)
     data->group  = strdup(group->name);
     utarray_new(data->tags, neu_resp_tag_value_meta_icd());
 
-    read_report_group(global_timestamp,
+    read_report_group(false, global_timestamp,
                       neu_group_get_interval(group->group) *
                           NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
                       neu_adapter_get_tag_cache_type(&group->driver->adapter),
@@ -2078,7 +2087,7 @@ static int read_callback(void *usr_data)
     return 0;
 }
 
-static void read_report_group(int64_t timestamp, int64_t timeout,
+static void read_report_group(bool sub, int64_t timestamp, int64_t timeout,
                               neu_tag_cache_type_e cache_type,
                               neu_driver_cache_t *cache, const char *group,
                               UT_array *tags, UT_array *tag_values)
@@ -2088,7 +2097,7 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
         neu_driver_cache_value_t  value     = { 0 };
         neu_resp_tag_value_meta_t tag_value = { 0 };
 
-        if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE)) {
+        if (sub && neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE)) {
             if (neu_driver_cache_meta_get_changed(cache, group, tag->name,
                                                   &value, tag_value.metas,
                                                   NEU_TAG_META_SIZE) != 0) {
@@ -2447,6 +2456,13 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
         tag_value.datatag.option      = tag->option;
         memcpy(tag_value.datatag.meta, tag->meta, NEU_TAG_META_LENGTH);
 
+        if (tag->attribute == NEU_ATTRIBUTE_WRITE) {
+            tag_value.value.type         = NEU_TYPE_STRING;
+            tag_value.value.value.str[0] = 0;
+            utarray_push_back(tag_values, &tag_value);
+            continue;
+        }
+
         if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
                                       tag_value.metas,
                                       NEU_TAG_META_SIZE) != 0) {

src/argparse.c

@@ -80,6 +80,7 @@ const char *usage_text =
 "    --plugin_dir <DIR>   directory from which neuron loads plugin lib files\n"
 "    --syslog_host <HOST> syslog server host to which neuron will send logs\n"
 "    --syslog_port <PORT> syslog server port (default 541 if not provided)\n"
+"    --sub_filter_error The subscribe attribute only detects the last read value and does not report any error tags\n"
 "\n";
 // clang-format on
 
@@ -181,6 +182,19 @@ static inline int load_env(neu_cli_args_t *args, char **log_level_out,
             }
         }
 
+        char *sub_filter_e = getenv(NEU_ENV_SUB_FILTER_ERROR);
+        if (sub_filter_e != NULL) {
+            if (strcmp(sub_filter_e, "1") == 0) {
+                args->sub_filter_err = true;
+            } else if (strcmp(sub_filter_e, "0") == 0) {
+                args->sub_filter_err = false;
+            } else {
+                printf("neuron NEURON_SUB_FILTER_ERROR setting error!\n");
+                ret = -1;
+                break;
+            }
+        }
+
         char *log_level = getenv(NEU_ENV_LOG_LEVEL);
         if (log_level != NULL) {
             if (*log_level_out != NULL) {
@@ -289,6 +303,7 @@ static inline int load_config_file(int argc, char *argv[],
         { .name = "port", .t = NEU_JSON_INT },
         { .name = "syslog_host", .t = NEU_JSON_STR },
         { .name = "syslog_port", .t = NEU_JSON_INT },
+        { .name = "sub_filter_error", .t = NEU_JSON_INT },
     };
 
     resolve_config_file_path(argc, argv, long_options, opts, &config_file);
@@ -369,6 +384,16 @@ static inline int load_config_file(int argc, char *argv[],
                 break;
             }
 
+            if (elems[5].v.val_int == 1) {
+                args->sub_filter_err = true;
+            } else if (elems[5].v.val_int == 0) {
+                args->sub_filter_err = false;
+            } else {
+                printf("config file %s sub_filter_error setting error!\n",
+                       config_file);
+                break;
+            }
+
             ret = 0;
         } else {
             printf("config file %s elems error! must had ip, port and "
@@ -424,6 +449,7 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
         { "stop", no_argument, NULL, 's' },
         { "syslog_host", required_argument, NULL, 'S' },
         { "syslog_port", required_argument, NULL, 'P' },
+        { "sub_filter_error", no_argument, NULL, 'f' },
         { NULL, 0, NULL, 0 },
     };
 
@@ -517,6 +543,9 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
             }
             args->syslog_host = strdup(optarg);
             break;
+        case 'f':
+            args->sub_filter_err = true;
+            break;
         case '?':
         default:
             usage();

src/argparse.h

@@ -35,6 +35,7 @@
 #define NEU_ENV_PLUGIN_DIR "NEURON_PLUGIN_DIR"
 #define NEU_ENV_SYSLOG_HOST "NEURON_SYSLOG_HOST"
 #define NEU_ENV_SYSLOG_PORT "NEURON_SYSLOG_PORT"
+#define NEU_ENV_SUB_FILTER_ERROR "NEURON_SUB_FILTER_ERROR"
 
 #define NEURON_CONFIG_FNAME "./config/neuron.json"
 
@@ -64,6 +65,7 @@ typedef struct {
     int      port;
     char *   syslog_host;
     uint16_t syslog_port;
+    bool     sub_filter_err;
 } neu_cli_args_t;
 
 /** Parse command line arguments.

src/base/group.c

@@ -209,6 +209,34 @@ filter_tags(tag_elem_t *tags,
     return array;
 }
 
+static inline UT_array *filter_tags_by_page(
+    tag_elem_t *tags, bool (*predicate)(const neu_datatag_t *, void *data),
+    void *data, int current_page, int page_size, int *total_count)
+{
+    tag_elem_t *el = NULL, *tmp = NULL;
+    UT_array *  array       = NULL;
+    int         count       = 0;
+    int         start_index = (current_page - 1) * page_size;
+    int         end_index   = start_index + page_size;
+
+    utarray_new(array, neu_tag_get_icd());
+    HASH_ITER(hh, tags, el, tmp)
+    {
+        if (predicate(el->tag, data)) {
+            if (count >= start_index && count < end_index) {
+                utarray_push_back(array, el->tag);
+            }
+            count++;
+        }
+    }
+
+    if (total_count) {
+        *total_count = count;
+    }
+
+    return array;
+}
+
 static inline bool is_readable(const neu_datatag_t *tag, void *data)
 {
     (void) data;
@@ -275,6 +303,25 @@ UT_array *neu_group_query_read_tag(neu_group_t *group, const char *name,
     return array;
 }
 
+UT_array *neu_group_query_read_tag_paginate(neu_group_t *group,
+                                            const char *name, const char *desc,
+                                            int current_page, int page_size,
+                                            int *total_count)
+{
+    UT_array *   array = NULL;
+    struct query q     = {
+        .name = (char *) name,
+        .desc = (char *) desc,
+    };
+
+    pthread_mutex_lock(&group->mtx);
+    array = filter_tags_by_page(group->tags, match_query, &q, current_page,
+                                page_size, total_count);
+    pthread_mutex_unlock(&group->mtx);
+
+    return array;
+}
+
 UT_array *neu_group_get_read_tag(neu_group_t *group)
 {
     UT_array *array = NULL;

src/base/group.h

@@ -43,6 +43,10 @@ UT_array *   neu_group_query_tag(neu_group_t *group, const char *name);
 UT_array *   neu_group_get_read_tag(neu_group_t *group);
 UT_array *   neu_group_query_read_tag(neu_group_t *group, const char *name,
                                       const char *desc);
+UT_array *   neu_group_query_read_tag_paginate(neu_group_t *group,
+                                               const char *name, const char *desc,
+                                               int current_page, int page_size,
+                                               int *total_count);
 uint16_t     neu_group_tag_size(const neu_group_t *group);
 neu_datatag_t *neu_group_find_tag(neu_group_t *group, const char *tag);
 void neu_group_split_static_tags(neu_group_t *group, UT_array **static_tags,

src/base/metrics.c

@@ -313,6 +313,7 @@ int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name,
         entry->value = init;
     }
 
+    entry->init = init;
     entry->name = name;
     entry->type = type;
     entry->help = help;

src/connection/mqtt_client.c

@@ -123,6 +123,7 @@ struct neu_mqtt_client_s {
     neu_mqtt_client_connection_cb_t disconnect_cb;
     void *                          disconnect_cb_data;
     nng_mqtt_sqlite_option *        sqlite_cfg;
+    char *                          db;
     bool                            receiving;
     nng_aio *                       recv_aio;
     subscription_t *                subscriptions;
@@ -541,8 +542,10 @@ static void connect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     neu_mqtt_client_t *             client = arg;
     neu_mqtt_client_connection_cb_t cb     = NULL;
     void *                          data   = NULL;
+    int                             reason = 0;
 
-    log(notice, "mqtt client connected");
+    nng_pipe_get_int(p, NNG_OPT_MQTT_CONNECT_REASON, &reason);
+    log(notice, "mqtt client connected, reason: %d", reason);
 
     nng_mtx_lock(client->mtx);
     // start receiving
@@ -567,8 +570,11 @@ static void disconnect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     subscription_t *                sub    = NULL;
     neu_mqtt_client_connection_cb_t cb     = NULL;
     void *                          data   = NULL;
+    int                             reason = 0;
+
+    nng_pipe_get_int(p, NNG_OPT_MQTT_DISCONNECT_REASON, &reason);
+    log(notice, "mqtt client disconnected, reason: %d", reason);
 
-    log(notice, "mqtt client disconnected");
     nng_mtx_lock(client->mtx);
     client->connected = false;
     cb                = client->disconnect_cb;
@@ -901,26 +907,10 @@ static inline nng_mqtt_sqlite_option *
 alloc_sqlite_config(neu_mqtt_client_t *client)
 {
     int                     rv;
-    char *                  db  = NULL;
     nng_mqtt_sqlite_option *cfg = NULL;
-    const mqtt_buf          client_id =
-        nng_mqtt_msg_get_connect_client_id(client->conn_msg);
-
-    if (NULL == client_id.buf || 0 == client_id.length) {
-        log(error, "nng_mqtt_msg_get_connect_client_id fail");
-        return NULL;
-    }
-
-    neu_asprintf(&db, "neuron-mqtt-client-%.*s.db", (unsigned) client_id.length,
-                 (char *) client_id.buf);
-    if (NULL == db) {
-        log(error, "neu_asprintf nano-mqtt cilent db fail");
-        return NULL;
-    }
 
     if ((rv = nng_mqtt_alloc_sqlite_opt(&cfg)) != 0) {
         log(error, "nng_mqtt_alloc_sqlite_opt fail: %s", nng_strerror(rv));
-        free(db);
         return NULL;
     }
 
@@ -928,9 +918,7 @@ alloc_sqlite_config(neu_mqtt_client_t *client)
     nng_mqtt_set_sqlite_flush_threshold(cfg, 256);
     nng_mqtt_set_sqlite_max_rows(cfg, 256);
     nng_mqtt_set_sqlite_db_dir(cfg, "persistence/");
-    nng_mqtt_sqlite_db_init(cfg, db, client->version);
 
-    free(db);
     return cfg;
 }
 
@@ -989,6 +977,7 @@ void neu_mqtt_client_free(neu_mqtt_client_t *client)
         subscriptions_free(client->subscriptions);
         tasks_free(client->task_free_list);
         nng_msg_free(client->conn_msg);
+        free(client->db);
         free(client->url);
         free(client->host);
         nng_mtx_free(client->mtx);
@@ -1213,6 +1202,76 @@ end:
     return rv;
 }
 
+static void rand_str(char *dst, size_t len)
+{
+    char alphabet[] = "0123456789"
+                      "abcdefghijklmnopqrstuvwxyz"
+                      "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
+
+    while (len-- > 1) {
+        size_t i = (double) rand() / RAND_MAX * ((sizeof alphabet) - 1);
+        *dst++   = alphabet[i];
+    }
+    *dst = '\0';
+}
+
+static int neu_mqtt_client_set_cache_db(neu_mqtt_client_t *client)
+{
+    char str[8] = { 0 };
+    rand_str(str, sizeof(str));
+
+    char *db = NULL;
+    neu_asprintf(&db, "neuron-mqtt-client-%s.db", str);
+    if (NULL == db) {
+        log(error, "neu_asprintf nano-mqtt cilent db fail");
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (client->db && 0 == strcmp(client->db, db)) {
+        free(db);
+        return 0;
+    }
+
+    neu_mqtt_client_remove_cache_db(client);
+
+    free(client->db);
+    client->db = db;
+
+    if (client->sqlite_cfg) {
+        nng_mqtt_sqlite_db_init(client->sqlite_cfg, db, client->version);
+    }
+
+    log(notice, "set cache db %s", db);
+
+    return 0;
+}
+
+void neu_mqtt_client_remove_cache_db(neu_mqtt_client_t *client)
+{
+    if (NULL == client->db) {
+        return;
+    }
+
+    const char *dir       = "persistence";
+    int         path_size = strlen(dir) + strlen(client->db) + 6;
+    char *      path      = calloc(path_size, 1);
+    if (NULL == path) {
+        return;
+    }
+
+    snprintf(path, path_size, "%s/%s", dir, client->db);
+    log(notice, "rm cache db %s", path);
+    remove(path);
+    snprintf(path, path_size, "%s/%s-shm", dir, client->db);
+    log(notice, "rm cache db %s", path);
+    remove(path);
+    snprintf(path, path_size, "%s/%s-wal", dir, client->db);
+    log(notice, "rm cache db %s", path);
+    remove(path);
+
+    free(path);
+}
+
 int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
                                    size_t mem_size_bytes, size_t db_size_bytes)
 {
@@ -1227,6 +1286,7 @@ int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
         if (client->sqlite_cfg) {
             nng_mqtt_free_sqlite_opt(client->sqlite_cfg);
             client->sqlite_cfg = NULL;
+            neu_mqtt_client_remove_cache_db(client);
         }
         goto end;
     }
@@ -1240,6 +1300,10 @@ int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
         }
     }
 
+    if (0 != (rv = neu_mqtt_client_set_cache_db(client))) {
+        return rv;
+    }
+
     // FIXME: until NanoSDK cache size limit feature out
     // Assume that,
     // 1. every upload message follows the below template

src/core/manager.c

@@ -128,6 +128,8 @@ neu_manager_t *neu_manager_create()
 
     manager->timestamp_lev_manager = 0;
 
+    strncpy(g_status, "loading", sizeof(g_status));
+
     neu_metrics_init();
     start_static_adapter(manager, DEFAULT_DASHBOARD_PLUGIN_NAME);
 
@@ -156,6 +158,7 @@ neu_manager_t *neu_manager_create()
     manager->timer_timestamp =
         neu_event_add_timer(manager->events, timestamp_timer_param);
 
+    strncpy(g_status, "ready", sizeof(g_status));
     nlog_notice("manager start");
     return manager;
 }
@@ -705,9 +708,11 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     }
     case NEU_REQ_ADD_NODE: {
         neu_req_add_node_t *cmd = (neu_req_add_node_t *) &header[1];
-        int                 error =
+        nlog_notice("add node name:%s plugin:%s", cmd->node, cmd->plugin);
+        int error =
             neu_manager_add_node(manager, cmd->node, cmd->plugin, cmd->setting,
                                  NEU_NODE_RUNNING_STATE_INIT, false);
+
         neu_resp_error_t e = { .error = error };
 
         if (error == NEU_ERR_SUCCESS) {
@@ -726,7 +731,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_UPDATE_NODE: {
         neu_req_update_node_t *cmd = (neu_req_update_node_t *) &header[1];
         neu_resp_error_t       e   = { 0 };
-
+        nlog_notice("update node name:%s new_name:%s", cmd->node,
+                    cmd->new_name);
         if (NULL == neu_node_manager_find(manager->node_manager, cmd->node)) {
             e.error = NEU_ERR_NODE_NOT_EXIST;
         } else if (NULL !=
@@ -750,7 +756,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_DEL_NODE: {
         neu_req_del_node_t *cmd   = (neu_req_del_node_t *) &header[1];
         neu_resp_error_t    error = { 0 };
-        neu_adapter_t *     adapter =
+        nlog_notice("del node name:%s", cmd->node);
+        neu_adapter_t *adapter =
             neu_node_manager_find(manager->node_manager, cmd->node);
         bool single =
             neu_node_manager_is_single(manager->node_manager, cmd->node);
@@ -794,10 +801,22 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             }
             utarray_free(subscriptions);
         }
+
         neu_reqresp_node_deleted_t resp = { 0 };
         strcpy(resp.node, header->receiver);
+        // notify MQTT about node removal
+        if (0 == strcmp(adapter->module->module_name, "MQTT")) {
+            msg = neu_msg_new(NEU_REQRESP_NODE_DELETED, NULL, &resp);
+            if (NULL != msg) {
+                neu_reqresp_head_t *hd = neu_msg_get_header(msg);
+                strcpy(hd->receiver, cmd->node);
+                strcpy(hd->sender, "manager");
+                reply(manager, hd, &resp);
+            }
+        }
         header->type = NEU_REQ_NODE_UNINIT;
         forward_msg(manager, header, header->receiver);
+
         if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_DRIVER) {
             UT_array *apps = neu_subscribe_manager_find_by_driver(
                 manager->subscribe_manager, resp.node);
@@ -848,6 +867,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_resp_error_t     error    = { 0 };
         uint16_t             app_port = 0;
 
+        nlog_notice("add sub app_name:%s driver_node:%s group:%s params:%s",
+                    cmd->app, cmd->driver, cmd->group,
+                    cmd->params != NULL ? cmd->params : "");
+
         error.error = neu_manager_subscribe(manager, cmd->app, cmd->driver,
                                             cmd->group, cmd->params, &app_port);
 
@@ -873,7 +896,9 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         for (uint16_t i = 0; i < cmd->n_group; ++i) {
             neu_req_subscribe_group_info_t *info = &cmd->groups[i];
-
+            nlog_notice("add sub app_name:%s driver_node:%s group:%s params:%s",
+                        cmd->app, info->driver, info->group,
+                        info->params != NULL ? info->params : "");
             error.error =
                 neu_manager_subscribe(manager, cmd->app, info->driver,
                                       info->group, info->params, &info->port);
@@ -901,7 +926,9 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: {
         neu_req_subscribe_t *cmd   = (neu_req_subscribe_t *) &header[1];
         neu_resp_error_t     error = { 0 };
-
+        nlog_notice("update sub app_name:%s driver_node:%s group:%s params:%s",
+                    cmd->app, cmd->driver, cmd->group,
+                    cmd->params != NULL ? cmd->params : "");
         error.error = neu_manager_update_subscribe(
             manager, cmd->app, cmd->driver, cmd->group, cmd->params);
 
@@ -921,7 +948,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_UNSUBSCRIBE_GROUP: {
         neu_req_unsubscribe_t *cmd   = (neu_req_unsubscribe_t *) &header[1];
         neu_resp_error_t       error = { 0 };
-
+        nlog_notice("del sub app_name:%s driver_node:%s group:%s", cmd->app,
+                    cmd->driver, cmd->group);
         error.error =
             neu_manager_unsubscribe(manager, cmd->app, cmd->driver, cmd->group);
 

src/core/manager_internal.c

@@ -243,9 +243,11 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
     // guard against empty mqtt topic parameter
     // this is not an elegant solution due to the current architecture
     if (params && 0 == strcmp(adapter->module->module_name, "MQTT")) {
-        neu_json_elem_t elem = { .name = "topic", .t = NEU_JSON_STR };
-        neu_parse_param(params, NULL, 1, &elem);
-        if (elem.v.val_str && 0 == strlen(elem.v.val_str)) {
+        neu_json_elem_t elem = { .name      = "topic",
+                                 .t         = NEU_JSON_STR,
+                                 .v.val_str = NULL };
+        int             ret  = neu_parse_param(params, NULL, 1, &elem);
+        if (ret != 0 || (elem.v.val_str && 0 == strlen(elem.v.val_str))) {
             free(elem.v.val_str);
             return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
         }

src/event/event_linux.c

@@ -108,7 +108,7 @@ static void *event_loop(void *arg)
     neu_events_t *events   = (neu_events_t *) arg;
     int           epoll_fd = events->epoll_fd;
 
-    while (true) {
+    while (!events->stop) {
         struct epoll_event event = { 0 };
         struct event_data *data  = NULL;
 

src/main.c

@@ -39,8 +39,10 @@ static bool           exit_flag         = false;
 static neu_manager_t *g_manager         = NULL;
 zlog_category_t *     neuron            = NULL;
 bool                  disable_jwt       = false;
+bool                  sub_filter_err    = false;
 int                   default_log_level = ZLOG_LEVEL_NOTICE;
 char                  host_port[32]     = { 0 };
+char                  g_status[32]      = { 0 };
 
 int64_t global_timestamp = 0;
 
@@ -164,7 +166,8 @@ int main(int argc, char *argv[])
     global_timestamp = neu_time_ms();
     neu_cli_args_init(&args, argc, argv);
 
-    disable_jwt = args.disable_auth;
+    disable_jwt    = args.disable_auth;
+    sub_filter_err = args.sub_filter_err;
     snprintf(host_port, sizeof(host_port), "http://%s:%d", args.ip, args.port);
 
     if (args.daemonized) {

tests/ft/http_api/test_http_api.py

@@ -222,12 +222,13 @@ driver_2 = {
     ]
 }
 
+
 class TestHttp:
 
     @description(given="running neuron", when="add driver/app node/group/tag, app subscribes driver group, and get global config", then="success")
     def test_get_global_config(self):
         response = api.put_global_config(json=global_config)
-        assert 200 == response.status_code     
+        assert 200 == response.status_code
         assert error.NEU_ERR_SUCCESS == response.json()['error']
 
         response = api.get_global_config()
@@ -258,7 +259,7 @@ class TestHttp:
         assert config_data['settings'][0]['node'] == 'mqtt'
 
         response = api.put_global_config(json=global_config_del_modbus)
-        assert 200 == response.status_code     
+        assert 200 == response.status_code
         assert error.NEU_ERR_SUCCESS == response.json()['error']
 
         response = api.get_global_config()
@@ -286,41 +287,39 @@ class TestHttp:
         for node in global_config['nodes']:
             api.del_node(node['name'])
 
-
     @description(given="running neuron", when="put drivers with too long node name", then="should fail")
     def test_put_drivers_with_long_node_name(self):
-        response = api.put_driver(name='long'*100, plugin=driver_1['plugin'], params=driver_1['params'])
+        response = api.put_driver(
+            name='long'*100, plugin=driver_1['plugin'], params=driver_1['params'])
         assert 400 == response.status_code
         assert error.NEU_ERR_NODE_NAME_TOO_LONG == response.json()['error']
 
-
     @description(given="running neuron", when="put drivers with duplicate node name", then="should fail")
     def test_put_drivers_with_duplicate_node(self):
         response = api.put_drivers([driver_1, driver_1])
         assert 400 == response.status_code
         assert error.NEU_ERR_BODY_IS_WRONG == response.json()['error']
 
-
     @description(given="running neuron", when="put drivers with non-existent plugin", then="should fail")
     def test_put_drivers_with_nonexistent_plugin(self):
-        response = api.put_driver(name=driver_1['name'], plugin='unkown', params=driver_1['params'])
+        response = api.put_driver(
+            name=driver_1['name'], plugin='unkown', params=driver_1['params'])
         assert 404 == response.status_code
         assert error.NEU_ERR_LIBRARY_NOT_FOUND == response.json()['error']
 
-
     @description(given="running neuron", when="put drivers with too long plugin name", then="should fail")
     def test_put_drivers_with_long_plugin_name(self):
-        response = api.put_driver(name=driver_1['name'], plugin='long'*100, params=driver_1['params'])
+        response = api.put_driver(
+            name=driver_1['name'], plugin='long'*100, params=driver_1['params'])
         assert 400 == response.status_code
         assert error.NEU_ERR_PLUGIN_NAME_TOO_LONG == response.json()['error']
 
-
     @description(given="running neuron", when="put drivers with north plugin", then="should fail")
     def test_put_drivers_with_north_plugin(self):
         response = api.put_driver(name='not-driver', plugin='MQTT', params={})
         assert 400 == response.status_code
-        assert error.NEU_ERR_PLUGIN_TYPE_NOT_SUPPORT == response.json()['error']
-
+        assert error.NEU_ERR_PLUGIN_TYPE_NOT_SUPPORT == response.json()[
+            'error']
 
     @description(given="running neuron", when="put drivers with too long group name", then="should fail")
     def test_put_drivers_with_long_group_name(self):
@@ -330,7 +329,6 @@ class TestHttp:
         assert 400 == response.status_code
         assert error.NEU_ERR_GROUP_NAME_TOO_LONG == response.json()['error']
 
-
     @description(given="running neuron", when="put drivers with duplicate group name", then="should fail")
     def test_put_drivers_with_duplicate_group(self):
         driver = copy.deepcopy(driver_1)
@@ -339,15 +337,14 @@ class TestHttp:
         assert 400 == response.status_code
         assert error.NEU_ERR_BODY_IS_WRONG == response.json()['error']
 
-
     @description(given="running neuron", when="put drivers with invalid group interval", then="should fail")
     def test_put_drivers_with_invalid_group_interval(self):
         driver = copy.deepcopy(driver_1)
         driver["groups"][0]["interval"] = 10
         response = api.put_drivers([driver])
         assert 400 == response.status_code
-        assert error.NEU_ERR_GROUP_PARAMETER_INVALID == response.json()['error']
-
+        assert error.NEU_ERR_GROUP_PARAMETER_INVALID == response.json()[
+            'error']
 
     @description(given="running neuron", when="put drivers with too many groups", then="should fail")
     def test_put_drivers_with_too_many_groups(self):
@@ -361,7 +358,6 @@ class TestHttp:
         assert 400 == response.status_code
         assert error.NEU_ERR_GROUP_MAX_GROUPS == response.json()['error']
 
-
     @description(given="running neuron", when="put drivers with duplicate tag name", then="should fail")
     def test_put_drivers_with_duplicate_tag(self):
         driver = copy.deepcopy(driver_1)
@@ -370,7 +366,6 @@ class TestHttp:
         assert 400 == response.status_code
         assert error.NEU_ERR_BODY_IS_WRONG == response.json()['error']
 
-
     @description(given="running neuron", when="put drivers", then="should success")
     def test_put_drivers(self):
         drivers = [driver_1, driver_2]
@@ -394,7 +389,6 @@ class TestHttp:
 
             api.del_node(driver["name"])
 
-
     @description(given="running neuron", when="put drivers that exists", then="should replace")
     def test_put_drivers_already_exist(self):
         response = api.put_drivers([driver_1])
@@ -426,7 +420,6 @@ class TestHttp:
 
         api.del_node(driver["name"])
 
-
     @description(given="running neuron", when="put drivers", then="should be persisted")
     def test_put_drivers_persistence(self, class_setup_and_teardown):
         drivers = [driver_1, driver_2]
@@ -443,7 +436,6 @@ class TestHttp:
                 assert len(group["tags"]) == len(response.json()["tags"])
             api.del_node(driver["name"])
 
-
     @description(given="running neuron", when="get drivers", then="should success")
     def test_get_drivers(self):
         drivers = [driver_1, driver_2]
@@ -474,25 +466,26 @@ class TestHttp:
         # filter by driver_1 and driver_2
         response = api.get_drivers(names=[driver_1["name"], driver_2["name"]])
         assert 200 == response.status_code
-        resp_drivers = sorted(response.json()["nodes"], key=lambda x: x["name"])
+        resp_drivers = sorted(
+            response.json()["nodes"], key=lambda x: x["name"])
         assert 2 == len(resp_drivers)
         assert drivers == resp_drivers
 
         # no filter
         response = api.get_drivers()
         assert 200 == response.status_code
-        resp_drivers = sorted(response.json()["nodes"], key=lambda x: x["name"])
+        resp_drivers = sorted(
+            response.json()["nodes"], key=lambda x: x["name"])
         assert 2 == len(resp_drivers)
         assert drivers == resp_drivers
 
         for driver in drivers:
             api.del_node(driver["name"])
 
-
     @description(given="running neuron", when="get subscribe info", then="should success")
     def test_get_subscribe_info(self):
         app = 'MQTT-1'
-        params = {'topic':'/neuron/test'}
+        params = {'topic': '/neuron/test'}
         driver_1 = 'modbus-01'
         driver_2 = 'modbus-02'
         group_1 = 'group-01'
@@ -513,7 +506,8 @@ class TestHttp:
             # match all drivers
             response = api.get_subscribe_group(app, driver='modbus')
             assert 200 == response.status_code
-            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            groups = sorted(response.json()['groups'], key=lambda g: (
+                g['driver'], g['group']))
             assert len(groups) == 4
             assert groups[0]['driver'] == driver_1
             assert groups[0]['group'] == group_1
@@ -527,7 +521,8 @@ class TestHttp:
             # match one driver
             response = api.get_subscribe_group(app, driver=driver_1)
             assert 200 == response.status_code
-            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            groups = sorted(response.json()['groups'], key=lambda g: (
+                g['driver'], g['group']))
             assert len(groups) == 2
             assert groups[0]['driver'] == driver_1
             assert groups[0]['group'] == group_1
@@ -542,7 +537,8 @@ class TestHttp:
             # match all groups
             response = api.get_subscribe_group(app, group='group')
             assert 200 == response.status_code
-            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            groups = sorted(response.json()['groups'], key=lambda g: (
+                g['driver'], g['group']))
             assert len(groups) == 4
             assert groups[0]['driver'] == driver_1
             assert groups[0]['group'] == group_1
@@ -556,7 +552,8 @@ class TestHttp:
             # match one group
             response = api.get_subscribe_group(app, group=group_1)
             assert 200 == response.status_code
-            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            groups = sorted(response.json()['groups'], key=lambda g: (
+                g['driver'], g['group']))
             assert len(groups) == 2
             assert groups[0]['driver'] == driver_1
             assert groups[0]['group'] == group_1
@@ -569,15 +566,19 @@ class TestHttp:
             assert len(response.json()['groups']) == 0
 
             # match driver and group
-            response = api.get_subscribe_group(app, driver=driver_1, group=group_1)
+            response = api.get_subscribe_group(
+                app, driver=driver_1, group=group_1)
             assert 200 == response.status_code
-            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            groups = sorted(response.json()['groups'], key=lambda g: (
+                g['driver'], g['group']))
             assert len(groups) == 1
             assert groups[0]['driver'] == driver_1
             assert groups[0]['group'] == group_1
-            response = api.get_subscribe_group(app, driver=driver_2, group=group_2)
+            response = api.get_subscribe_group(
+                app, driver=driver_2, group=group_2)
             assert 200 == response.status_code
-            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            groups = sorted(response.json()['groups'], key=lambda g: (
+                g['driver'], g['group']))
             assert len(groups) == 1
             assert groups[0]['driver'] == driver_2
             assert groups[0]['group'] == group_2
@@ -608,6 +609,11 @@ class TestHttp:
         response = api.get_version()
         assert 200 == response.status_code
 
+    @description(given="running neuron", when="get status", then="success")
+    def test_get_status(self):
+        response = api.get_status()
+        assert 200 == response.status_code
+
     @description(given="running neuron", when="test jwt error", then="failed")
     def test_jwt_err(self):
         response = api.change_password(new_password='123456', jwt='invalid')
@@ -626,4 +632,4 @@ class TestHttp:
         assert 404 == response.status_code
 
         response = api.delete_virtual_license(jwt='invalid')
-        assert 404 == response.status_code
+        assert 404 == response.status_code

tests/ft/login/neuron.json

@@ -3,5 +3,6 @@
 	"port": 7000,
 	"disable_auth": 1,
 	"syslog_host": "",
-	"syslog_port": 541
+	"syslog_port": 541,
+	"sub_filter_error": 0
 }

tests/ft/login/sub_filter_error.json

@@ -0,0 +1,8 @@
+{
+	"ip": "0.0.0.0",
+	"port": 7000,
+	"disable_auth": 0,
+	"syslog_host": "",
+	"syslog_port": 541,
+	"sub_filter_error": 1
+}

tests/ft/login/test_launch.py

@@ -1,14 +1,157 @@
+import pytest
 import neuron.api as api
 from neuron.config import *
-from neuron.common import description
+from neuron.common import (
+    description,
+    random_port,
+)
 import neuron.process as process
+import neuron.config as config
 from neuron.error import *
 import shutil
 import os
+from neuron.mqtt import Mock
+import time
 
+mqtt_node = "mqtt"
+upload_topic = "/neuron/test/sub_filter"
+
+modbus_port = random_port()
+modbus_node = "modbus_tcp"
+
+hold_bit = [{"name": "hold_bit", "address": "1!400001.15",
+             "attribute": config.NEU_TAG_ATTRIBUTE_READ_SUBSCRIBE, "type": config.NEU_TYPE_BIT}]
+hold_int16 = [{"name": "hold_int16", "address": "1!400001",
+               "attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_INT16}]
+hold_uint16 = [{"name": "hold_uint16", "address": "1!400002",
+                "attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_UINT16}]
+hold_int32 = [{"name": "hold_int32", "address": "1!400003",
+               "attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_INT32}]
+hold_uint32 = [{"name": "hold_uint32", "address": "1!400015",
+                "attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_UINT32}]
+hold_float = [{"name": "hold_float", "address": "1!400017",
+               "attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_FLOAT}]
+hold_string = [{"name": "hold_string", "address": "1!400020.10",
+                "attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_STRING}]
+
+
+@pytest.fixture(autouse=True, scope="class")
+def mocker():
+    mock = Mock()
+    yield mock
+    mock.done()
 
 class TestLaunch:
 
+    @description(
+        given="sub_filter_error(command line)",
+        when="start neuron",
+        then="should make sense",
+    )
+    def test_sub_filter_error_command(self, mocker):
+        process.remove_persistence()
+
+        p = process.NeuronProcess()
+        p.start_sub_filter_err()
+
+        api.add_node(node=mqtt_node, plugin=config.PLUGIN_MQTT)
+        api.mqtt_node_setting(node=mqtt_node, client_id=mocker.rand_id(), host="127.0.0.1", port=mocker.port)
+
+        api.add_node(node=modbus_node, plugin=config.PLUGIN_MODBUS_TCP)
+        api.modbus_tcp_node_setting(node=modbus_node, port=modbus_port)
+        api.add_group(node=modbus_node, group='group')
+        api.add_tags_check(node=modbus_node, group='group', tags=hold_bit)
+        api.add_tags_check(node=modbus_node, group='group', tags=hold_int16)
+        api.add_tags_check(node=modbus_node, group='group', tags=hold_uint16)
+        api.add_tags_check(node=modbus_node, group='group', tags=hold_int32)
+        api.add_tags_check(node=modbus_node, group='group', tags=hold_uint32)
+        api.add_tags_check(node=modbus_node, group='group', tags=hold_float)
+        api.add_tags_check(node=modbus_node, group='group', tags=hold_string)
+
+        api.subscribe_group(app=mqtt_node, driver=modbus_node, group='group', params={"topic":upload_topic})
+
+        msg = mocker.get(upload_topic, timeout=3)
+        assert msg is None
+
+        p_s = process.start_simulator(
+            ["./modbus_simulator", "tcp", f"{modbus_port}", "ip_v4"]
+        )
+
+        time.sleep(5)
+
+        msg = mocker.get(upload_topic, timeout=3)
+        assert msg is None        
+
+        p.stop()
+        process.stop_simulator(p_s)
+
+    @description(
+        given="sub_filter_error(environment variable)",
+        when="start neuron",
+        then="should make sense",
+    )
+    def test_sub_filter_error_environment(self, mocker):
+
+        os.environ['NEURON_SUB_FILTER_ERROR'] = '1'
+
+        p = process.NeuronProcess()
+        p.start()
+
+        msg = mocker.get(upload_topic, timeout=3)
+        assert msg is None
+
+        p_s = process.start_simulator(
+            ["./modbus_simulator", "tcp", f"{modbus_port}", "ip_v4"]
+        )
+
+        time.sleep(5)
+
+        msg = mocker.get(upload_topic, timeout=3)
+        assert msg is None        
+
+        del os.environ['NEURON_SUB_FILTER_ERROR']
+
+        p.stop()
+        process.stop_simulator(p_s)
+
+    @description(
+        given="sub_filter_error(configuration file)",
+        when="start neuron",
+        then="should make sense",
+    )
+    def test_sub_filter_error_file(self, mocker):
+        dst_file = './build/config/neuron.json'
+        dst_backup = dst_file + '_backup'
+        os.rename(dst_file, dst_backup)
+        try:
+            shutil.copy2('./tests/ft/login/sub_filter_error.json', dst_file)
+        except Exception as e:
+            if os.path.exists(dst_backup):
+                os.rename(dst_backup, dst_file)
+            return
+
+        p = process.NeuronProcess()
+        p.start()
+
+        msg = mocker.get(upload_topic, timeout=3)
+        assert msg is None
+
+        p_s = process.start_simulator(
+            ["./modbus_simulator", "tcp", f"{modbus_port}", "ip_v4"]
+        )
+
+        time.sleep(5)
+
+        msg = mocker.get(upload_topic, timeout=3)
+        assert msg is None        
+
+        p.stop()
+        process.stop_simulator(p_s)
+
+        os.remove(dst_file)
+        os.rename(dst_backup, dst_file)
+
+
     @description(given="encironment variables 0, configuration file 0 , command line 1",
                  when="start neuron",
                  then="command line has the highest priority")

tests/ft/neuron/api.py

@@ -22,6 +22,7 @@ def gen_check(func):
 def ping():
     return requests.post(url=config.BASE_URL + '/api/v2/ping')
 
+
 def login(user='admin', password='0000'):
     return requests.post(url=config.BASE_URL + '/api/v2/login', json={"name": user, "pass": password})
 
@@ -157,15 +158,18 @@ def read_tags(node, group, sync=False, query=None):
         body["query"] = query
     return requests.post(url=config.BASE_URL + "/api/v2/read", headers={"Authorization": config.default_jwt}, json=body)
 
+
 def read_tags_paginate(node, group, sync=False, query=None):
     body = {"node": node, "group": group, "sync": sync}
     if query:
         body["query"] = query
     return requests.post(url=config.BASE_URL + "/api/v2/read/paginate", headers={"Authorization": config.default_jwt}, json=body)
 
+
 def test_read_tag(json):
     return requests.post(url=config.BASE_URL + '/api/v2/read/test', headers={"Authorization": config.default_jwt}, json=json)
 
+
 def read_tag(node, group, tag, sync=False):
     response = read_tags(node, group, sync, query={"name": tag})
     assert 200 == response.status_code
@@ -243,12 +247,12 @@ def get_subscribe_group(app, driver=None, group=None):
 
 
 @gen_check
-def subscribe_group(app, driver, group, params={}):
+def subscribe_group(app, driver, group, params={"topic": "/neuron/mqtt/test"}):
     return requests.post(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group, "params": params})
 
 
 @gen_check
-def subscribe_group_update(app, driver, group, params={}):
+def subscribe_group_update(app, driver, group, params={"topic": "/neuron/mqtt/test"}):
     return requests.put(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group, "params": params})
 
 
@@ -284,6 +288,10 @@ def put_drivers(drivers, jwt=config.default_jwt):
     return requests.put(url=config.BASE_URL + '/api/v2/global/drivers', headers={"Authorization": jwt}, json={"nodes": drivers})
 
 
+def get_status():
+    return requests.get(url=config.BASE_URL + "/api/v2/status", headers={"Authorization": config.default_jwt})
+
+
 def get_driver(name, jwt=config.default_jwt):
     return get_drivers(names=[name])
 
@@ -328,6 +336,7 @@ def modbus_rtu_node_setting(node, port=502, connection_mode=0, transport_mode=0,
                                     "max_retries": max_retries, "link": link, "device": device, "stop": stop,
                                     "parity": parity, "baud": baud, "data": data})
 
+
 def modbus_tcp_node_setting_endian(node, port, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, endianness=1):
     return node_setting(node, json={"connection_mode": connection_mode, "transport_mode": transport_mode, "interval": interval,
                                     "host": host, "port": port, "timeout": timeout, "max_retries": max_retries, "endianess": endianness})
@@ -339,6 +348,7 @@ def modbus_rtu_node_setting_endian(node, port=502, connection_mode=0, transport_
                                     "max_retries": max_retries, "link": link, "device": device, "stop": stop,
                                     "parity": parity, "baud": baud, "data": data, "endianess": endianness})
 
+
 def modbus_tcp_node_setting_base(node, port, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, base=1):
     return node_setting(node, json={"connection_mode": connection_mode, "transport_mode": transport_mode, "interval": interval,
                                     "host": host, "port": port, "timeout": timeout, "max_retries": max_retries, "address_base": base})
@@ -351,10 +361,11 @@ def modbus_rtu_node_setting_base(node, port=502, connection_mode=0, transport_mo
                                     "parity": parity, "baud": baud, "data": data, "address_base": base})
 
 
-def mqtt_node_setting(node):
-    return node_setting(node, json={"client-id": "neuron_aBcDeF", "qos": 0, "format": 0,
+def mqtt_node_setting(node, client_id="neuron_aBcDeF", host="broker.emqx.io", port=1883):
+    return node_setting(node, json={"client-id": client_id, "qos": 0, "format": 0,
                                     "write-req-topic": f"/neuron/{node}/write/req", "write-resp-topic": f"/neuron/{node}/write/resp",
-                                    "offline-cache": False, "cache-sync-interval": 100, "host": "broker.emqx.io", "port": 1883, "username": "", "password": "", "ssl": False})
+                                    "offline-cache": False, "cache-sync-interval": 100, "host": host, "port": port, "username": "", "password": "", "ssl": False})
+
 
 def get_nodes_disable_auth(type):
     return requests.get(url=config.BASE_URL + '/api/v2/node', params={"type": type})

tests/ft/neuron/process.py

@@ -30,6 +30,9 @@ class NeuronProcess:
     def start_debug(self):
         self.p = start_neuron_debug(self.dir)
 
+    def start_sub_filter_err(self):
+        self.p = start_neuron_sub_filter_err(self.dir)
+
 def start_neuron_disable_auth(dir='build/'):
     command = ['./neuron', '--disable_auth']
     process = subprocess.Popen(
@@ -46,6 +49,14 @@ def start_neuron_debug(dir='build/'):
     assert process.poll() is None
     return process
 
+def start_neuron_sub_filter_err(dir='build/'):
+    command = ['./neuron', '--sub_filter_error']
+    process = subprocess.Popen(
+        command, stderr=subprocess.PIPE, cwd=dir)
+    time.sleep(1)
+    assert process.poll() is None
+    return process
+
 def start_neuron(dir='build/'):
     process = subprocess.Popen(
         ['./neuron'], stderr=subprocess.PIPE, cwd=dir)

version

@@ -1 +1 @@
-2.9.0-beta3
+2.9.4