Merge pull request #2128 from eeff/v2.8-batch

feat(xinao-mqtt): use `dev` key as node and group name

CMakeLists.txt

@@ -171,6 +171,7 @@ file(COPY ${CMAKE_SOURCE_DIR}/default_plugins.json DESTINATION ${CMAKE_BINARY_DI
 
 add_subdirectory(plugins/modbus)
 add_subdirectory(plugins/mqtt)
+add_subdirectory(plugins/mqtt-xinao)
 add_subdirectory(plugins/ekuiper)
 
 add_subdirectory(simulator)

default_plugins.json

@@ -1,6 +1,7 @@
 {
 	"plugins": [
 		"libplugin-mqtt.so",
+		"libplugin-mqtt-xinao.so",
 		"libplugin-ekuiper.so",
 		"libplugin-modbus-tcp.so",
 		"libplugin-modbus-rtu.so"

include/neuron/errcodes.h

@@ -135,6 +135,7 @@ typedef enum {
     NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS  = 3017,
     NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC   = 3018,
     NEU_ERR_PLUGIN_TYPE_NOT_SUPPORT        = 3019,
+    NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_BATCH = 3021,
 
     NEU_ERR_MQTT_FAILURE                        = 4000,
     NEU_ERR_MQTT_NO_CERTFILESET                 = 4001,

include/neuron/msg.h

@@ -118,6 +118,9 @@ typedef enum neu_reqresp_type {
     NEU_REQ_PRGFILE_PROCESS,
     NEU_RESP_PRGFILE_PROCESS,
 
+    NEU_REQ_WRITE_BATCH,
+    NEU_RESP_WRITE_BATCH,
+
 } neu_reqresp_type_e;
 
 static const char *neu_reqresp_type_string_t[] = {
@@ -195,6 +198,9 @@ static const char *neu_reqresp_type_string_t[] = {
     [NEU_REQ_PRGFILE_UPLOAD]   = "NEU_REQ_PRGFILE_UPLOAD",
     [NEU_REQ_PRGFILE_PROCESS]  = "NEU_REQ_PRGFILE_PROCESS",
     [NEU_RESP_PRGFILE_PROCESS] = "NEU_RESP_PRGFILE_PROCESS",
+
+    [NEU_REQ_WRITE_BATCH]  = "NEU_REQ_WRITE_BATCH",
+    [NEU_RESP_WRITE_BATCH] = "NEU_RESP_WRITE_BATCH",
 };
 
 inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type)
@@ -713,6 +719,87 @@ static inline void neu_req_write_gtags_fini(neu_req_write_gtags_t *req)
     free(req->groups);
 }
 
+typedef struct {
+    char *         name;  // name of the tag to write
+    neu_datatag_t *tag;   // actual tag, if NULL, plugin should skip over
+    neu_dvalue_t   value; // value to write
+} neu_write_batch_tag_value_t;
+
+static inline void
+neu_write_batch_tag_value_fini(neu_write_batch_tag_value_t *val)
+{
+    free(val->name);
+    neu_tag_free(val->tag);
+}
+
+typedef struct {
+    char *                       group;
+    neu_err_code_e               error; // if nonzero, plugin should skip over
+    int                          n_tag;
+    neu_write_batch_tag_value_t *tags;
+} neu_write_batch_group_t;
+
+static inline void neu_write_batch_group_fini(neu_write_batch_group_t *grp)
+{
+    for (int i = 0; i < grp->n_tag; ++i) {
+        neu_write_batch_tag_value_fini(&grp->tags[i]);
+    }
+    free(grp->tags);
+    free(grp->group);
+}
+
+static inline void neu_write_batch_group_set_error(neu_write_batch_group_t *grp,
+                                                   int i, neu_err_code_e e)
+{
+    grp->tags[i].value.type      = NEU_TYPE_ERROR;
+    grp->tags[i].value.value.i32 = e;
+}
+
+typedef struct {
+    char *                   driver;
+    neu_err_code_e           error; // if nonzero, should skip over request
+    int                      n_group;
+    neu_write_batch_group_t *groups;
+} neu_write_batch_driver_t;
+
+static inline void neu_write_batch_driver_fini(neu_write_batch_driver_t *drv)
+{
+    for (int i = 0; i < drv->n_group; ++i) {
+        neu_write_batch_group_fini(&drv->groups[i]);
+    }
+    free(drv->groups);
+    free(drv->driver);
+}
+
+// Batch write, supporting multiple driver/group/tags
+// This data structure is used for both request and response messages.
+typedef struct {
+    int                      n_done_; // number of completed driver write
+    int                      n_driver;
+    neu_write_batch_driver_t drivers[];
+} neu_write_batch_t;
+
+static inline void neu_write_batch_free(neu_write_batch_t *batch)
+{
+    if (NULL == batch) {
+        return;
+    }
+    for (int i = 0; i < batch->n_driver; ++i) {
+        neu_write_batch_driver_fini(&batch->drivers[i]);
+    }
+    free(batch);
+}
+
+typedef struct {
+    neu_write_batch_t *batch;
+    int                index_; // index to driver to write
+} neu_reqresp_write_batch_t;
+
+static inline void neu_reqresp_write_batch_fini(neu_reqresp_write_batch_t *req)
+{
+    neu_write_batch_free(req->batch);
+}
+
 typedef struct {
     char *driver;
     char *group;

include/neuron/plugin.h

@@ -109,6 +109,9 @@ typedef struct neu_plugin_intf_funs {
                             neu_datatag_t *tags,
                             int            n_tag); // create tags by API
             int (*del_tags)(neu_plugin_t *plugin, int n_tag);
+
+            int (*write_batch)(neu_plugin_t *plugin, void *req,
+                               neu_write_batch_driver_t *batch);
         } driver;
     };
 

plugins/mqtt-xinao/CMakeLists.txt

@@ -0,0 +1,22 @@
+cmake_minimum_required(VERSION 3.12)
+
+project(plugin-mqtt-xinao)
+
+set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins")
+
+file(COPY ${CMAKE_SOURCE_DIR}/plugins/mqtt-xinao/mqtt-xinao.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
+
+add_library(${PROJECT_NAME} SHARED
+  mqtt_config.c
+  mqtt_handle.c
+  mqtt_plugin.c
+  xinao.c
+)
+
+target_include_directories(${PROJECT_NAME} PRIVATE 
+  ${CMAKE_SOURCE_DIR}/include/neuron
+  ${CMAKE_SOURCE_DIR}/plugins/mqtt-xinao
+)
+
+target_link_libraries(${PROJECT_NAME} neuron-base)
+target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT})

plugins/mqtt-xinao/mqtt-xinao.json

@@ -0,0 +1,246 @@
+{
+  "client-id": {
+    "name": "Client ID",
+    "name_zh": "客户端 ID",
+    "description": "MQTT client id",
+    "description_zh": "MQTT 客户端 ID",
+    "type": "string",
+    "attribute": "required",
+    "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": "/edge/algorithmwrite",
+    "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": "/cmd/algorithmwrite/cack",
+    "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
+    }
+  }
+}

plugins/mqtt-xinao/mqtt_config.c

@@ -0,0 +1,431 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#include "utils/asprintf.h"
+#include "json/json.h"
+#include "json/neu_json_param.h"
+
+#include "mqtt_config.h"
+#include "mqtt_plugin.h"
+
+#define MB 1000000
+
+static inline int decode_b64_param(neu_plugin_t *plugin, neu_json_elem_t *el)
+{
+    int   len = 0;
+    char *s   = (char *) neu_decode64(&len, el->v.val_str);
+
+    if (NULL == s) {
+        plog_error(plugin, "setting %s invalid base64", el->name);
+        return -1;
+    }
+
+    if (0 == len) {
+        plog_error(plugin, "setting empty %s", el->name);
+        free(s);
+        return -1;
+    }
+
+    free(el->v.val_str);
+    el->v.val_str = s;
+    return 0;
+}
+
+static inline int parse_b64_param(neu_plugin_t *plugin, const char *setting,
+                                  neu_json_elem_t *el)
+{
+    if (0 != neu_parse_param(setting, NULL, 1, el)) {
+        plog_error(plugin, "setting no %s", el->name);
+        return -1;
+    }
+
+    if (0 != decode_b64_param(plugin, el)) {
+        free(el->v.val_str);
+        el->v.val_str = NULL;
+        return -1;
+    }
+
+    return 0;
+}
+
+static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
+                                   neu_json_elem_t *ssl, neu_json_elem_t *ca,
+                                   neu_json_elem_t *cert, neu_json_elem_t *key,
+                                   neu_json_elem_t *keypass)
+{
+    // ssl, optional
+    int ret = neu_parse_param(setting, NULL, 1, ssl);
+    if (0 != ret) {
+        plog_notice(plugin, "setting no ssl");
+        return 0;
+    }
+
+    if (false == ssl->v.val_bool) {
+        plog_notice(plugin, "setting ssl disabled");
+        return 0;
+    }
+
+    // ca, optional
+    ret = neu_parse_param(setting, NULL, 1, ca);
+    if (0 != ret) {
+        plog_notice(plugin, "setting no ca");
+        return 0;
+    }
+
+    if (0 != decode_b64_param(plugin, ca)) {
+        return -1;
+    }
+
+    // cert, optional
+    ret = neu_parse_param(setting, NULL, 1, cert);
+    if (0 != ret) {
+        plog_notice(plugin, "setting no cert");
+        return 0;
+    }
+
+    if (0 != decode_b64_param(plugin, cert)) {
+        return -1;
+    }
+
+    // key, required if cert enable
+    ret = parse_b64_param(plugin, setting, key);
+    if (0 != ret) {
+        return -1;
+    }
+
+    // keypass, optional
+    // keep it here for backward compatibility with version 2.5
+    ret = neu_parse_param(setting, NULL, 1, keypass);
+    if (0 != ret) {
+        plog_notice(plugin, "setting no keypass");
+    } else if (0 == strlen(keypass->v.val_str)) {
+        plog_error(plugin, "setting invalid keypass");
+        return -1;
+    }
+
+    return 0;
+}
+
+static int parse_cache_params(neu_plugin_t *plugin, const char *setting,
+                              neu_json_elem_t *offline_cache,
+                              neu_json_elem_t *cache_mem_size,
+                              neu_json_elem_t *cache_disk_size,
+                              neu_json_elem_t *cache_sync_interval)
+{
+    int   ret          = 0;
+    char *err_param    = NULL;
+    bool  flag_present = true; // whether setting has `offline-cache`
+
+    // offline-cache flag, optional
+    ret = neu_parse_param(setting, NULL, 1, offline_cache);
+    if (0 != ret) {
+        plog_notice(plugin, "setting no offline cache flag");
+        flag_present = false; // `offline-cache` not present in setting
+    }
+
+    if (flag_present && !offline_cache->v.val_bool) {
+        // cache explicitly disabled in setting
+        cache_mem_size->v.val_int      = 0;
+        cache_disk_size->v.val_int     = 0;
+        cache_sync_interval->v.val_int = NEU_MQTT_CACHE_SYNC_INTERVAL_DEFAULT;
+        return 0;
+    }
+
+    // we are here because one of the following:
+    //    1. `offline-cache` presents to be true in setting
+    //    2. no `offline-cache` in setting
+    // we require `cache_mem_size` and `cache_disk_size` to present in both
+    // cases, especially in case 2 for backward compatibility.
+
+    ret = neu_parse_param(setting, &err_param, 2, cache_mem_size,
+                          cache_disk_size);
+    if (0 != ret) {
+        plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
+        free(err_param);
+        return -1;
+    }
+
+    if (cache_mem_size->v.val_int > 1024 ||
+        (flag_present && cache_mem_size->v.val_int < 1)) {
+        plog_error(plugin, "setting invalid cache memory size: %" PRIi64,
+                   cache_mem_size->v.val_int);
+        return -1;
+    }
+
+    if (cache_disk_size->v.val_int > 10240 ||
+        (flag_present && cache_disk_size->v.val_int < 1)) {
+        plog_error(plugin, "setting invalid cache disk size: %" PRIi64,
+                   cache_disk_size->v.val_int);
+        return -1;
+    }
+
+    if (cache_mem_size->v.val_int > cache_disk_size->v.val_int) {
+        plog_error(plugin,
+                   "setting cache memory size %" PRIi64
+                   " larger than cache disk size %" PRIi64,
+                   cache_mem_size->v.val_int, cache_disk_size->v.val_int);
+        return -1;
+    }
+
+    if (0 == cache_mem_size->v.val_int && 0 != cache_disk_size->v.val_int) {
+        plog_error(plugin,
+                   "setting cache disk size %" PRIi64 " without memory cache",
+                   cache_disk_size->v.val_int);
+        return -1;
+    }
+
+    if (!flag_present) {
+        offline_cache->v.val_bool = cache_mem_size->v.val_int > 0;
+    }
+
+    // cache-sync-interval, optional for backward compatibility
+    ret = neu_parse_param(setting, NULL, 1, cache_sync_interval);
+    if (0 == ret) {
+        if (cache_sync_interval->v.val_int < NEU_MQTT_CACHE_SYNC_INTERVAL_MIN ||
+            NEU_MQTT_CACHE_SYNC_INTERVAL_MAX < cache_sync_interval->v.val_int) {
+            plog_error(plugin, "setting invalid cache sync interval: %" PRIi64,
+                       cache_sync_interval->v.val_int);
+            return -1;
+        }
+    } else {
+        plog_notice(plugin, "setting no cache sync interval");
+        cache_sync_interval->v.val_int = NEU_MQTT_CACHE_SYNC_INTERVAL_DEFAULT;
+    }
+
+    return 0;
+}
+
+int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
+                      mqtt_config_t *config)
+{
+    int         ret         = 0;
+    char *      err_param   = NULL;
+    const char *placeholder = "********";
+
+    neu_json_elem_t client_id = { .name = "client-id", .t = NEU_JSON_STR };
+    neu_json_elem_t qos       = {
+        .name      = "qos",
+        .t         = NEU_JSON_INT,
+        .v.val_int = NEU_MQTT_QOS0,               // default to QoS0
+        .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
+    };
+    neu_json_elem_t format          = { .name = "format", .t = NEU_JSON_INT };
+    neu_json_elem_t write_req_topic = {
+        .name      = "write-req-topic",
+        .t         = NEU_JSON_STR,
+        .v.val_str = NULL,
+        .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
+    };
+    neu_json_elem_t write_resp_topic = {
+        .name      = "write-resp-topic",
+        .t         = NEU_JSON_STR,
+        .v.val_str = NULL,
+        .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
+    };
+    neu_json_elem_t offline_cache       = { .name = "offline-cache",
+                                      .t    = NEU_JSON_BOOL };
+    neu_json_elem_t cache_mem_size      = { .name = "cache-mem-size",
+                                       .t    = NEU_JSON_INT };
+    neu_json_elem_t cache_disk_size     = { .name = "cache-disk-size",
+                                        .t    = NEU_JSON_INT };
+    neu_json_elem_t cache_sync_interval = { .name = "cache-sync-interval",
+                                            .t    = NEU_JSON_INT };
+    neu_json_elem_t host                = { .name = "host", .t = NEU_JSON_STR };
+    neu_json_elem_t port                = { .name = "port", .t = NEU_JSON_INT };
+    neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR };
+    neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR };
+    neu_json_elem_t ssl      = { .name = "ssl", .t = NEU_JSON_BOOL };
+    neu_json_elem_t ca       = { .name = "ca", .t = NEU_JSON_STR };
+    neu_json_elem_t cert     = { .name = "cert", .t = NEU_JSON_STR };
+    neu_json_elem_t key      = { .name = "key", .t = NEU_JSON_STR };
+    neu_json_elem_t keypass  = { .name = "keypass", .t = NEU_JSON_STR };
+
+    if (NULL == setting || NULL == config) {
+        plog_error(plugin, "invalid argument, null pointer");
+        return -1;
+    }
+
+    ret = neu_parse_param(setting, &err_param, 7, &client_id, &qos, &format,
+                          &write_req_topic, &write_resp_topic, &host, &port);
+    if (0 != ret) {
+        plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
+        goto error;
+    }
+
+    // client-id, required
+    if (0 == strlen(client_id.v.val_str)) {
+        plog_error(plugin, "setting empty client-id");
+        goto error;
+    }
+
+    // qos, optional, default to QoS0
+    if (qos.v.val_int < NEU_MQTT_QOS0 || NEU_MQTT_QOS2 < qos.v.val_int) {
+        plog_error(plugin, "setting invalid qos: %" PRIi64, qos.v.val_int);
+        goto error;
+    }
+
+    // format, required
+    if (MQTT_UPLOAD_FORMAT_VALUES != format.v.val_int &&
+        MQTT_UPLOAD_FORMAT_TAGS != format.v.val_int) {
+        plog_error(plugin, "setting invalid format: %" PRIi64,
+                   format.v.val_int);
+        goto error;
+    }
+
+    // write request topic
+    if (NULL == write_req_topic.v.val_str &&
+        0 > neu_asprintf(&write_req_topic.v.val_str, "/neuron/%s/write/req",
+                         plugin->common.name)) {
+        plog_error(plugin, "setting write request topic error");
+        goto error;
+    }
+
+    // write response topic
+    if (NULL == write_resp_topic.v.val_str &&
+        0 > neu_asprintf(&write_resp_topic.v.val_str, "/neuron/%s/write/resp",
+                         plugin->common.name)) {
+        plog_error(plugin, "setting write response topic error");
+        goto error;
+    }
+
+    // offline cache
+    ret = parse_cache_params(plugin, setting, &offline_cache, &cache_mem_size,
+                             &cache_disk_size, &cache_sync_interval);
+    if (0 != ret) {
+        goto error;
+    }
+
+    // host, required
+    if (0 == strlen(host.v.val_str)) {
+        plog_error(plugin, "setting invalid host: `%s`", host.v.val_str);
+        goto error;
+    }
+
+    // port, required
+    if (0 == port.v.val_int || port.v.val_int > 65535) {
+        plog_error(plugin, "setting invalid port: %" PRIi64, port.v.val_int);
+        goto error;
+    }
+
+    // username, optional
+    ret = neu_parse_param(setting, NULL, 1, &username);
+    if (0 != ret) {
+        plog_notice(plugin, "setting no username");
+    }
+
+    // password, optional
+    ret = neu_parse_param(setting, NULL, 1, &password);
+    if (0 != ret) {
+        plog_notice(plugin, "setting no password");
+    }
+
+    ret = parse_ssl_params(plugin, setting, &ssl, &ca, &cert, &key, &keypass);
+    if (0 != ret) {
+        goto error;
+    }
+
+    config->client_id           = client_id.v.val_str;
+    config->qos                 = qos.v.val_int;
+    config->format              = format.v.val_int;
+    config->write_req_topic     = write_req_topic.v.val_str;
+    config->write_resp_topic    = write_resp_topic.v.val_str;
+    config->cache               = offline_cache.v.val_bool;
+    config->cache_mem_size      = cache_mem_size.v.val_int * MB;
+    config->cache_disk_size     = cache_disk_size.v.val_int * MB;
+    config->cache_sync_interval = cache_sync_interval.v.val_int;
+    config->host                = host.v.val_str;
+    config->port                = port.v.val_int;
+    config->username            = username.v.val_str;
+    config->password            = password.v.val_str;
+    config->ssl                 = ssl.v.val_bool;
+    config->ca                  = ca.v.val_str;
+    config->cert                = cert.v.val_str;
+    config->key                 = key.v.val_str;
+    config->keypass             = keypass.v.val_str;
+
+    plog_notice(plugin, "config client-id       : %s", config->client_id);
+    plog_notice(plugin, "config qos             : %d", config->qos);
+    plog_notice(plugin, "config format          : %s",
+                mqtt_upload_format_str(config->format));
+    plog_notice(plugin, "config write-req-topic : %s", config->write_req_topic);
+    plog_notice(plugin, "config write-resp-topic: %s",
+                config->write_resp_topic);
+    plog_notice(plugin, "config cache           : %zu", config->cache);
+    plog_notice(plugin, "config cache-mem-size  : %zu", config->cache_mem_size);
+    plog_notice(plugin, "config cache-disk-size : %zu",
+                config->cache_disk_size);
+    plog_notice(plugin, "config cache-sync-interval : %zu",
+                config->cache_sync_interval);
+    plog_notice(plugin, "config host            : %s", config->host);
+    plog_notice(plugin, "config port            : %" PRIu16, config->port);
+
+    if (config->username) {
+        plog_notice(plugin, "config username        : %s", config->username);
+    }
+    if (config->password) {
+        plog_notice(plugin, "config password        : %s",
+                    0 == strlen(config->password) ? "" : placeholder);
+    }
+
+    plog_notice(plugin, "config ssl             : %d", config->ssl);
+    if (config->ca) {
+        plog_notice(plugin, "config ca              : %s", placeholder);
+    }
+    if (config->cert) {
+        plog_notice(plugin, "config cert            : %s", placeholder);
+    }
+    if (config->key) {
+        plog_notice(plugin, "config key             : %s", placeholder);
+    }
+    if (config->keypass) {
+        plog_notice(plugin, "config keypass         : %s", placeholder);
+    }
+
+    return 0;
+
+error:
+    free(err_param);
+    free(client_id.v.val_str);
+    free(write_req_topic.v.val_str);
+    free(write_resp_topic.v.val_str);
+    free(host.v.val_str);
+    free(username.v.val_str);
+    free(password.v.val_str);
+    free(ca.v.val_str);
+    free(cert.v.val_str);
+    free(key.v.val_str);
+    free(keypass.v.val_str);
+    return -1;
+}
+
+void mqtt_config_fini(mqtt_config_t *config)
+{
+    free(config->client_id);
+    free(config->write_req_topic);
+    free(config->write_resp_topic);
+    free(config->host);
+    free(config->username);
+    free(config->password);
+    free(config->ca);
+    free(config->cert);
+    free(config->key);
+    free(config->keypass);
+
+    memset(config, 0, sizeof(*config));
+}

plugins/mqtt-xinao/mqtt_config.h

@@ -0,0 +1,81 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#ifndef NEURON_PLUGIN_MQTT_CONFIG_H
+#define NEURON_PLUGIN_MQTT_CONFIG_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stdint.h>
+#include <stdlib.h>
+
+#include "connection/mqtt_client.h"
+#include "plugin.h"
+
+typedef enum {
+    MQTT_UPLOAD_FORMAT_VALUES = 0,
+    MQTT_UPLOAD_FORMAT_TAGS   = 1,
+} mqtt_upload_format_e;
+
+static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
+{
+    switch (f) {
+    case MQTT_UPLOAD_FORMAT_VALUES:
+        return "format-values";
+    case MQTT_UPLOAD_FORMAT_TAGS:
+        return "format-tags";
+    default:
+        return NULL;
+    }
+}
+
+typedef struct {
+    char *               client_id;           // client id
+    neu_mqtt_qos_e       qos;                 // message QoS
+    mqtt_upload_format_e format;              // upload format
+    char *               write_req_topic;     // write request topic
+    char *               write_resp_topic;    // write response topic
+    size_t               cache;               // cache enable flag
+    size_t               cache_mem_size;      // cache memory size in bytes
+    size_t               cache_disk_size;     // cache disk size in bytes
+    size_t               cache_sync_interval; // cache sync interval
+    char *               host;                // broker host
+    uint16_t             port;                // broker port
+    char *               username;            // user name
+    char *               password;            // user password
+    bool                 ssl;                 // ssl flag
+    char *               ca;                  // CA
+    char *               cert;                // client cert
+    char *               key;                 // client key
+    char *               keypass;             // client key password
+                                              // remove in 2.6, keep it here
+                                              // for backward compatibility
+} mqtt_config_t;
+
+int  mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
+                       mqtt_config_t *config);
+void mqtt_config_fini(mqtt_config_t *config);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

plugins/mqtt-xinao/mqtt_handle.c

@@ -0,0 +1,516 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#include "connection/mqtt_client.h"
+#include "errcodes.h"
+#include "msg.h"
+#include "utils/asprintf.h"
+#include "version.h"
+#include "json/neu_json_mqtt.h"
+#include "json/neu_json_rw.h"
+
+#include "mqtt_handle.h"
+#include "mqtt_plugin.h"
+
+static int tag_values_to_json(UT_array *tags, neu_json_read_resp_t *json)
+{
+    int index = 0;
+
+    if (0 == utarray_len(tags)) {
+        return 0;
+    }
+
+    json->n_tag = utarray_len(tags);
+    json->tags  = (neu_json_read_resp_tag_t *) calloc(
+        json->n_tag, sizeof(neu_json_read_resp_tag_t));
+    if (NULL == json->tags) {
+        return -1;
+    }
+
+    utarray_foreach(tags, neu_resp_tag_value_meta_t *, tag_value)
+    {
+        neu_tag_value_to_json(tag_value, &json->tags[index]);
+        index += 1;
+    }
+
+    return 0;
+}
+
+static char *generate_upload_json(neu_plugin_t *            plugin,
+                                  neu_reqresp_trans_data_t *data,
+                                  mqtt_upload_format_e      format)
+{
+    char *                   json_str = NULL;
+    neu_json_read_periodic_t header   = { .group     = (char *) data->group,
+                                        .node      = (char *) data->driver,
+                                        .timestamp = global_timestamp };
+    neu_json_read_resp_t     json     = { 0 };
+
+    if (0 != tag_values_to_json(data->tags, &json)) {
+        plog_error(plugin, "tag_values_to_json fail");
+        return NULL;
+    }
+
+    if (MQTT_UPLOAD_FORMAT_VALUES == format) { // values
+        neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp1, &header,
+                                  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_read_periodic_resp,
+                                  &json_str);
+    } else {
+        plog_warn(plugin, "invalid upload format: %d", format);
+    }
+
+    for (int i = 0; i < json.n_tag; i++) {
+        if (json.tags[i].n_meta > 0) {
+            free(json.tags[i].metas);
+        }
+    }
+
+    if (json.tags) {
+        free(json.tags);
+    }
+    return json_str;
+}
+
+static char *generate_read_resp_json(neu_plugin_t *         plugin,
+                                     neu_json_mqtt_t *      mqtt,
+                                     neu_resp_read_group_t *data)
+{
+    // neu_resp_tag_value_meta_t *tags     = data->tags;
+    // uint16_t                   len      = data->n_tag;
+    char *               json_str = NULL;
+    neu_json_read_resp_t json     = { 0 };
+
+    if (0 != tag_values_to_json(data->tags, &json)) {
+        plog_error(plugin, "tag_values_to_json fail");
+        return NULL;
+    }
+
+    neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, mqtt,
+                              neu_json_encode_mqtt_resp, &json_str);
+
+    if (json.tags) {
+        free(json.tags);
+    }
+    return json_str;
+}
+
+static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
+                                neu_json_read_req_t *req)
+{
+    plog_notice(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid,
+                req->group, req->node);
+
+    neu_reqresp_head_t header = { 0 };
+    header.ctx                = mqtt;
+    header.type               = NEU_REQ_READ_GROUP;
+    neu_req_read_group_t cmd  = { 0 };
+    cmd.driver                = req->node;
+    cmd.group                 = req->group;
+    cmd.sync                  = req->sync;
+    req->node                 = NULL; // ownership moved
+    req->group                = NULL; // ownership moved
+    if (0 != neu_plugin_op(plugin, header, &cmd)) {
+        neu_req_read_group_fini(&cmd);
+        plog_error(plugin, "neu_plugin_op(NEU_REQ_READ_GROUP) fail");
+        return -1;
+    }
+
+    return 0;
+}
+
+static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic,
+                       uint8_t *payload, uint32_t len, void *data)
+{
+    (void) qos;
+    (void) topic;
+    (void) len;
+
+    neu_plugin_t *plugin = data;
+
+    if (0 == errcode) {
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, len, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, len, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, len, NULL);
+    } else {
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
+                                 NULL);
+    }
+
+    free(payload);
+}
+
+static inline int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
+                          char *payload, size_t payload_len)
+{
+
+    int rv =
+        neu_mqtt_client_publish(plugin->client, qos, topic, (uint8_t *) payload,
+                                (uint32_t) payload_len, plugin, publish_cb);
+    if (0 != rv) {
+        plog_error(plugin, "pub [%s, QoS%d] fail", topic, qos);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
+                                 NULL);
+        free(payload);
+        rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
+    }
+
+    return rv;
+}
+
+void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
+                      const uint8_t *payload, uint32_t len, void *data)
+{
+    neu_plugin_t *plugin = data;
+
+    (void) qos;
+    (void) topic;
+
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL);
+
+    xinao_ctx_t *ctx = xinao_ctx_from_json_buf(plugin, (char *) payload, len);
+    if (NULL == ctx) {
+        return;
+    }
+
+    neu_reqresp_head_t header = {
+        .ctx  = ctx,
+        .type = ctx->type,
+    };
+
+    void *cmd;
+    if (NEU_REQ_WRITE_TAG == ctx->type) {
+        cmd = &((xinao_wtag_ctx_t *) ctx)->wtag;
+    } else if (NEU_REQ_WRITE_BATCH == ctx->type) {
+        cmd = &((xinao_batch_ctx_t *) ctx)->batch;
+    } else {
+        plog_error(plugin, "logic error, req ctx bad type:%s",
+                   neu_reqresp_type_string(ctx->type));
+        return;
+    }
+
+    if (0 != neu_plugin_op(plugin, header, cmd)) {
+        plog_error(plugin, "neu_plugin_op(%s) fail",
+                   neu_reqresp_type_string(ctx->type));
+        xinao_ctx_free(ctx);
+        return;
+    }
+
+    if (NEU_REQ_WRITE_TAG == ctx->type) {
+        xinao_wtag_ctx_t *wtag_ctx = (xinao_wtag_ctx_t *) ctx;
+        wtag_ctx->wtag.driver      = NULL; // ownership moved
+        wtag_ctx->wtag.group       = NULL; // ownership moved
+        wtag_ctx->wtag.tag         = NULL; // ownership moved
+    }
+}
+
+int handle_write_resp(neu_plugin_t *plugin, xinao_ctx_t *ctx, void *data)
+{
+    int   rv       = 0;
+    char *json_str = NULL;
+
+    if (NULL == plugin->client) {
+        rv = NEU_ERR_MQTT_IS_NULL;
+        goto end;
+    }
+
+    if (0 == plugin->config.cache &&
+        !neu_mqtt_client_is_connected(plugin->client)) {
+        // cache disable and we are disconnected
+        rv = NEU_ERR_MQTT_FAILURE;
+        goto end;
+    }
+
+    if (NEU_REQ_WRITE_TAG == ctx->type) {
+        json_str = xinao_ctx_gen_wtag_resp_json((xinao_wtag_ctx_t *) ctx, data);
+    } else if (NEU_REQ_WRITE_BATCH == ctx->type) {
+        json_str = xinao_ctx_gen_batch_resp_json((xinao_batch_ctx_t *) ctx);
+    } else {
+        plog_error(plugin, "logic error, resp ctx bad type:%s",
+                   neu_reqresp_type_string(ctx->type));
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    if (NULL == json_str) {
+        plog_error(plugin, "generate %s resp json fail",
+                   neu_reqresp_type_string(ctx->type));
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    char *         topic = plugin->config.write_resp_topic;
+    neu_mqtt_qos_e qos   = plugin->config.qos;
+    rv       = publish(plugin, qos, topic, json_str, strlen(json_str));
+    json_str = NULL;
+
+end:
+    xinao_ctx_free(ctx);
+    return rv;
+}
+
+void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
+                     const uint8_t *payload, uint32_t len, void *data)
+{
+    int                  rv     = 0;
+    neu_plugin_t *       plugin = data;
+    neu_json_read_req_t *req    = NULL;
+
+    (void) qos;
+    (void) topic;
+
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL);
+
+    char *json_str = malloc(len + 1);
+    if (NULL == json_str) {
+        return;
+    }
+
+    memcpy(json_str, payload, len);
+    json_str[len] = '\0';
+
+    neu_json_mqtt_t *mqtt = NULL;
+    rv                    = neu_json_decode_mqtt_req(json_str, &mqtt);
+    if (0 != rv) {
+        plog_error(plugin, "neu_json_decode_mqtt_req failed");
+        free(json_str);
+        return;
+    }
+
+    rv = neu_json_decode_read_req(json_str, &req);
+    if (0 != rv) {
+        plog_error(plugin, "neu_json_decode_read_req fail");
+        neu_json_decode_mqtt_req_free(mqtt);
+        free(json_str);
+        return;
+    }
+
+    rv = send_read_req(plugin, mqtt, req);
+    if (0 != rv) {
+        neu_json_decode_mqtt_req_free(mqtt);
+    }
+
+    neu_json_decode_read_req_free(req);
+    free(json_str);
+}
+
+int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
+                         neu_resp_read_group_t *data)
+{
+    int   rv       = 0;
+    char *json_str = NULL;
+
+    if (NULL == plugin->client) {
+        rv = NEU_ERR_MQTT_IS_NULL;
+        goto end;
+    }
+
+    if (0 == plugin->config.cache &&
+        !neu_mqtt_client_is_connected(plugin->client)) {
+        // cache disable and we are disconnected
+        rv = NEU_ERR_MQTT_FAILURE;
+        goto end;
+    }
+
+    json_str = generate_read_resp_json(plugin, mqtt_json, data);
+    if (NULL == json_str) {
+        plog_error(plugin, "generate read resp json fail");
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    char *         topic = plugin->read_resp_topic;
+    neu_mqtt_qos_e qos   = plugin->config.qos;
+    rv       = publish(plugin, qos, topic, json_str, strlen(json_str));
+    json_str = NULL;
+
+end:
+    neu_json_decode_mqtt_req_free(mqtt_json);
+    return rv;
+}
+
+int handle_trans_data(neu_plugin_t *            plugin,
+                      neu_reqresp_trans_data_t *trans_data)
+{
+    int rv = 0;
+
+    if (NULL == plugin->client) {
+        return NEU_ERR_MQTT_IS_NULL;
+    }
+
+    if (0 == plugin->config.cache &&
+        !neu_mqtt_client_is_connected(plugin->client)) {
+        // cache disable and we are disconnected
+        return NEU_ERR_MQTT_FAILURE;
+    }
+
+    const route_entry_t *route = route_tbl_get(
+        &plugin->route_tbl, trans_data->driver, trans_data->group);
+    if (NULL == route) {
+        plog_error(plugin, "no route for driver:%s group:%s",
+                   trans_data->driver, trans_data->group);
+        return NEU_ERR_GROUP_NOT_SUBSCRIBE;
+    }
+
+    char *json_str =
+        generate_upload_json(plugin, trans_data, plugin->config.format);
+    if (NULL == json_str) {
+        plog_error(plugin, "generate upload json fail");
+        return NEU_ERR_EINTERNAL;
+    }
+
+    char *         topic = route->topic;
+    neu_mqtt_qos_e qos   = plugin->config.qos;
+    rv       = publish(plugin, qos, topic, json_str, strlen(json_str));
+    json_str = NULL;
+
+    return rv;
+}
+
+static inline char *default_upload_topic(neu_req_subscribe_t *info)
+{
+    char *t = NULL;
+    neu_asprintf(&t, "/neuron/%s/%s/%s", info->app, info->driver, info->group);
+    return t;
+}
+
+int handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info)
+{
+    int rv = 0;
+
+    neu_json_elem_t topic = { .name = "topic", .t = NEU_JSON_STR };
+    if (NULL == sub_info->params) {
+        // no parameters, try default topic
+        topic.v.val_str = default_upload_topic(sub_info);
+        if (NULL == topic.v.val_str) {
+            rv = NEU_ERR_EINTERNAL;
+            goto end;
+        }
+    } else if (0 != neu_parse_param(sub_info->params, NULL, 1, &topic)) {
+        plog_error(plugin, "parse `%s` for topic fail", sub_info->params);
+        rv = NEU_ERR_GROUP_PARAMETER_INVALID;
+        goto end;
+    }
+
+    rv = route_tbl_add_new(&plugin->route_tbl, sub_info->driver,
+                           sub_info->group, topic.v.val_str);
+    // topic.v.val_str ownership moved
+    if (0 != rv) {
+        plog_error(plugin, "route driver:%s group:%s fail, `%s`",
+                   sub_info->driver, sub_info->group, sub_info->params);
+        goto end;
+    }
+
+    plog_notice(plugin, "route driver:%s group:%s to topic:%s",
+                sub_info->driver, sub_info->group, topic.v.val_str);
+
+end:
+    free(sub_info->params);
+    return rv;
+}
+
+int handle_update_subscribe(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info)
+{
+    int rv = 0;
+
+    if (NULL == sub_info->params) {
+        rv = NEU_ERR_GROUP_PARAMETER_INVALID;
+        goto end;
+    }
+
+    neu_json_elem_t topic = { .name = "topic", .t = NEU_JSON_STR };
+    if (0 != neu_parse_param(sub_info->params, NULL, 1, &topic)) {
+        plog_error(plugin, "parse `%s` for topic fail", sub_info->params);
+        rv = NEU_ERR_GROUP_PARAMETER_INVALID;
+        goto end;
+    }
+
+    rv = route_tbl_update(&plugin->route_tbl, sub_info->driver, sub_info->group,
+                          topic.v.val_str);
+    // topic.v.val_str ownership moved
+    if (0 != rv) {
+        plog_error(plugin, "route driver:%s group:%s fail, `%s`",
+                   sub_info->driver, sub_info->group, sub_info->params);
+        goto end;
+    }
+
+    plog_notice(plugin, "route driver:%s group:%s to topic:%s",
+                sub_info->driver, sub_info->group, topic.v.val_str);
+
+end:
+    free(sub_info->params);
+    return rv;
+}
+
+int handle_unsubscribe_group(neu_plugin_t *         plugin,
+                             neu_req_unsubscribe_t *unsub_info)
+{
+    route_tbl_del(&plugin->route_tbl, unsub_info->driver, unsub_info->group);
+    plog_notice(plugin, "del route driver:%s group:%s", unsub_info->driver,
+                unsub_info->group);
+    return 0;
+}
+
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req)
+{
+    route_tbl_del(&plugin->route_tbl, req->driver, req->group);
+    plog_notice(plugin, "del route driver:%s group:%s", req->driver,
+                req->group);
+    return 0;
+}
+
+int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req)
+{
+    route_tbl_update_group(&plugin->route_tbl, req->driver, req->group,
+                           req->new_name);
+    plog_notice(plugin, "update route driver:%s group:%s to %s", req->driver,
+                req->group, req->new_name);
+    return 0;
+}
+
+int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req)
+{
+    route_tbl_update_driver(&plugin->route_tbl, req->node, req->new_name);
+    plog_notice(plugin, "update route driver:%s to %s", req->node,
+                req->new_name);
+    return 0;
+}
+
+int handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req)
+{
+    route_tbl_del_driver(&plugin->route_tbl, req->node);
+    plog_notice(plugin, "delete route driver:%s", req->node);
+    return 0;
+}

plugins/mqtt-xinao/mqtt_handle.h

@@ -0,0 +1,62 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#ifndef NEURON_PLUGIN_MQTT_HANDLE_H
+#define NEURON_PLUGIN_MQTT_HANDLE_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "connection/mqtt_client.h"
+#include "neuron.h"
+
+#include "xinao.h"
+
+void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
+                      const uint8_t *payload, uint32_t len, void *data);
+
+int handle_write_resp(neu_plugin_t *plugin, xinao_ctx_t *ctx, void *data);
+
+void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
+                     const uint8_t *payload, uint32_t len, void *data);
+
+int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
+                         neu_resp_read_group_t *data);
+
+int handle_trans_data(neu_plugin_t *            plugin,
+                      neu_reqresp_trans_data_t *trans_data);
+
+int handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info);
+int handle_update_subscribe(neu_plugin_t *       plugin,
+                            neu_req_subscribe_t *sub_info);
+int handle_unsubscribe_group(neu_plugin_t *         plugin,
+                             neu_req_unsubscribe_t *unsub_info);
+
+int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req);
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *cmd);
+
+int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req);
+int handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

plugins/mqtt-xinao/mqtt_plugin.c

@@ -0,0 +1,432 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#include "utils/asprintf.h"
+#include "utils/time.h"
+
+#include "mqtt_config.h"
+#include "mqtt_handle.h"
+#include "mqtt_plugin.h"
+
+const neu_plugin_module_t neu_plugin_module;
+
+static void connect_cb(void *data)
+{
+    neu_plugin_t *plugin      = data;
+    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
+    plog_notice(plugin, "plugin `%s` connected", neu_plugin_module.module_name);
+}
+
+static void disconnect_cb(void *data)
+{
+    neu_plugin_t *plugin      = data;
+    plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1, NULL);
+    plog_notice(plugin, "plugin `%s` disconnected",
+                neu_plugin_module.module_name);
+}
+
+static neu_plugin_t *mqtt_plugin_open(void)
+{
+    neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t));
+    neu_plugin_common_init(&plugin->common);
+    return plugin;
+}
+
+static int mqtt_plugin_close(neu_plugin_t *plugin)
+{
+    const char *name = neu_plugin_module.module_name;
+    plog_notice(plugin, "success to free plugin:%s", name);
+
+    free(plugin);
+    return NEU_ERR_SUCCESS;
+}
+
+static int mqtt_plugin_init(neu_plugin_t *plugin, bool load)
+{
+    (void) load;
+
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_CACHED_MSGS_NUM, 0);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 600000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1800000);
+
+    plog_notice(plugin, "initialize plugin `%s` success",
+                neu_plugin_module.module_name);
+    return NEU_ERR_SUCCESS;
+}
+
+static int mqtt_plugin_uninit(neu_plugin_t *plugin)
+{
+    mqtt_config_fini(&plugin->config);
+    if (plugin->client) {
+        neu_mqtt_client_close(plugin->client);
+        neu_mqtt_client_free(plugin->client);
+        plugin->client = NULL;
+    }
+
+    free(plugin->read_req_topic);
+    plugin->read_req_topic = NULL;
+    free(plugin->read_resp_topic);
+    plugin->read_resp_topic = NULL;
+
+    route_tbl_free(plugin->route_tbl);
+
+    plog_notice(plugin, "uninitialize plugin `%s` success",
+                neu_plugin_module.module_name);
+    return NEU_ERR_SUCCESS;
+}
+
+static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client,
+                              const mqtt_config_t *config)
+{
+    int rv = 0;
+
+    if (NULL == client) {
+        return 0;
+    }
+
+    // set log category as soon as possible to ease debugging
+    rv = neu_mqtt_client_set_zlog_category(client, plugin->common.log);
+    if (0 != rv) {
+        plog_error(plugin, "neu_mqtt_client_set_zlog_category fail");
+        return -1;
+    }
+
+    rv = neu_mqtt_client_set_addr(client, config->host, config->port);
+    if (0 != rv) {
+        plog_error(plugin, "neu_mqtt_client_set_host fail");
+        return -1;
+    }
+
+    rv = neu_mqtt_client_set_id(client, config->client_id);
+    if (0 != rv) {
+        plog_error(plugin, "neu_mqtt_client_set_id fail");
+        return -1;
+    }
+
+    rv = neu_mqtt_client_set_connect_cb(client, connect_cb, plugin);
+    if (0 != rv) {
+        plog_error(plugin, "neu_mqtt_client_set_connect_cb fail");
+        return -1;
+    }
+
+    rv = neu_mqtt_client_set_disconnect_cb(client, disconnect_cb, plugin);
+    if (0 != rv) {
+        plog_error(plugin, "neu_mqtt_client_set_disconnect_cb fail");
+        return -1;
+    }
+
+    rv = neu_mqtt_client_set_cache_size(client, config->cache_mem_size,
+                                        config->cache_disk_size);
+    if (0 != rv) {
+        plog_error(plugin, "neu_mqtt_client_set_msg_cache_limit fail");
+        return -1;
+    }
+
+    rv = neu_mqtt_client_set_cache_sync_interval(client,
+                                                 config->cache_sync_interval);
+    if (0 != rv) {
+        plog_error(plugin, "neu_mqtt_client_set_msg_cache_sync_interval fail");
+        return -1;
+    }
+
+    if (NULL != config->username) {
+        rv = neu_mqtt_client_set_user(client, config->username,
+                                      config->password);
+        if (0 != rv) {
+            plog_error(plugin, "neu_mqtt_client_set_user fail");
+        }
+    }
+
+    rv = neu_mqtt_client_set_tls(client, config->ssl, config->ca, config->cert,
+                                 config->key, config->keypass);
+    if (0 != rv) {
+        plog_error(plugin, "neu_mqtt_client_set_tsl fail");
+        return -1;
+    }
+
+    return rv;
+}
+
+static int create_topic(neu_plugin_t *plugin)
+{
+    if (plugin->read_req_topic) {
+        // topics already created
+        return 0;
+    }
+
+    neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
+                 plugin->common.name);
+    if (NULL == plugin->read_req_topic) {
+        return -1;
+    }
+
+    neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
+                 plugin->common.name);
+    if (NULL == plugin->read_resp_topic) {
+        free(plugin->read_req_topic);
+        plugin->read_req_topic = NULL;
+        return -1;
+    }
+
+    return 0;
+}
+
+static int subscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
+{
+    if (0 != create_topic(plugin)) {
+        plog_error(plugin, "create topics fail");
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (0 !=
+        neu_mqtt_client_subscribe(plugin->client, config->qos,
+                                  plugin->read_req_topic, plugin,
+                                  handle_read_req)) {
+        plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic);
+        return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+    }
+
+    if (0 !=
+        neu_mqtt_client_subscribe(plugin->client, config->qos,
+                                  config->write_req_topic, plugin,
+                                  handle_write_req)) {
+        plog_error(plugin, "subscribe [%s] fail",
+                   plugin->config.write_req_topic);
+        return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+    }
+
+    return 0;
+}
+
+static int unsubscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
+{
+    neu_mqtt_client_unsubscribe(plugin->client, plugin->read_req_topic);
+    neu_mqtt_client_unsubscribe(plugin->client, config->write_req_topic);
+    neu_msleep(100); // wait for message completion
+    return 0;
+}
+
+static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
+{
+    int           rv          = 0;
+    const char *  plugin_name = neu_plugin_module.module_name;
+    mqtt_config_t config      = { 0 };
+    bool          started     = false;
+
+    rv = mqtt_config_parse(plugin, setting, &config);
+    if (0 != rv) {
+        plog_error(plugin, "neu_mqtt_config_parse fail");
+        return NEU_ERR_NODE_SETTING_INVALID;
+    }
+
+    if (NULL == plugin->client) {
+        plugin->client = neu_mqtt_client_new(NEU_MQTT_VERSION_V311);
+        if (NULL == plugin->client) {
+            plog_error(plugin, "neu_mqtt_client_new fail");
+            rv = NEU_ERR_EINTERNAL;
+            goto error;
+        }
+    } else if (neu_mqtt_client_is_open(plugin->client)) {
+        started = true;
+        unsubscribe(plugin, &plugin->config);
+        rv = neu_mqtt_client_close(plugin->client);
+        if (0 != rv) {
+            plog_error(plugin, "neu_mqtt_client_close fail");
+            rv = NEU_ERR_EINTERNAL;
+            goto error;
+        }
+    }
+
+    rv = config_mqtt_client(plugin, plugin->client, &config);
+    if (0 != rv) {
+        rv = NEU_ERR_MQTT_INIT_FAILURE;
+        goto error;
+    }
+
+    if (started) {
+        if (0 != neu_mqtt_client_open(plugin->client)) {
+            plog_error(plugin, "neu_mqtt_client_open fail");
+            rv = NEU_ERR_MQTT_CONNECT_FAILURE;
+            goto error;
+        }
+        if (0 != (rv = subscribe(plugin, &config))) {
+            goto error;
+        }
+    }
+
+    if (plugin->config.host) {
+        // already configured
+        mqtt_config_fini(&plugin->config);
+    }
+    memmove(&plugin->config, &config, sizeof(config));
+
+    plog_notice(plugin, "config plugin `%s` success", plugin_name);
+    return 0;
+
+error:
+    plog_error(plugin, "config plugin `%s` fail", plugin_name);
+    mqtt_config_fini(&config);
+    return rv;
+}
+
+static int mqtt_plugin_start(neu_plugin_t *plugin)
+{
+    int         rv          = 0;
+    const char *plugin_name = neu_plugin_module.module_name;
+
+    if (NULL == plugin->client) {
+        plog_error(plugin, "mqtt client is NULL due to init failure");
+        rv = NEU_ERR_MQTT_IS_NULL;
+        goto end;
+    }
+
+    if (0 != neu_mqtt_client_open(plugin->client)) {
+        plog_error(plugin, "neu_mqtt_client_open fail");
+        rv = NEU_ERR_MQTT_CONNECT_FAILURE;
+        goto end;
+    }
+
+    rv = subscribe(plugin, &plugin->config);
+
+end:
+    if (0 == rv) {
+        plog_notice(plugin, "start plugin `%s` success", plugin_name);
+    } else {
+        plog_error(plugin, "start plugin `%s` failed, error %d", plugin_name,
+                   rv);
+        neu_mqtt_client_close(plugin->client);
+    }
+    return rv;
+}
+
+static int mqtt_plugin_stop(neu_plugin_t *plugin)
+{
+    if (plugin->client) {
+        unsubscribe(plugin, &plugin->config);
+        neu_mqtt_client_close(plugin->client);
+        plog_notice(plugin, "mqtt client closed");
+    }
+
+    plog_notice(plugin, "stop plugin `%s` success",
+                neu_plugin_module.module_name);
+    return NEU_ERR_SUCCESS;
+}
+
+static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
+                               void *data)
+{
+    neu_err_code_e error = NEU_ERR_SUCCESS;
+
+    // update cached messages number per seconds
+    if (NULL != plugin->client &&
+        (global_timestamp - plugin->cache_metric_update_ts) >= 1000) {
+        NEU_PLUGIN_UPDATE_METRIC(
+            plugin, NEU_METRIC_CACHED_MSGS_NUM,
+            neu_mqtt_client_get_cached_msgs_num(plugin->client), NULL);
+        plugin->cache_metric_update_ts = global_timestamp;
+    }
+
+    switch (head->type) {
+    case NEU_RESP_ERROR:
+        error = handle_write_resp(plugin, head->ctx, data);
+        break;
+    case NEU_RESP_WRITE_BATCH:
+        error = handle_write_resp(plugin, head->ctx, data);
+        break;
+    case NEU_RESP_READ_GROUP:
+        error = handle_read_response(plugin, head->ctx, data);
+        break;
+    case NEU_REQRESP_TRANS_DATA: {
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 1, NULL);
+        error = handle_trans_data(plugin, data);
+        break;
+    }
+    case NEU_REQ_SUBSCRIBE_GROUP:
+        error = handle_subscribe_group(plugin, data);
+        break;
+    case NEU_REQ_UPDATE_SUBSCRIBE_GROUP:
+        error = handle_update_subscribe(plugin, data);
+        break;
+    case NEU_REQ_UNSUBSCRIBE_GROUP:
+        error = handle_unsubscribe_group(plugin, data);
+        break;
+    case NEU_REQ_UPDATE_GROUP:
+        error = handle_update_group(plugin, data);
+        break;
+    case NEU_REQ_DEL_GROUP:
+        error = handle_del_group(plugin, data);
+        break;
+    case NEU_REQ_UPDATE_NODE:
+        error = handle_update_driver(plugin, data);
+        break;
+    case NEU_REQRESP_NODE_DELETED:
+        error = handle_del_driver(plugin, data);
+        break;
+    default:
+        error = NEU_ERR_MQTT_FAILURE;
+        break;
+    }
+
+    return error;
+}
+
+static const neu_plugin_intf_funs_t plugin_intf_funs = {
+    .open    = mqtt_plugin_open,
+    .close   = mqtt_plugin_close,
+    .init    = mqtt_plugin_init,
+    .uninit  = mqtt_plugin_uninit,
+    .start   = mqtt_plugin_start,
+    .stop    = mqtt_plugin_stop,
+    .setting = mqtt_plugin_config,
+    .request = mqtt_plugin_request,
+};
+
+#define DESCRIPTION "Northbound MQTT plugin bases on NanoSDK."
+#define DESCRIPTION_ZH "基于 NanoSDK 的北向应用 MQTT 插件"
+
+const neu_plugin_module_t neu_plugin_module = {
+    .version         = NEURON_PLUGIN_VER_1_0,
+    .schema          = "mqtt-xinao",
+    .module_name     = "MQTT XINAO",
+    .module_descr    = DESCRIPTION,
+    .module_descr_zh = DESCRIPTION_ZH,
+    .intf_funs       = &plugin_intf_funs,
+    .kind            = NEU_PLUGIN_KIND_SYSTEM,
+    .type            = NEU_NA_TYPE_APP,
+    .display         = true,
+    .single          = false,
+};

plugins/mqtt-xinao/mqtt_plugin.h

@@ -0,0 +1,182 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#ifndef NEURON_PLUGIN_MQTT_H
+#define NEURON_PLUGIN_MQTT_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "connection/mqtt_client.h"
+#include "neuron.h"
+
+#include "mqtt_config.h"
+
+typedef struct {
+    char driver[NEU_NODE_NAME_LEN];
+    char group[NEU_GROUP_NAME_LEN];
+} route_key_t;
+
+typedef struct {
+    route_key_t key;
+
+    char *topic;
+
+    UT_hash_handle hh;
+} route_entry_t;
+
+struct neu_plugin {
+    neu_plugin_common_t common;
+    mqtt_config_t       config;
+    neu_mqtt_client_t * client;
+    int64_t             cache_metric_update_ts;
+    char *              read_req_topic;
+    char *              read_resp_topic;
+    route_entry_t *     route_tbl;
+};
+
+static inline void route_entry_free(route_entry_t *e)
+{
+    free(e->topic);
+    free(e);
+}
+
+static inline void route_tbl_free(route_entry_t *tbl)
+{
+    route_entry_t *e = NULL, *tmp = NULL;
+    HASH_ITER(hh, tbl, e, tmp)
+    {
+        HASH_DEL(tbl, e);
+        route_entry_free(e);
+    }
+}
+
+static inline route_entry_t *
+route_tbl_get(route_entry_t **tbl, const char *driver, const char *group)
+{
+    route_entry_t *find = NULL;
+    route_key_t    key  = { 0 };
+
+    strncpy(key.driver, driver, sizeof(key.driver));
+    strncpy(key.group, group, sizeof(key.group));
+
+    HASH_FIND(hh, *tbl, &key, sizeof(key), find);
+    return find;
+}
+
+// NOTE: we take ownership of `topic`
+static inline int route_tbl_add_new(route_entry_t **tbl, const char *driver,
+                                    const char *group, char *topic)
+{
+    route_entry_t *find = NULL;
+
+    find = route_tbl_get(tbl, driver, group);
+    if (find) {
+        free(topic);
+        return NEU_ERR_GROUP_ALREADY_SUBSCRIBED;
+    }
+
+    find = calloc(1, sizeof(*find));
+    if (NULL == find) {
+        free(topic);
+        return NEU_ERR_EINTERNAL;
+    }
+
+    strncpy(find->key.driver, driver, sizeof(find->key.driver));
+    strncpy(find->key.group, group, sizeof(find->key.group));
+    find->topic = topic;
+    HASH_ADD(hh, *tbl, key, sizeof(find->key), find);
+
+    return 0;
+}
+
+// NOTE: we take ownership of `topic`
+static inline int route_tbl_update(route_entry_t **tbl, const char *driver,
+                                   const char *group, char *topic)
+{
+    route_entry_t *find = NULL;
+
+    find = route_tbl_get(tbl, driver, group);
+    if (NULL == find) {
+        free(topic);
+        return NEU_ERR_GROUP_NOT_SUBSCRIBE;
+    }
+
+    free(find->topic);
+    find->topic = topic;
+
+    return 0;
+}
+
+static inline void route_tbl_update_driver(route_entry_t **tbl,
+                                           const char *    driver,
+                                           const char *    new_name)
+{
+    route_entry_t *e = NULL, *tmp = NULL;
+    HASH_ITER(hh, *tbl, e, tmp)
+    {
+        if (0 == strcmp(e->key.driver, driver)) {
+            HASH_DEL(*tbl, e);
+            strncpy(e->key.driver, new_name, sizeof(e->key.driver));
+            HASH_ADD(hh, *tbl, key, sizeof(e->key), e);
+        }
+    }
+}
+
+static inline void route_tbl_update_group(route_entry_t **tbl,
+                                          const char *driver, const char *group,
+                                          const char *new_name)
+{
+    route_entry_t *e = route_tbl_get(tbl, driver, group);
+    if (e) {
+        HASH_DEL(*tbl, e);
+        strncpy(e->key.group, new_name, sizeof(e->key.group));
+        HASH_ADD(hh, *tbl, key, sizeof(e->key), e);
+    }
+}
+
+static inline void route_tbl_del_driver(route_entry_t **tbl, const char *driver)
+{
+    route_entry_t *e = NULL, *tmp = NULL;
+    HASH_ITER(hh, *tbl, e, tmp)
+    {
+        if (0 == strcmp(e->key.driver, driver)) {
+            HASH_DEL(*tbl, e);
+            route_entry_free(e);
+        }
+    }
+}
+
+static inline void route_tbl_del(route_entry_t **tbl, const char *driver,
+                                 const char *group)
+{
+    route_entry_t *find = NULL;
+    find                = route_tbl_get(tbl, driver, group);
+    if (find) {
+        HASH_DEL(*tbl, find);
+        route_entry_free(find);
+    }
+}
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

plugins/mqtt-xinao/xinao.c

@@ -0,0 +1,469 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#include <stdlib.h>
+#include <string.h>
+
+#include <jansson.h>
+
+#include "define.h"
+#include "metrics.h"
+#include "msg.h"
+#include "tag.h"
+#include "type.h"
+#include "utils/log.h"
+#include "utils/uthash.h"
+
+#include "xinao.h"
+
+static inline int json_value_to_tag_value(neu_dvalue_t *          value,
+                                          neu_json_type_e         t,
+                                          const neu_json_value_u *jval)
+{
+    switch (t) {
+    case NEU_JSON_INT:
+        value->type      = NEU_TYPE_INT64;
+        value->value.i64 = jval->val_int;
+        break;
+    case NEU_JSON_STR:
+        value->type = NEU_TYPE_STRING;
+        strcpy(value->value.str, jval->val_str);
+        break;
+    case NEU_JSON_BYTES:
+        value->type               = NEU_TYPE_BYTES;
+        value->value.bytes.length = jval->val_bytes.length;
+        memcpy(value->value.bytes.bytes, jval->val_bytes.bytes,
+               jval->val_bytes.length);
+        break;
+    case NEU_JSON_DOUBLE:
+        value->type      = NEU_TYPE_DOUBLE;
+        value->value.d64 = jval->val_double;
+        break;
+    case NEU_JSON_BOOL:
+        value->type          = NEU_TYPE_BOOL;
+        value->value.boolean = jval->val_bool;
+        break;
+    default:
+        return -1;
+    }
+
+    return 0;
+}
+
+int neu_json_decode_xinao_write_tag_req_json(neu_plugin_t *    plugin,
+                                             void *            json_root,
+                                             xinao_wtag_ctx_t *ctx)
+{
+    json_t *data = json_object_get(json_root, "data");
+    if (NULL == data || !json_is_object(data)) {
+        plog_error(plugin, "no `data` object");
+        return -1;
+    }
+
+    json_t *sysId = json_object_get(data, "sysId");
+    if (NULL == sysId || !json_is_string(sysId)) {
+        plog_error(plugin, "data no valid key `sysId`");
+        return -1;
+    }
+
+    json_t *node_name  = json_object_get(data, "dev");
+    json_t *group_name = node_name;
+    if (NULL == node_name || !json_is_string(node_name)) {
+        plog_error(plugin, "data no valid key `dev`");
+        return -1;
+    }
+
+    json_t *tag = json_object_get(data, "m");
+
+    neu_json_elem_t elem = {
+        .name = "v",
+        .t    = NEU_JSON_VALUE,
+    };
+
+    if (neu_json_decode_by_json(data, 1, &elem)) {
+        plog_error(plugin, "sysId:%s dev:%s m:%s decode v fail",
+                   json_string_value(sysId), json_string_value(node_name),
+                   json_string_value(tag));
+        return -1;
+    }
+
+    int ret = json_value_to_tag_value(&ctx->wtag.value, elem.t, &elem.v);
+    if (NEU_JSON_STR == elem.t) {
+        free(elem.v.val_str);
+    }
+    if (0 != ret) {
+        return -1;
+    }
+
+    ctx->type        = NEU_REQ_WRITE_TAG;
+    ctx->wtag.driver = strdup(json_string_value(node_name));
+    ctx->wtag.group  = strdup(json_string_value(group_name));
+    ctx->wtag.tag    = strdup(json_string_value(tag));
+
+    if (NULL == ctx->wtag.driver || NULL == ctx->wtag.group ||
+        NULL == ctx->wtag.tag) {
+        free(ctx->wtag.driver);
+        free(ctx->wtag.group);
+        free(ctx->wtag.tag);
+        return -1;
+    }
+
+    return 0;
+}
+
+int neu_json_decode_xinao_write_batch_req_json(neu_plugin_t *     plugin,
+                                               void *             json_root,
+                                               xinao_batch_ctx_t *ctx)
+{
+    json_t *cmd = json_object_get(json_root, "cmd");
+    if (NULL == cmd || !json_is_array(cmd)) {
+        plog_error(plugin, "no `cmd` array");
+        return -1;
+    }
+
+    int n_cmd = json_array_size(cmd);
+    if (0 == n_cmd) {
+        plog_error(plugin, "empty `cmd` array");
+        return -1;
+    }
+
+    index_map_t        map   = NULL;
+    neu_write_batch_t *batch = NULL;
+
+    // quick check
+    for (int i = 0; i < n_cmd; ++i) {
+        json_t *a_cmd = json_array_get(cmd, i);
+
+        json_t *sysId = json_object_get(a_cmd, "sysId");
+        if (NULL == sysId || !json_is_string(sysId)) {
+            plog_error(plugin, "cmd[%d] no valid key `sysId`", i);
+            goto error;
+        }
+
+        json_t *node_name  = json_object_get(a_cmd, "dev");
+        json_t *group_name = node_name;
+        if (NULL == node_name || !json_is_string(node_name)) {
+            plog_error(plugin, "cmd[%d] no valid key `dev`", i);
+            goto error;
+        }
+
+        json_t *tags = json_object_get(a_cmd, "d");
+        if (NULL == tags || !json_is_array(tags)) {
+            plog_error(plugin, "cmd[%d] no `d` array", i);
+            goto error;
+        }
+
+        int n_tag = json_array_size(tags);
+        if (0 == n_tag) {
+            plog_error(plugin, "cmd[%d] empty `d` array", i);
+            goto error;
+        }
+
+        for (int j = 0; j < n_tag; ++j) {
+            json_t *tag_value = json_array_get(tags, j);
+            json_t *tag       = json_object_get(tag_value, "m");
+            json_t *value     = json_object_get(tag_value, "v");
+
+            if (NULL == tag || !json_is_string(tag)) {
+                plog_error(plugin, "cmd[%d].d[%d] no valid key `m`", i, j);
+                goto error;
+            }
+
+            if (NULL == value || json_is_object(value)) {
+                plog_error(plugin, "cmd[%d].d[%d] no valid key `v`", i, j);
+                goto error;
+            }
+        }
+
+        if (0 !=
+            index_map_add(&map, json_string_value(node_name),
+                          json_string_value(group_name))) {
+            plog_error(plugin, "cmd[%d] {sysId:%s, dev:%s} duplicate", i,
+                       json_string_value(sysId), json_string_value(node_name));
+            goto error;
+        }
+    }
+
+    batch =
+        calloc(1, sizeof(*batch) + HASH_COUNT(map) * sizeof(batch->drivers[0]));
+    if (NULL == batch) {
+        return -1;
+    }
+
+    batch->n_driver = HASH_COUNT(map);
+
+    for (int i = 0; i < n_cmd; ++i) {
+        json_t *a_cmd      = json_array_get(cmd, i);
+        json_t *sysId      = json_object_get(a_cmd, "sysId");
+        json_t *node_name  = json_object_get(a_cmd, "dev");
+        json_t *group_name = node_name;
+
+        node_index_t * n = NULL;
+        group_index_t *g = NULL;
+        HASH_FIND_STR(map, json_string_value(node_name), n);
+        HASH_FIND_STR(n->grp_indices, json_string_value(group_name), g);
+
+        neu_write_batch_driver_t *batch_driver = &batch->drivers[n->ind];
+        if (NULL == batch_driver->driver) {
+            batch_driver->driver = strdup(json_string_value(node_name));
+            if (NULL == batch_driver->driver) {
+                goto error;
+            }
+        }
+
+        if (NULL == batch_driver->groups) {
+            batch_driver->n_group = HASH_COUNT(g);
+            batch_driver->groups =
+                calloc(HASH_COUNT(g), sizeof(batch_driver->groups[0]));
+            if (NULL == batch_driver->groups) {
+                goto error;
+            }
+        }
+
+        neu_write_batch_group_t *batch_group = &batch_driver->groups[g->ind];
+        batch_group->group = strdup(json_string_value(group_name));
+        if (NULL == batch_group->group) {
+            goto error;
+        }
+
+        json_t *tags  = json_object_get(a_cmd, "d");
+        int     n_tag = json_array_size(tags);
+
+        batch_group->tags = calloc(n_tag, sizeof(batch_group->tags[0]));
+        if (NULL == batch_group->tags) {
+            goto error;
+        }
+
+        batch_group->n_tag = n_tag;
+
+        for (int j = 0; j < n_tag; ++j) {
+            json_t *tag_value = json_array_get(tags, j);
+            json_t *tag       = json_object_get(tag_value, "m");
+
+            batch_group->tags[j].name = strdup(json_string_value(tag));
+            if (NULL == batch_group->tags[j].name) {
+                goto error;
+            }
+
+            neu_json_elem_t elem = {
+                .name = "v",
+                .t    = NEU_JSON_VALUE,
+            };
+
+            int ret = neu_json_decode_by_json(tag_value, 1, &elem);
+            if (0 != ret) {
+                plog_error(plugin, "sysId:%s dev:%s m:%s decode v fail",
+                           json_string_value(sysId), batch_driver->driver,
+                           batch_group->tags[j].name);
+                goto error;
+            }
+
+            ret = json_value_to_tag_value(&batch_group->tags[j].value, elem.t,
+                                          &elem.v);
+            if (NEU_JSON_STR == elem.t) {
+                free(elem.v.val_str);
+            }
+            if (0 != ret) {
+                goto error;
+            }
+        }
+    }
+
+    ctx->type        = NEU_REQ_WRITE_BATCH;
+    ctx->batch.batch = batch;
+    ctx->map         = map;
+    return 0;
+
+error:
+    index_map_free(&map);
+    neu_write_batch_free(batch);
+    free(batch);
+    return -1;
+}
+
+int neu_json_decode_xinao_ctx_json(neu_plugin_t *plugin, void *json_root,
+                                   xinao_ctx_t **ctx_p)
+{
+    json_t *ver  = json_object_get(json_root, "ver");
+    json_t *pKey = json_object_get(json_root, "pKey");
+    json_t *sn   = json_object_get(json_root, "sn");
+    json_t *type = json_object_get(json_root, "type");
+    json_t *seq  = json_object_get(json_root, "seq");
+    json_t *ts   = json_object_get(json_root, "ts");
+
+    if (NULL == ver || !json_is_string(ver)) {
+        plog_error(plugin, "no valid key `ver`");
+        return -1;
+    }
+
+    if (NULL == pKey || !json_is_string(pKey)) {
+        plog_error(plugin, "no valid key `pKey`");
+        return -1;
+    }
+
+    if (NULL == sn || !json_is_string(sn)) {
+        plog_error(plugin, "no valid key `sn`");
+        return -1;
+    }
+
+    if (NULL == seq || !json_is_integer(seq)) {
+        plog_error(plugin, "no valid key `seq`");
+        return -1;
+    }
+
+    if (NULL == ts || !json_is_integer(ts)) {
+        plog_error(plugin, "no valid key `ts`");
+        return -1;
+    }
+
+    if (NULL == type || !json_is_string(type)) {
+        plog_error(plugin, "no valid key `type`");
+        return -1;
+    }
+
+    int ret = -1;
+    if (0 == strcmp("algset", json_string_value(type))) {
+        xinao_batch_ctx_t *batch_ctx = calloc(1, sizeof(*batch_ctx));
+        if (NULL != batch_ctx) {
+            ret = neu_json_decode_xinao_write_batch_req_json(plugin, json_root,
+                                                             batch_ctx);
+            if (0 == ret) {
+                *ctx_p = (xinao_ctx_t *) batch_ctx;
+            } else {
+                plog_error(plugin, "decode write batch json fail");
+                free(batch_ctx);
+            }
+        }
+    } else if (0 == strcmp("cmd/set", json_string_value(type))) {
+        xinao_wtag_ctx_t *wtag_ctx = calloc(1, sizeof(*wtag_ctx));
+        if (NULL != wtag_ctx) {
+            ret = neu_json_decode_xinao_write_tag_req_json(plugin, json_root,
+                                                           wtag_ctx);
+            if (0 == ret) {
+                *ctx_p = (xinao_ctx_t *) wtag_ctx;
+            } else {
+                plog_error(plugin, "decode write tag json fail");
+                free(wtag_ctx);
+            }
+        }
+    } else {
+        plog_error(plugin, "unknown key `type`:%s", json_string_value(type));
+    }
+
+    return ret;
+}
+
+xinao_ctx_t *xinao_ctx_from_json_buf(neu_plugin_t *plugin, char *buf,
+                                     size_t len)
+{
+    void *root = neu_json_decode_newb(buf, len);
+    if (NULL == root) {
+        plog_error(plugin, "invalid batch req json");
+        return NULL;
+    }
+
+    xinao_ctx_t *ctx = NULL;
+    if (0 != neu_json_decode_xinao_ctx_json(plugin, root, &ctx)) {
+        plog_error(plugin, "decode write req json fail");
+        neu_json_decode_free(root);
+        return NULL;
+    }
+
+    ctx->root = root;
+    return ctx;
+}
+
+char *xinao_ctx_gen_wtag_resp_json(xinao_wtag_ctx_t *ctx,
+                                   neu_resp_error_t *resp)
+{
+    json_object_set_new(ctx->root, "ts", json_integer(global_timestamp));
+    json_object_set_new(ctx->root, "type", json_string("cmd/set/cack"));
+
+    json_t *data = json_object_get(ctx->root, "data");
+
+    int  valid    = true;
+    char error[6] = { 0 };
+    if (0 != resp->error) {
+        valid = false;
+        snprintf(error, sizeof(error), "%d", resp->error);
+    }
+
+    json_object_set_new(data, "valid", json_integer(valid));
+    json_object_set_new(data, "remark", json_string(error));
+
+    char *result = json_dumps(ctx->root, 0);
+    return result;
+}
+
+char *xinao_ctx_gen_batch_resp_json(xinao_batch_ctx_t *ctx)
+{
+    json_t *cmd   = json_object_get(ctx->root, "cmd");
+    int     n_cmd = json_array_size(cmd);
+
+    // rename to `data`
+    json_incref(cmd);
+    json_object_del(ctx->root, "cmd");
+    json_object_set_new(ctx->root, "data", cmd);
+
+    json_object_set_new(ctx->root, "ts", json_integer(global_timestamp));
+    json_object_set_new(ctx->root, "type", json_string("algset/cack"));
+
+    for (int i = 0; i < n_cmd; ++i) {
+        json_t *a_cmd      = json_array_get(cmd, i);
+        json_t *node_name  = json_object_get(a_cmd, "dev");
+        json_t *group_name = node_name;
+
+        node_index_t * n = NULL;
+        group_index_t *g = NULL;
+        HASH_FIND_STR(ctx->map, json_string_value(node_name), n);
+        HASH_FIND_STR(n->grp_indices, json_string_value(group_name), g);
+        neu_write_batch_driver_t *batch_driver =
+            &ctx->batch.batch->drivers[n->ind];
+        neu_write_batch_group_t *batch_group = &batch_driver->groups[g->ind];
+
+        json_t *tags     = json_object_get(a_cmd, "d");
+        int     n_tag    = json_array_size(tags);
+        char    error[6] = { 0 };
+
+        for (int j = 0; j < n_tag; ++j) {
+            json_t *tag_value = json_array_get(tags, j);
+            int     valid;
+            if (0 != batch_driver->error) {
+                valid = false;
+                snprintf(error, sizeof(error), "%d", batch_driver->error);
+            } else if (0 != batch_group->error) {
+                valid = false;
+                snprintf(error, sizeof(error), "%d", batch_group->error);
+            } else if (NEU_TYPE_ERROR == batch_group->tags[j].value.type) {
+                valid = false;
+                snprintf(error, sizeof(error), "%d",
+                         batch_group->tags[j].value.value.i32);
+            } else {
+                valid    = true;
+                error[0] = '\0';
+            }
+
+            json_object_set_new(tag_value, "valid", json_integer(valid));
+            json_object_set_new(tag_value, "remark", json_string(error));
+        }
+    }
+
+    char *result = json_dumps(ctx->root, 0);
+    return result;
+}

plugins/mqtt-xinao/xinao.h

@@ -0,0 +1,151 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#ifndef NEURON_PLUGIN_MQTT_XINAO
+#define NEURON_PLUGIN_MQTT_XINAO
+
+#include "msg.h"
+#include "plugin.h"
+#include "tag.h"
+#include "utils/uthash.h"
+#include "json/json.h"
+
+#include "mqtt_plugin.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct {
+    const char *   group;
+    int            ind;
+    UT_hash_handle hh;
+} group_index_t;
+
+typedef struct {
+    const char *   node;
+    int            ind;
+    group_index_t *grp_indices;
+    UT_hash_handle hh;
+} node_index_t;
+
+typedef node_index_t *index_map_t;
+
+typedef struct {
+    neu_reqresp_type_e  type;
+    void *              root;
+    neu_req_write_tag_t wtag;
+} xinao_wtag_ctx_t;
+
+typedef struct {
+    neu_reqresp_type_e        type;
+    void *                    root;
+    index_map_t               map;
+    neu_reqresp_write_batch_t batch;
+} xinao_batch_ctx_t;
+
+typedef struct {
+    neu_reqresp_type_e type;
+    void *             root;
+} xinao_ctx_t;
+
+static inline void index_map_free(index_map_t *map)
+{
+    node_index_t * n = NULL, *ntmp = NULL;
+    group_index_t *g = NULL, *gtmp = NULL;
+    HASH_ITER(hh, *map, n, ntmp)
+    {
+        HASH_ITER(hh, n->grp_indices, g, gtmp)
+        {
+            HASH_DEL(n->grp_indices, g);
+            free(g);
+        }
+        HASH_DEL(*map, n);
+        free(n);
+    }
+}
+
+static inline int index_map_add(index_map_t *map, const char *node,
+                                const char *group)
+{
+    node_index_t *n = NULL;
+    HASH_FIND_STR(*map, node, n);
+    if (NULL == n) {
+        n = calloc(1, sizeof(*n));
+        if (NULL == n) {
+            return -1;
+        }
+        n->node = node;
+        n->ind  = HASH_COUNT(*map);
+        HASH_ADD_KEYPTR(hh, *map, n->node, strlen(n->node), n);
+    }
+
+    group_index_t *g = NULL;
+    HASH_FIND_STR(n->grp_indices, group, g);
+    if (NULL != g) {
+        return -1;
+    }
+
+    g = calloc(1, sizeof(*g));
+    if (NULL == g) {
+        return -1;
+    }
+    g->ind   = HASH_COUNT(n->grp_indices);
+    g->group = group;
+    HASH_ADD_KEYPTR(hh, n->grp_indices, g->group, strlen(g->group), g);
+
+    return 0;
+}
+
+int neu_json_decode_xinao_write_tag_req_json(neu_plugin_t *    plugin,
+                                             void *            json_root,
+                                             xinao_wtag_ctx_t *ctx);
+
+int neu_json_decode_xinao_write_batch_req_json(neu_plugin_t *     plugin,
+                                               void *             json_root,
+                                               xinao_batch_ctx_t *ctx);
+
+xinao_ctx_t *xinao_ctx_from_json_buf(neu_plugin_t *plugin, char *buf,
+                                     size_t len);
+
+char *xinao_ctx_gen_wtag_resp_json(xinao_wtag_ctx_t *ctx,
+                                   neu_resp_error_t *resp);
+char *xinao_ctx_gen_batch_resp_json(xinao_batch_ctx_t *ctx);
+
+static inline void xinao_ctx_free(xinao_ctx_t *ctx)
+{
+    if (ctx) {
+        neu_json_decode_free(ctx->root);
+        if (NEU_REQ_WRITE_TAG == ctx->type) {
+            xinao_wtag_ctx_t *wtag = (xinao_wtag_ctx_t *) ctx;
+            neu_req_write_tag_fini(&wtag->wtag);
+        } else if (NEU_REQ_WRITE_BATCH == ctx->type) {
+            xinao_batch_ctx_t *batch = (xinao_batch_ctx_t *) ctx;
+            index_map_free(&batch->map);
+            neu_reqresp_write_batch_fini(&batch->batch);
+        }
+        free(ctx);
+    }
+}
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/adapter/adapter.c

@@ -707,6 +707,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQRESP_NODES_STATE:
     case NEU_REQ_PRGFILE_PROCESS:
     case NEU_RESP_PRGFILE_PROCESS:
+    case NEU_RESP_WRITE_BATCH:
         adapter->module->intf_funs->request(
             adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
         neu_msg_free(msg);
@@ -803,6 +804,24 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
         break;
     }
+    case NEU_REQ_WRITE_BATCH: {
+        neu_reqresp_write_batch_t *cmd =
+            (neu_reqresp_write_batch_t *) &header[1];
+        neu_write_batch_t *       batch  = cmd->batch;
+        neu_write_batch_driver_t *driver = &batch->drivers[cmd->index_];
+
+        if (adapter->module->type != NEU_NA_TYPE_DRIVER) {
+            driver->error = NEU_ERR_GROUP_NOT_ALLOW;
+            header->type  = NEU_RESP_WRITE_BATCH;
+            neu_msg_exchange(header);
+            reply(adapter, header, cmd);
+        } else {
+            neu_adapter_driver_write_batch((neu_adapter_driver_t *) adapter,
+                                           header);
+        }
+
+        break;
+    }
     case NEU_REQ_NODE_SETTING: {
         neu_req_node_setting_t *cmd   = (neu_req_node_setting_t *) &header[1];
         neu_resp_error_t        error = { 0 };

src/adapter/driver/driver.c

@@ -162,6 +162,12 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
         neu_req_write_gtags_fini((neu_req_write_gtags_t *) &req[1]);
     }
 
+    if (NEU_REQ_WRITE_BATCH == req->type) {
+        req->type = NEU_RESP_WRITE_BATCH;
+        adapter->cb_funs.response(adapter, req, &req[1]);
+        return;
+    }
+
     req->type = NEU_RESP_ERROR;
 
     nlog_notice("write tag response <%p>", req->ctx);
@@ -979,6 +985,83 @@ void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver,
     store_write_tag(first_g, &wtag);
 }
 
+void neu_adapter_driver_write_batch(neu_adapter_driver_t *driver,
+                                    neu_reqresp_head_t *  req)
+{
+    neu_reqresp_write_batch_t *cmd = (neu_reqresp_write_batch_t *) &req[1];
+    neu_write_batch_driver_t * batch_driver = &cmd->batch->drivers[cmd->index_];
+
+    if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) {
+        batch_driver->error = NEU_ERR_PLUGIN_NOT_RUNNING;
+        driver->adapter.cb_funs.driver.write_response(
+            &driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING);
+        return;
+    }
+
+    if (driver->adapter.module->intf_funs->driver.write_batch == NULL) {
+        batch_driver->error = NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_BATCH;
+        driver->adapter.cb_funs.driver.write_response(
+            &driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS);
+        return;
+    }
+
+    bool need_write = false;
+    for (int i = 0; i < batch_driver->n_group; i++) {
+        neu_write_batch_group_t *batch_group = &batch_driver->groups[i];
+        group_t *                g = find_group(driver, batch_group->group);
+        if (NULL == g) {
+            // mark all group tags error
+            batch_group->error = NEU_ERR_GROUP_NOT_EXIST;
+            continue;
+        }
+
+        for (int j = 0; j < batch_group->n_tag; j++) {
+            neu_datatag_t *tag =
+                neu_group_find_tag(g->group, batch_group->tags[j].name);
+
+            if (NULL == tag) {
+                neu_write_batch_group_set_error(batch_group, j,
+                                                NEU_ERR_TAG_NOT_EXIST);
+                continue;
+            }
+
+            if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_WRITE) ||
+                neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC)) {
+                neu_write_batch_group_set_error(
+                    batch_group, j, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE);
+                continue;
+            }
+
+            if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
+                if (batch_group->tags[j].value.type == NEU_TYPE_INT64) {
+                    batch_group->tags[j].value.value.d64 =
+                        (double) batch_group->tags[j].value.value.i64;
+                }
+            }
+
+            if (tag->decimal != 0) {
+                cal_decimal(tag->type, batch_group->tags[j].value.type,
+                            &batch_group->tags[j].value.value, tag->decimal);
+            }
+
+            fix_value(tag, batch_group->tags[j].value.type,
+                      &batch_group->tags[j].value);
+
+            batch_group->tags[j].tag = tag;
+            need_write               = true;
+        }
+    }
+
+    if (!need_write) {
+        // nothing to request plugin write
+        driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, 0);
+        return;
+    }
+
+    driver->adapter.module->intf_funs->driver.write_batch(
+        driver->adapter.plugin, req, batch_driver);
+}
+
 void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
                                   neu_reqresp_head_t *  req)
 {

src/adapter/driver/driver_internal.h

@@ -42,6 +42,8 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
                                    neu_reqresp_head_t *  req);
 void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver,
                                     neu_reqresp_head_t *  req);
+void neu_adapter_driver_write_batch(neu_adapter_driver_t *driver,
+                                    neu_reqresp_head_t *  req);
 
 int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
                                  uint32_t interval);

src/base/msg.c

@@ -26,5 +26,7 @@ void neu_msg_gen(neu_reqresp_head_t *header, void *data)
 {
     size_t data_size = neu_reqresp_size(header->type);
     assert(header->len >= sizeof(neu_reqresp_head_t) + data_size);
-    memcpy((uint8_t *) &header[1], data, data_size);
-}
+    if (data && &header[1] != data) {
+        memcpy((uint8_t *) &header[1], data, data_size);
+    }
+}

src/base/msg_internal.h

@@ -95,7 +95,9 @@ extern "C" {
     XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t)         \
     XX(NEU_REQ_PRGFILE_UPLOAD, neu_req_prgfile_upload_t)             \
     XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t)           \
-    XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t)
+    XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t)         \
+    XX(NEU_REQ_WRITE_BATCH, neu_reqresp_write_batch_t)               \
+    XX(NEU_RESP_WRITE_BATCH, neu_reqresp_write_batch_t)
 
 static inline size_t neu_reqresp_size(neu_reqresp_type_e t)
 {

src/core/manager.c

@@ -1058,6 +1058,54 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         break;
     }
+    case NEU_REQ_WRITE_BATCH: {
+        neu_reqresp_write_batch_t *cmd =
+            (neu_reqresp_write_batch_t *) &header[1];
+        neu_write_batch_t *batch = cmd->batch;
+
+        for (int i = 0; i < batch->n_driver; ++i) {
+            const char *driver = batch->drivers[i].driver;
+
+            if (NULL == neu_node_manager_find(manager->node_manager, driver)) {
+                // mark errors and done
+                batch->n_done_ += 1;
+                batch->drivers[i].error = NEU_ERR_NODE_NOT_EXIST;
+                continue;
+            }
+
+            cmd->index_ = i;
+            if (i + 1 == batch->n_driver) {
+                forward_msg(manager, header, driver); // save a copy
+            } else {
+                forward_msg_copy(manager, header, driver);
+            }
+        }
+
+        if (batch->n_done_ == batch->n_driver) {
+            // all done, reply response
+            header->type = NEU_RESP_WRITE_BATCH;
+            neu_msg_exchange(header);
+            forward_msg(manager, header, header->receiver);
+        }
+
+        break;
+    }
+    case NEU_RESP_WRITE_BATCH: {
+        neu_reqresp_write_batch_t *cmd =
+            (neu_reqresp_write_batch_t *) &header[1];
+        neu_write_batch_t *batch = cmd->batch;
+
+        batch->n_done_ += 1;
+        if (batch->n_done_ == batch->n_driver) {
+            // all done, reply response
+            header->type = NEU_RESP_WRITE_BATCH;
+            forward_msg(manager, header, header->receiver);
+        } else {
+            neu_msg_free(msg);
+        }
+
+        break;
+    }
     case NEU_REQ_PRGFILE_PROCESS:
     case NEU_REQ_PRGFILE_UPLOAD:
     case NEU_REQ_GET_NODE_STATE: {