Merge pull request #2737 from vulgraph/backport/CVE-2024-10964-v2.9

[v2.9] Backport CVE-2024-10964: plugin: buffer overflow in library strcpy

.github/workflows/check.yml

@@ -9,7 +9,7 @@ on:
 
 jobs:
   clang_format_check:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-24.04
 
     steps:
       # checkout
@@ -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,13 +8,13 @@ on:
       - published
 
 env:
-  VERSION: 2.9.1
+  VERSION: 2.9.5
   UI_VERSION: 2.6.1
   BRANCH: v2.9 
 
 jobs:
   cross-build:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-24.04
     strategy:
       matrix:
         arch:
@@ -52,14 +52,14 @@ jobs:
         run: |
           ./package-sdk.sh -p ${{ matrix.arch[2] }} -n neuron-sdk-$VERSION
 
-      - uses: actions/upload-artifact@v3
+      - uses: actions/upload-artifact@v4
         with:
-          name: neuron-sdk
+          name: neuron-sdk-${{ matrix.arch[2] }}
           path: |
             neuron-sdk-*.tar.gz
 
   build-debug-sdk:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-24.04
     strategy:
       matrix:
         arch:
@@ -83,7 +83,7 @@ jobs:
         run: |
           ./package-sdk.sh -p ${{ matrix.arch[2] }} -n neuron-sdk-debug-$VERSION
 
-      - uses: actions/upload-artifact@v3
+      - uses: actions/upload-artifact@v4
         with:
           name: neuron-sdk
           path: |
@@ -102,7 +102,7 @@ jobs:
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
-      - uses: actions/download-artifact@v3
+      - uses: actions/download-artifact@v4
         with:
           name: neuron-sdk
       

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/connection/neu_connection.h

@@ -254,6 +254,8 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len);
 ssize_t neu_conn_udp_recvfrom(neu_conn_t *conn, uint8_t *buf, ssize_t len,
                               void *src);
 
+void neu_conn_clear_recv_buffer(neu_conn_t *conn);
+
 /**
  * @brief Specify the client to send data.
  *

include/neuron/define.h

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 9
-#define NEU_VERSION_FIX 1
+#define NEU_VERSION_FIX 5
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)
@@ -34,7 +34,7 @@
 #define NEU_TAG_NAME_LEN 128
 #define NEU_TAG_ADDRESS_LEN 128
 #define NEU_TAG_DESCRIPTION_LEN 128
-#define NEU_GROUP_MAX_PER_NODE 512
+#define NEU_GROUP_MAX_PER_NODE 1024
 #define NEU_GROUP_NAME_LEN 128
 #define NEU_GROUP_INTERVAL_LIMIT 100
 #define NEU_DEFAULT_GROUP_INTERVAL 100
@@ -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/json/neu_json_rw.h

@@ -63,6 +63,7 @@ typedef struct {
 
 int neu_json_encode_read_resp(void *json_object, void *param);
 int neu_json_encode_read_resp1(void *json_object, void *param);
+int neu_json_encode_read_resp2(void *json_object, void *param);
 int neu_json_encode_read_paginate_resp(void *json_object, void *param);
 
 typedef struct {

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/tag.h

@@ -44,6 +44,10 @@ typedef enum {
     NEU_DATATAG_ENDIAN_BL32 = 5, // #BL 2,1,4,3
     NEU_DATATAG_ENDIAN_L64  = 6, // #L  8,7,6,5,4,3,2,1
     NEU_DATATAG_ENDIAN_B64  = 7, // #B  1,2,3,4,5,6,7,8
+    NEU_DATATAG_ENDIAN_LL64 = 8,
+    NEU_DATATAG_ENDIAN_LB64 = 9,
+    NEU_DATATAG_ENDIAN_BB64 = 10,
+    NEU_DATATAG_ENDIAN_BL64 = 11,
 } neu_datatag_endian_e;
 
 typedef enum {
@@ -63,6 +67,7 @@ typedef union {
     } value32;
     struct {
         neu_datatag_endian_e endian;
+        bool                 is_default;
     } value64;
     struct {
         uint16_t                  length;

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;
@@ -506,6 +738,42 @@ static inline uint64_t neu_htonll(uint64_t u)
     return ret;
 }
 
+static inline uint64_t neu_htonlb(uint64_t u)
+{
+    uint8_t *in        = (uint8_t *) &u;
+    uint64_t out       = 0;
+    uint8_t *out_bytes = (uint8_t *) &out;
+
+    out_bytes[0] = in[1];
+    out_bytes[1] = in[0];
+    out_bytes[2] = in[3];
+    out_bytes[3] = in[2];
+    out_bytes[4] = in[5];
+    out_bytes[5] = in[4];
+    out_bytes[6] = in[7];
+    out_bytes[7] = in[6];
+
+    return out;
+}
+
+static inline uint64_t neu_htonbl(uint64_t u)
+{
+    uint8_t *in        = (uint8_t *) &u;
+    uint64_t out       = 0;
+    uint8_t *out_bytes = (uint8_t *) &out;
+
+    out_bytes[0] = in[6];
+    out_bytes[1] = in[7];
+    out_bytes[2] = in[4];
+    out_bytes[3] = in[5];
+    out_bytes[4] = in[2];
+    out_bytes[5] = in[3];
+    out_bytes[6] = in[0];
+    out_bytes[7] = in[1];
+
+    return out;
+}
+
 static inline uint64_t neu_ntohll(uint64_t u)
 {
     return neu_htonll(u);
@@ -513,7 +781,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 +792,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 +803,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 +812,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-rtu.json

@@ -18,11 +18,11 @@
 		},
 		{
 			"type": 7,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 8,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 9,
@@ -30,7 +30,7 @@
 		},
 		{
 			"type": 10,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 11,
@@ -106,6 +106,35 @@
 			"max": 10000
 		}
 	},
+	"endianess_64": {
+		"name": "Endianess of 8-Byte Data",
+		"name_zh": "8 字节数据字节序",
+		"description": "Tag byte order, applicable to int64, uint64 and double types",
+		"description_zh": "点位字节序，适用于 int64，uint64 和 double 类型",
+		"attribute": "required",
+		"type": "map",
+		"default": 1,
+		"valid": {
+			"map": [
+				{
+					"key": "12 34 56 78",
+					"value": 1
+				},
+				{
+					"key": "21 43 65 87",
+					"value": 2
+				},
+				{
+					"key": "87 65 43 21",
+					"value": 3
+				},
+				{
+					"key": "78 56 34 12",
+					"value": 4
+				}
+			]
+		}
+	},
 	"interval": {
 		"name": "Send Interval",
 		"name_zh": "指令发送间隔",

plugins/modbus/modbus-tcp.json

@@ -18,11 +18,11 @@
 		},
 		{
 			"type": 7,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 8,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 9,
@@ -30,7 +30,7 @@
 		},
 		{
 			"type": 10,
-			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 11,
@@ -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": "最大重试次数",
@@ -93,6 +112,35 @@
 			"max": 10000
 		}
 	},
+	"endianess_64": {
+		"name": "Endianess of 8-Byte Data",
+		"name_zh": "8 字节数据字节序",
+		"description": "Tag byte order, applicable to int64, uint64 and double types",
+		"description_zh": "点位字节序，适用于 int64，uint64 和 double 类型",
+		"attribute": "required",
+		"type": "map",
+		"default": 1,
+		"valid": {
+			"map": [
+				{
+					"key": "12 34 56 78",
+					"value": 1
+				},
+				{
+					"key": "21 43 65 87",
+					"value": 2
+				},
+				{
+					"key": "87 65 43 21",
+					"value": 3
+				},
+				{
+					"key": "78 56 34 12",
+					"value": 4
+				}
+			]
+		}
+	},
 	"interval": {
 		"name": "Send Interval",
 		"name_zh": "指令发送间隔",

plugins/modbus/modbus.h

@@ -66,6 +66,13 @@ typedef enum modbus_area {
     MODBUS_AREA_HOLD_REGISTER  = 4,
 } modbus_area_e;
 
+typedef enum modbus_endianess_64 {
+    MODBUS_LL = 1, // 12 34 56 78
+    MODBUS_LB = 2, // 21 43 65 87
+    MODBUS_BB = 3, // 87 65 43 21
+    MODBUS_BL = 4, // 78 56 34 12
+} modbus_endianess_64;
+
 struct modbus_header {
     uint16_t seq;
     uint16_t protocol;

plugins/modbus/modbus_point.c

@@ -230,7 +230,8 @@ modbus_read_cmd_sort_t *modbus_tag_sort(UT_array *tags, uint16_t max_byte)
     return sort_result;
 }
 
-int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option)
+int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option,
+               modbus_endianess_64 endianess_64, bool default_tag_endian_64)
 {
     int n = 0;
     switch (type) {
@@ -249,6 +250,9 @@ int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option)
     case NEU_TYPE_DOUBLE:
     case NEU_TYPE_INT64:
     case NEU_TYPE_UINT64:
+        if (default_tag_endian_64) {
+            modbus_convert_endianess_64(value, endianess_64);
+        }
         n          = sizeof(uint64_t);
         value->u64 = neu_htonll(value->u64);
         break;
@@ -282,7 +286,8 @@ int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option)
     return n;
 }
 
-modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *tags)
+modbus_write_cmd_sort_t *
+modbus_write_tags_sort(UT_array *tags, modbus_endianess_64 endianess_64)
 {
     neu_tag_sort_result_t *result =
         neu_tag_sort(tags, tag_sort_write, tag_cmp_write);
@@ -304,14 +309,17 @@ modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *tags)
         utarray_foreach(result->sorts[i].tags, modbus_point_write_t **, tag_s)
         {
             if ((*tag_s)->point.area == MODBUS_AREA_COIL) {
-                n_byte_tag = cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
-                                        (*tag_s)->point.option);
+                n_byte_tag =
+                    cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
+                               (*tag_s)->point.option, endianess_64, true);
                 data_bit[k / 8] += ((*tag_s)->value.i8) << k % 8;
                 n_byte += n_byte_tag;
                 k++;
             } else {
-                n_byte_tag = cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
-                                        (*tag_s)->point.option);
+                n_byte_tag =
+                    cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
+                               (*tag_s)->point.option, endianess_64,
+                               (*tag_s)->point.option.value64.is_default);
                 memcpy(sort_result->cmd[i].bytes +
                            2 *
                                ((*tag_s)->point.start_address -
@@ -521,4 +529,24 @@ static bool tag_sort_write(neu_tag_sort_t *sort, void *tag,
     }
 
     return true;
+}
+
+void modbus_convert_endianess_64(neu_value_u *       value,
+                                 modbus_endianess_64 endianess_64)
+{
+    switch (endianess_64) {
+    case MODBUS_LL:
+        break;
+    case MODBUS_BB:
+        value->u64 = neu_htonll(value->u64);
+        break;
+    case MODBUS_LB:
+        value->u64 = neu_htonlb(value->u64);
+        break;
+    case MODBUS_BL:
+        value->u64 = neu_htonbl(value->u64);
+        break;
+    default:
+        break;
+    }
 }

plugins/modbus/modbus_point.h

@@ -79,9 +79,13 @@ typedef struct modbus_write_cmd_sort {
     modbus_write_cmd_t *cmd;
 } modbus_write_cmd_sort_t;
 
-modbus_read_cmd_sort_t * modbus_tag_sort(UT_array *tags, uint16_t max_byte);
-modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array *tags);
-void                     modbus_tag_sort_free(modbus_read_cmd_sort_t *cs);
+modbus_read_cmd_sort_t *modbus_tag_sort(UT_array *tags, uint16_t max_byte);
+modbus_write_cmd_sort_t *
+     modbus_write_tags_sort(UT_array *tags, modbus_endianess_64 endianess_64);
+void modbus_tag_sort_free(modbus_read_cmd_sort_t *cs);
+
+void modbus_convert_endianess_64(neu_value_u *       value,
+                                 modbus_endianess_64 endianess_64);
 
 #ifdef __cplusplus
 }

plugins/modbus/modbus_req.c

@@ -107,6 +107,8 @@ int modbus_send_msg(void *ctx, uint16_t n_byte, uint8_t *bytes)
     neu_plugin_t *plugin = (neu_plugin_t *) ctx;
     int           ret    = 0;
 
+    neu_conn_clear_recv_buffer(plugin->conn);
+
     plog_send_protocol(plugin, bytes, n_byte);
 
     if (plugin->is_server) {
@@ -206,6 +208,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 +220,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 +250,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 +259,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 +281,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 +292,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(
@@ -406,6 +414,10 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
             case NEU_TYPE_DOUBLE:
             case NEU_TYPE_INT64:
             case NEU_TYPE_UINT64:
+                if ((*p_tag)->option.value64.is_default) {
+                    modbus_convert_endianess_64(&dvalue.value,
+                                                plugin->endianess_64);
+                }
                 dvalue.value.u64 = neu_ntohll(dvalue.value.u64);
                 break;
             case NEU_TYPE_BIT: {
@@ -669,6 +681,9 @@ int modbus_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
     case NEU_TYPE_DOUBLE:
     case NEU_TYPE_INT64:
     case NEU_TYPE_UINT64:
+        if (tag->option.value64.is_default) {
+            modbus_convert_endianess_64(&value, plugin->endianess_64);
+        }
         value.u64 = neu_htonll(value.u64);
         n_byte    = sizeof(uint64_t);
         break;
@@ -736,7 +751,7 @@ int modbus_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags)
 
         utarray_push_back(gtags->tags, &p);
     }
-    gtags->cmd_sort = modbus_write_tags_sort(gtags->tags);
+    gtags->cmd_sort = modbus_write_tags_sort(gtags->tags, plugin->endianess_64);
     for (uint16_t i = 0; i < gtags->cmd_sort->n_cmd; i++) {
         uint16_t response_size = 0;
 

plugins/modbus/modbus_req.h

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

plugins/modbus/modbus_rtu.c

@@ -167,6 +167,9 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
     neu_json_elem_t retry_interval = { .name = "retry_interval",
                                        .t    = NEU_JSON_INT };
 
+    neu_json_elem_t endianess_64 = { .name = "endianess_64",
+                                     .t    = NEU_JSON_INT };
+
     ret = neu_parse_param((char *) config, &err_param, 3, &link, &timeout,
                           &interval);
     if (ret != 0) {
@@ -206,9 +209,16 @@ 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, &endianess_64);
+    if (ret != 0) {
+        free(err_param);
+        endianess_64.v.val_int = MODBUS_LL;
+    }
+
     param.log              = plugin->common.log;
     plugin->max_retries    = max_retries.v.val_int;
     plugin->retry_interval = retry_interval.v.val_int;
+    plugin->endianess_64   = endianess_64.v.val_int;
 
     if (link.v.val_int == 0) {
         param.type = NEU_CONN_TTY_CLIENT;

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,10 @@ 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 };
+    neu_json_elem_t  endianess_64   = { .name = "endianess_64",
+                                     .t    = NEU_JSON_INT };
 
     ret = neu_parse_param((char *) config, &err_param, 5, &port, &host, &mode,
                           &timeout, &interval);
@@ -189,10 +193,24 @@ 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, &endianess_64);
+    if (ret != 0) {
+        free(err_param);
+        endianess_64.v.val_int = MODBUS_LL;
+    }
+
+    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->endianess_64   = endianess_64.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.json

@@ -1,247 +1,247 @@
 {
-  "client-id": {
-    "name": "Client ID",
-    "name_zh": "客户端 ID",
-    "description": "MQTT client id",
-    "description_zh": "MQTT 客户端 ID",
-    "type": "string",
-    "attribute": "required",
-    "default": "neuron_${random_str}",
-    "valid": {
-      "length": 255
-    }
-  },
-  "qos": {
-    "name": "QoS Level",
-    "name_zh": "QoS 等级",
-    "description": "MQTT QoS level for message delivery",
-    "description_zh": "MQTT 消息传输使用的服务质量等级",
-    "type": "map",
-    "attribute": "optional",
-    "default": 0,
-    "valid": {
-      "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 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": "values-format",
-          "value": 0
-        },
-        {
-          "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/${random_str}/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/${random_str}/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）。应该小于缓存磁盘大小。",
-    "type": "int",
-    "attribute": "required",
-    "condition": {
-      "field": "offline-cache",
-      "value": true
-    },
-    "valid": {
-      "min": 1,
-      "max": 1024
-    }
-  },
-  "cache-disk-size": {
-    "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。",
-    "type": "int",
-    "attribute": "required",
-    "condition": {
-      "field": "offline-cache",
-      "value": true
-    },
-    "valid": {
-      "min": 1,
-      "max": 10240
-    }
-  },
-  "cache-sync-interval": {
-    "name": "Cache Sync Interval (MS)",
-    "name_zh": "缓存消息重传间隔（MS）",
-    "description": "When MQTT connection restores, cached messages will be synchronised to the broker. This controls the interval in milliseconds between each message to synchronise.",
-    "description_zh": "当 MQTT 连接重建后，缓存的消息会被同步到服务器。此配置项控制消息重传时每条消息之间的时间间隔，以毫秒为单位。",
-    "type": "int",
-    "attribute": "required",
-    "condition": {
-      "field": "offline-cache",
-      "value": true
-    },
-    "default": 100,
-    "valid": {
-      "min": 10,
-      "max": 120000
-    }
-  },
-  "host": {
-    "name": "Broker Host",
-    "name_zh": "服务器地址",
-    "description": "MQTT broker host",
-    "description_zh": "MQTT 服务器地址",
-    "attribute": "required",
-    "type": "string",
-    "default": "broker.emqx.io",
-    "valid": {
-      "length": 255
-    }
-  },
-  "port": {
-    "name": "Broker Port",
-    "name_zh": "服务器端口",
-    "description": "MQTT broker port",
-    "description_zh": "MQTT 服务器端口",
-    "attribute": "required",
-    "type": "int",
-    "default": 1883,
-    "valid": {
-      "min": 1024,
-      "max": 65535
-    }
-  },
-  "username": {
-    "name": "Username",
-    "name_zh": "用户名",
-    "description": "MQTT user name",
-    "description_zh": "MQTT 用户名",
-    "attribute": "optional",
-    "type": "string",
-    "default": "",
-    "valid": {
-      "length": 255
-    }
-  },
-  "password": {
-    "name": "Password",
-    "name_zh": "密码",
-    "description": "MQTT user password",
-    "description_zh": "MQTT 用户密码",
-    "attribute": "optional",
-    "type": "string",
-    "default": "",
-    "valid": {
-      "length": 255
-    }
-  },
-  "ssl": {
-    "name": "SSL",
-    "name_zh": "SSL",
-    "description": "Enable SSL connection",
-    "description_zh": "是否启用 SSL 连接",
-    "attribute": "optional",
-    "type": "bool",
-    "default": false,
-    "valid": {}
-  },
-  "ca": {
-    "name": "CA",
-    "name_zh": "CA 证书",
-    "description": "CA certificate which signs the server certificate",
-    "description_zh": "签发服务器证书的 CA 证书",
-    "attribute": "optional",
-    "type": "file",
-    "condition": {
-      "field": "ssl",
-      "value": true
-    },
-    "valid": {
-      "length": 81960
-    }
-  },
-  "cert": {
-    "name": "Client Cert",
-    "name_zh": "客户端证书",
-    "description": "Client x509 certificate when using two way authentication",
-    "description_zh": "使用双向认证时，客户端的 x509 证书",
-    "attribute": "optional",
-    "type": "file",
-    "condition": {
-      "field": "ssl",
-      "value": true
-    },
-    "valid": {
-      "length": 81960
-    }
-  },
-  "key": {
-    "name": "Client Private Key",
-    "name_zh": "客户端私钥",
-    "description": "Client private key when using two way authentication",
-    "description_zh": "使用双向认证时，客户端的私钥",
-    "attribute": "optional",
-    "type": "file",
-    "condition": {
-      "field": "ssl",
-      "value": true
-    },
-    "valid": {
-      "length": 81960
-    }
-  }
-}
+	"client-id": {
+		"name": "Client ID",
+		"name_zh": "客户端 ID",
+		"description": "MQTT client id",
+		"description_zh": "MQTT 客户端 ID",
+		"type": "string",
+		"attribute": "required",
+		"default": "neuron_${random_str}",
+		"valid": {
+			"length": 255
+		}
+	},
+	"qos": {
+		"name": "QoS Level",
+		"name_zh": "QoS 等级",
+		"description": "MQTT QoS level for message delivery",
+		"description_zh": "MQTT 消息传输使用的服务质量等级",
+		"type": "map",
+		"attribute": "optional",
+		"default": 0,
+		"valid": {
+			"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 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": "values-format",
+					"value": 0
+				},
+				{
+					"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/${random_str}/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/${random_str}/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）。应该小于缓存磁盘大小。",
+		"type": "int",
+		"attribute": "required",
+		"condition": {
+			"field": "offline-cache",
+			"value": true
+		},
+		"valid": {
+			"min": 1,
+			"max": 1024
+		}
+	},
+	"cache-disk-size": {
+		"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。",
+		"type": "int",
+		"attribute": "required",
+		"condition": {
+			"field": "offline-cache",
+			"value": true
+		},
+		"valid": {
+			"min": 1,
+			"max": 10240
+		}
+	},
+	"cache-sync-interval": {
+		"name": "Cache Sync Interval (MS)",
+		"name_zh": "缓存消息重传间隔（MS）",
+		"description": "When MQTT connection restores, cached messages will be synchronised to the broker. This controls the interval in milliseconds between each message to synchronise.",
+		"description_zh": "当 MQTT 连接重建后，缓存的消息会被同步到服务器。此配置项控制消息重传时每条消息之间的时间间隔，以毫秒为单位。",
+		"type": "int",
+		"attribute": "required",
+		"condition": {
+			"field": "offline-cache",
+			"value": true
+		},
+		"default": 100,
+		"valid": {
+			"min": 10,
+			"max": 120000
+		}
+	},
+	"host": {
+		"name": "Broker Host",
+		"name_zh": "服务器地址",
+		"description": "MQTT broker host",
+		"description_zh": "MQTT 服务器地址",
+		"attribute": "required",
+		"type": "string",
+		"default": "broker.emqx.io",
+		"valid": {
+			"length": 255
+		}
+	},
+	"port": {
+		"name": "Broker Port",
+		"name_zh": "服务器端口",
+		"description": "MQTT broker port",
+		"description_zh": "MQTT 服务器端口",
+		"attribute": "required",
+		"type": "int",
+		"default": 1883,
+		"valid": {
+			"min": 1,
+			"max": 65535
+		}
+	},
+	"username": {
+		"name": "Username",
+		"name_zh": "用户名",
+		"description": "MQTT user name",
+		"description_zh": "MQTT 用户名",
+		"attribute": "optional",
+		"type": "string",
+		"default": "",
+		"valid": {
+			"length": 255
+		}
+	},
+	"password": {
+		"name": "Password",
+		"name_zh": "密码",
+		"description": "MQTT user password",
+		"description_zh": "MQTT 用户密码",
+		"attribute": "optional",
+		"type": "string",
+		"default": "",
+		"valid": {
+			"length": 255
+		}
+	},
+	"ssl": {
+		"name": "SSL",
+		"name_zh": "SSL",
+		"description": "Enable SSL connection",
+		"description_zh": "是否启用 SSL 连接",
+		"attribute": "optional",
+		"type": "bool",
+		"default": false,
+		"valid": {}
+	},
+	"ca": {
+		"name": "CA",
+		"name_zh": "CA 证书",
+		"description": "CA certificate which signs the server certificate",
+		"description_zh": "签发服务器证书的 CA 证书",
+		"attribute": "optional",
+		"type": "file",
+		"condition": {
+			"field": "ssl",
+			"value": true
+		},
+		"valid": {
+			"length": 81960
+		}
+	},
+	"cert": {
+		"name": "Client Cert",
+		"name_zh": "客户端证书",
+		"description": "Client x509 certificate when using two way authentication",
+		"description_zh": "使用双向认证时，客户端的 x509 证书",
+		"attribute": "optional",
+		"type": "file",
+		"condition": {
+			"field": "ssl",
+			"value": true
+		},
+		"valid": {
+			"length": 81960
+		}
+	},
+	"key": {
+		"name": "Client Private Key",
+		"name_zh": "客户端私钥",
+		"description": "Client private key when using two way authentication",
+		"description_zh": "使用双向认证时，客户端的私钥",
+		"attribute": "optional",
+		"type": "file",
+		"condition": {
+			"field": "ssl",
+			"value": true
+		},
+		"valid": {
+			"length": 81960
+		}
+	}
+}

plugins/mqtt/mqtt_handle.c

@@ -70,7 +70,7 @@ char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
                                   neu_json_encode_read_periodic_resp,
                                   &json_str);
     } else if (MQTT_UPLOAD_FORMAT_TAGS == format) { // tags
-        neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, &header,
+        neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp2, &header,
                                   neu_json_encode_read_periodic_resp,
                                   &json_str);
     } else {

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

@@ -40,7 +40,7 @@ void handle_add_plugin(nng_aio *aio)
 
             header.ctx  = aio;
             header.type = NEU_REQ_ADD_PLUGIN;
-            strcpy(cmd.library, req->library);
+            strncpy(cmd.library, req->library, NEU_PLUGIN_LIBRARY_LEN);
             cmd.schema_file = req->schema_file;
             cmd.so_file     = req->so_file;
             ret             = neu_plugin_op(plugin, header, &cmd);
@@ -64,7 +64,7 @@ void handle_update_plugin(nng_aio *aio)
 
             header.ctx  = aio;
             header.type = NEU_REQ_UPDATE_PLUGIN;
-            strcpy(cmd.library, req->library);
+            strncpy(cmd.library, req->library, NEU_PLUGIN_LIBRARY_LEN);
             cmd.schema_file = req->schema_file;
             cmd.so_file     = req->so_file;
             ret             = neu_plugin_op(plugin, header, &cmd);

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/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

@@ -237,8 +237,7 @@ static void update_im(neu_adapter_t *adapter, const char *group,
     }
     neu_datatag_t *first = utarray_front(tags);
 
-    if (first == NULL ||
-        !neu_tag_attribute_test(first, NEU_ATTRIBUTE_SUBSCRIBE)) {
+    if (first == NULL) {
         utarray_free(tags);
         nlog_debug("update immediately, driver: %s, "
                    "group: %s, tag: %s, type: %s, "
@@ -582,7 +581,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());
 
@@ -748,9 +756,17 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
     case NEU_TYPE_LWORD:
         switch (tag->option.value64.endian) {
         case NEU_DATATAG_ENDIAN_B64:
+        case NEU_DATATAG_ENDIAN_BB64:
             value->value.u64 = neu_ntohll(value->value.u64);
             break;
+        case NEU_DATATAG_ENDIAN_LB64:
+            value->value.u64 = neu_htonlb(value->value.u64);
+            break;
+        case NEU_DATATAG_ENDIAN_BL64:
+            value->value.u64 = neu_htonbl(value->value.u64);
+            break;
         case NEU_DATATAG_ENDIAN_L64:
+        case NEU_DATATAG_ENDIAN_LL64:
         default:
             break;
         }
@@ -851,7 +867,8 @@ int check_value_decimal(neu_type_e write_type, double result, int64_t value_min,
         return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
     } else if (result < value_min || result > value_max) {
         return NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE;
-    } else if (almost_equal(ceil(result), result, EPSILON)) {
+    } else if (almost_equal(ceil(result), result, EPSILON) ||
+               almost_equal(floor(result), result, EPSILON)) {
         return NEU_ERR_SUCCESS;
     } else {
         return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
@@ -901,27 +918,26 @@ int is_value_in_range(neu_type_e tag_type, int64_t value, double value_d,
         case NEU_TYPE_UINT64:
             if (write_type != NEU_TYPE_INT64 && write_type != NEU_TYPE_DOUBLE) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else if (almost_equal(ceil(result), result, EPSILON)) {
+            } else if (almost_equal(ceil(result), result, EPSILON) ||
+                       almost_equal(floor(result), result, EPSILON)) {
                 return NEU_ERR_SUCCESS;
             } else {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
             }
         case NEU_TYPE_FLOAT:
-            if (write_type != NEU_TYPE_DOUBLE) {
+            if (write_type != NEU_TYPE_DOUBLE && write_type != NEU_TYPE_INT64) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else if (result < (int64_t) -FLT_MAX ||
-                       result > (int64_t) FLT_MAX) {
+            }
+            if (result < (int64_t) -FLT_MAX || result > (int64_t) FLT_MAX) {
                 return NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE;
             } else {
                 return NEU_ERR_SUCCESS;
             }
         case NEU_TYPE_DOUBLE:
-            if (write_type != NEU_TYPE_DOUBLE) {
+            if (write_type != NEU_TYPE_DOUBLE && write_type != NEU_TYPE_INT64) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else {
-                return NEU_ERR_SUCCESS;
             }
-            return check_value(write_type, result, INT16_MIN, INT16_MAX);
+            return NEU_ERR_SUCCESS;
         case NEU_TYPE_STRING:
         case NEU_TYPE_BOOL:
         case NEU_TYPE_BIT:
@@ -965,20 +981,19 @@ int is_value_in_range(neu_type_e tag_type, int64_t value, double value_d,
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
             }
         case NEU_TYPE_FLOAT:
-            if (write_type != NEU_TYPE_DOUBLE) {
+            if (write_type != NEU_TYPE_DOUBLE && write_type != NEU_TYPE_INT64) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else if (value < (int64_t) -FLT_MAX ||
-                       value > (int64_t) FLT_MAX) {
+            }
+            if (value < (int64_t) -FLT_MAX || value > (int64_t) FLT_MAX) {
                 return NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE;
             } else {
                 return NEU_ERR_SUCCESS;
             }
         case NEU_TYPE_DOUBLE:
-            if (write_type != NEU_TYPE_DOUBLE) {
+            if (write_type != NEU_TYPE_DOUBLE && write_type != NEU_TYPE_INT64) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else {
-                return NEU_ERR_SUCCESS;
             }
+            return NEU_ERR_SUCCESS;
         case NEU_TYPE_STRING:
             if (write_type != NEU_TYPE_STRING) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
@@ -2171,9 +2186,17 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
         case NEU_TYPE_UINT64:
             switch (tag->option.value64.endian) {
             case NEU_DATATAG_ENDIAN_B64:
+            case NEU_DATATAG_ENDIAN_BB64:
                 value.value.value.u64 = neu_htonll(value.value.value.u64);
                 break;
+            case NEU_DATATAG_ENDIAN_LB64:
+                value.value.value.u64 = neu_htonlb(value.value.value.u64);
+                break;
+            case NEU_DATATAG_ENDIAN_BL64:
+                value.value.value.u64 = neu_htonbl(value.value.value.u64);
+                break;
             case NEU_DATATAG_ENDIAN_L64:
+            case NEU_DATATAG_ENDIAN_LL64:
             default:
                 break;
             }
@@ -2336,9 +2359,17 @@ static void read_group(int64_t timestamp, int64_t timeout,
         case NEU_TYPE_UINT64:
             switch (tag->option.value64.endian) {
             case NEU_DATATAG_ENDIAN_B64:
+            case NEU_DATATAG_ENDIAN_BB64:
                 value.value.value.u64 = neu_htonll(value.value.value.u64);
                 break;
+            case NEU_DATATAG_ENDIAN_LB64:
+                value.value.value.u64 = neu_htonlb(value.value.value.u64);
+                break;
+            case NEU_DATATAG_ENDIAN_BL64:
+                value.value.value.u64 = neu_htonbl(value.value.value.u64);
+                break;
             case NEU_DATATAG_ENDIAN_L64:
+            case NEU_DATATAG_ENDIAN_LL64:
             default:
                 break;
             }
@@ -2447,6 +2478,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) {
@@ -2508,9 +2546,17 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
         case NEU_TYPE_UINT64:
             switch (tag->option.value64.endian) {
             case NEU_DATATAG_ENDIAN_B64:
+            case NEU_DATATAG_ENDIAN_BB64:
                 value.value.value.u64 = neu_htonll(value.value.value.u64);
                 break;
+            case NEU_DATATAG_ENDIAN_LB64:
+                value.value.value.u64 = neu_htonlb(value.value.value.u64);
+                break;
+            case NEU_DATATAG_ENDIAN_BL64:
+                value.value.value.u64 = neu_htonbl(value.value.value.u64);
+                break;
             case NEU_DATATAG_ENDIAN_L64:
+            case NEU_DATATAG_ENDIAN_LL64:
             default:
                 break;
             }

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

@@ -255,21 +255,35 @@ int neu_datatag_parse_addr_option(const neu_datatag_t *      datatag,
     case NEU_TYPE_UINT64: {
         char *op = find_last_character(datatag->address, '#');
 
-        option->value64.endian = NEU_DATATAG_ENDIAN_L64;
+        option->value64.endian     = NEU_DATATAG_ENDIAN_L64;
+        option->value64.is_default = true;
         if (op != NULL) {
-            char e = 0;
-            sscanf(op, "#%c", &e);
+            char e1 = 0;
+            char e2 = 0;
+            int  n  = sscanf(op, "#%c%c", &e1, &e2);
 
-            switch (e) {
-            case 'B':
-                option->value64.endian = NEU_DATATAG_ENDIAN_B64;
-                break;
-            case 'L':
-                option->value64.endian = NEU_DATATAG_ENDIAN_L64;
-                break;
-            default:
-                option->value64.endian = NEU_DATATAG_ENDIAN_L64;
-                break;
+            if (n == 2) {
+                if (e1 == 'B' && e2 == 'B') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_BB64;
+                    option->value64.is_default = false;
+                } else if (e1 == 'B' && e2 == 'L') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_BL64;
+                    option->value64.is_default = false;
+                } else if (e1 == 'L' && e2 == 'L') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_LL64;
+                    option->value64.is_default = false;
+                } else if (e1 == 'L' && e2 == 'B') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_LB64;
+                    option->value64.is_default = false;
+                }
+            } else if (n == 1) {
+                if (e1 == 'B') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_B64;
+                    option->value64.is_default = false;
+                } else if (e1 == 'L') {
+                    option->value64.endian     = NEU_DATATAG_ENDIAN_L64;
+                    option->value64.is_default = false;
+                }
             }
         }
 

src/connection/connection.c

@@ -99,7 +99,7 @@ neu_conn_t *neu_conn_new(neu_conn_param_t *param, void *data,
     conn->disconnected     = disconnected;
     conn->connected        = connected;
 
-    conn->buf_size = 2048;
+    conn->buf_size = 8192;
     conn->buf      = calloc(conn->buf_size, 1);
     conn->offset   = 0;
     conn->stop     = false;
@@ -407,6 +407,40 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
     return ret;
 }
 
+void neu_conn_clear_recv_buffer(neu_conn_t *conn)
+{
+    if (!conn->is_connected) {
+        return;
+    }
+
+    uint8_t temp_buf[256];
+    ssize_t ret;
+
+    switch (conn->param.type) {
+    case NEU_CONN_TCP_CLIENT:
+        do {
+            ret = recv(conn->fd, temp_buf, sizeof(temp_buf), MSG_DONTWAIT);
+
+            if (ret > 0) {
+                continue;
+            } else if (ret == 0) {
+                zlog_info(conn->param.log,
+                          "Connection closed while clearing buffer.");
+                break;
+            } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                break;
+            } else {
+                zlog_error(conn->param.log, "Error clearing buffer: %s",
+                           strerror(errno));
+                break;
+            }
+        } while (ret > 0);
+        break;
+    default:
+        break;
+    }
+}
+
 ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
 {
     ssize_t ret = 0;

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;
 }
@@ -798,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);

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

@@ -66,7 +66,7 @@ struct event_data {
     bool  use;
 };
 
-#define EVENT_SIZE 1400
+#define EVENT_SIZE 2800
 
 struct neu_events {
     int       epoll_fd;

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) {

src/parser/neu_json_rw.c

@@ -291,6 +291,53 @@ int neu_json_encode_read_resp1(void *json_object, void *param)
     return ret;
 }
 
+int neu_json_encode_read_resp2(void *json_object, void *param)
+{
+    int                   ret  = 0;
+    neu_json_read_resp_t *resp = (neu_json_read_resp_t *) param;
+
+    void *                    tag_array = neu_json_array();
+    neu_json_read_resp_tag_t *p_tag     = resp->tags;
+    for (int i = 0; i < resp->n_tag; i++) {
+        neu_json_elem_t tag_elems[2 + NEU_TAG_META_SIZE] = { 0 };
+
+        tag_elems[0].name      = "name";
+        tag_elems[0].t         = NEU_JSON_STR;
+        tag_elems[0].v.val_str = p_tag->name;
+
+        if (p_tag->error != 0) {
+            tag_elems[1].name      = "error";
+            tag_elems[1].t         = NEU_JSON_INT;
+            tag_elems[1].v.val_int = p_tag->error;
+        } else {
+            tag_elems[1].name      = "value";
+            tag_elems[1].t         = p_tag->t;
+            tag_elems[1].v         = p_tag->value;
+            tag_elems[1].precision = p_tag->precision;
+        }
+
+        for (int k = 0; k < p_tag->n_meta; k++) {
+            tag_elems[2 + k].name = p_tag->metas[k].name;
+            tag_elems[2 + k].t    = p_tag->metas[k].t;
+            tag_elems[2 + k].v    = p_tag->metas[k].value;
+        }
+
+        tag_array =
+            neu_json_encode_array(tag_array, tag_elems, 2 + p_tag->n_meta);
+        p_tag++;
+    }
+
+    neu_json_elem_t resp_elems[] = { {
+        .name         = "tags",
+        .t            = NEU_JSON_OBJECT,
+        .v.val_object = tag_array,
+    } };
+    ret = neu_json_encode_field(json_object, resp_elems,
+                                NEU_JSON_ELEM_SIZE(resp_elems));
+
+    return ret;
+}
+
 int neu_json_decode_write_req(char *buf, neu_json_write_req_t **result)
 {
     void *json_obj = neu_json_decode_new(buf);

tests/ft/driver/test_modbus.py

@@ -563,10 +563,6 @@ class TestModbus:
         assert 400 == response.status_code
         assert error.NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH == response.json()['error']
 
-        response = api.write_tag(node=param[0], group='group', tag=hold_float[0]['name'], value=1)
-        assert 400 == response.status_code
-        assert error.NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH == response.json()['error']
-
         response = api.write_tags(node=param[0], group='group', tag_values=[
             {"tag": hold_int16[0]['name'], "value": 32768},
             {"tag": hold_uint16[0]['name'], "value": 2},

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,29 +337,31 @@ 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):
+
         driver = {
             "name": driver_1["name"],
             "plugin": driver_1["plugin"],
             "params": driver_1["params"],
-            "groups": [{"group": "grp", "interval": 1000, "tags": []} for _ in range(1000)]
+            "groups": []
         }
+        for i in range(1025):
+            driver["groups"].append({"group": "grp" + str(i), "interval": 1000, "tags": []})
+
         response = api.put_drivers([driver])
         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 +370,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 +393,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 +424,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 +440,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 +470,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 +510,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 +525,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 +541,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 +556,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 +570,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 +613,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 +636,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.1
+2.9.5