Merge pull request #2312 from fengzeroz/v2.9

version: 2.9.5

.github/workflows/check.yml

@@ -18,7 +18,7 @@ jobs:
           submodules: recursive
 
       - name: Run clang-format style check for C/C++ programs.
-        uses: DoozyX/clang-format-lint-action@v0.12
+        uses: DoozyX/clang-format-lint-action@v0.18
         with:
           source: 'src plugins include tests'
           exclude: 'include/neuron/utils/uthash.h include/neuron/utils/utarray.h include/neuron/utils/utlist.h include/neuron/utils/zlog.h'

.github/workflows/ci.yml

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

include/neuron/connection/mqtt_client.h

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

include/neuron/define.h

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 9
-#define NEU_VERSION_FIX 2
+#define NEU_VERSION_FIX 5
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)

include/neuron/type.h

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

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_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;

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, "
@@ -860,7 +859,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;
@@ -910,7 +910,8 @@ 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;
@@ -2456,6 +2457,13 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
         tag_value.datatag.option      = tag->option;
         memcpy(tag_value.datatag.meta, tag->meta, NEU_TAG_META_LENGTH);
 
+        if (tag->attribute == NEU_ATTRIBUTE_WRITE) {
+            tag_value.value.type         = NEU_TYPE_STRING;
+            tag_value.value.value.str[0] = 0;
+            utarray_push_back(tag_values, &tag_value);
+            continue;
+        }
+
         if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
                                       tag_value.metas,
                                       NEU_TAG_META_SIZE) != 0) {

src/base/group.c

@@ -315,8 +315,8 @@ UT_array *neu_group_query_read_tag_paginate(neu_group_t *group,
     };
 
     pthread_mutex_lock(&group->mtx);
-    array = filter_tags_by_page(group->tags, is_readable_and_match_query, &q,
-                                current_page, page_size, total_count);
+    array = filter_tags_by_page(group->tags, match_query, &q, current_page,
+                                page_size, total_count);
     pthread_mutex_unlock(&group->mtx);
 
     return array;

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;

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

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

tests/ft/neuron/api.py

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

version

@@ -1 +1 @@
-2.9.2
+2.9.5