From 2efaaed1d03f0451c09e0aeaf78d89e8b9475d89 Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Thu, 16 May 2024 17:45:51 +0800
Subject: [PATCH] feat(xinao): support write both tag and batch

---
 plugins/mqtt-xinao/mqtt_handle.c | 290 ++++++-------------------------
 plugins/mqtt-xinao/mqtt_handle.h |   7 +-
 plugins/mqtt-xinao/mqtt_plugin.c |   6 +-
 plugins/mqtt-xinao/xinao.c       | 188 +++++++++++++++-----
 plugins/mqtt-xinao/xinao.h       |  41 ++++-
 5 files changed, 242 insertions(+), 290 deletions(-)

diff --git a/plugins/mqtt-xinao/mqtt_handle.c b/plugins/mqtt-xinao/mqtt_handle.c
index dfc3964c..8cd0c3c3 100644
--- a/plugins/mqtt-xinao/mqtt_handle.c
+++ b/plugins/mqtt-xinao/mqtt_handle.c
@@ -114,21 +114,6 @@ static char *generate_read_resp_json(neu_plugin_t *         plugin,
     return json_str;
 }
 
-static char *generate_write_resp_json(neu_plugin_t *    plugin,
-                                      neu_json_mqtt_t * mqtt,
-                                      neu_resp_error_t *data)
-{
-    (void) plugin;
-
-    neu_json_error_resp_t error    = { .error = data->error };
-    char *                json_str = NULL;
-
-    neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, mqtt,
-                              neu_json_encode_mqtt_resp, &json_str);
-
-    return json_str;
-}
-
 static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
                                 neu_json_read_req_t *req)
 {
@@ -153,121 +138,6 @@ static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
     return 0;
 }
 
-static int json_value_to_tag_value(union neu_json_value *req,
-                                   enum neu_json_type t, neu_dvalue_t *value)
-{
-    switch (t) {
-    case NEU_JSON_INT:
-        value->type      = NEU_TYPE_INT64;
-        value->value.u64 = req->val_int;
-        break;
-    case NEU_JSON_STR:
-        value->type = NEU_TYPE_STRING;
-        strncpy(value->value.str, req->val_str, sizeof(value->value.str));
-        break;
-    case NEU_JSON_DOUBLE:
-        value->type      = NEU_TYPE_DOUBLE;
-        value->value.d64 = req->val_double;
-        break;
-    case NEU_JSON_BOOL:
-        value->type          = NEU_TYPE_BOOL;
-        value->value.boolean = req->val_bool;
-        break;
-    case NEU_JSON_BYTES:
-        value->type               = NEU_TYPE_BYTES;
-        value->value.bytes.length = req->val_bytes.length;
-        memcpy(value->value.bytes.bytes, req->val_bytes.bytes,
-               req->val_bytes.length);
-        break;
-    default:
-        return -1;
-    }
-    return 0;
-}
-
-static int send_write_tag_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
-                              neu_json_write_req_t *req)
-{
-    plog_notice(plugin, "write tag uuid:%s, group:%s, node:%s", mqtt->uuid,
-                req->group, req->node);
-
-    neu_reqresp_head_t  header = { 0 };
-    neu_req_write_tag_t cmd    = { 0 };
-
-    header.ctx  = mqtt;
-    header.type = NEU_REQ_WRITE_TAG;
-
-    cmd.driver = req->node;
-    cmd.group  = req->group;
-    cmd.tag    = req->tag;
-
-    if (0 != json_value_to_tag_value(&req->value, req->t, &cmd.value)) {
-        plog_error(plugin, "invalid tag value type: %d", req->t);
-        return -1;
-    }
-
-    if (0 != neu_plugin_op(plugin, header, &cmd)) {
-        plog_error(plugin, "neu_plugin_op(NEU_REQ_WRITE_TAG) fail");
-        return -1;
-    }
-
-    req->node  = NULL; // ownership moved
-    req->group = NULL; // ownership moved
-    req->tag   = NULL; // ownership moved
-    return 0;
-}
-
-static int send_write_tags_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
-                               neu_json_write_tags_req_t *req)
-{
-    plog_notice(plugin, "write tags uuid:%s, group:%s, node:%s", mqtt->uuid,
-                req->group, req->node);
-
-    for (int i = 0; i < req->n_tag; i++) {
-        if (req->tags[i].t == NEU_JSON_STR) {
-            if (strlen(req->tags[i].value.val_str) >= NEU_VALUE_SIZE) {
-                return -1;
-            }
-        }
-    }
-
-    neu_reqresp_head_t header = {
-        .ctx  = mqtt,
-        .type = NEU_REQ_WRITE_TAGS,
-    };
-
-    neu_req_write_tags_t cmd = { 0 };
-    cmd.driver               = req->node;
-    cmd.group                = req->group;
-    cmd.n_tag                = req->n_tag;
-    cmd.tags                 = calloc(cmd.n_tag, sizeof(neu_resp_tag_value_t));
-    if (NULL == cmd.tags) {
-        return -1;
-    }
-
-    for (int i = 0; i < cmd.n_tag; i++) {
-        strcpy(cmd.tags[i].tag, req->tags[i].tag);
-        if (0 !=
-            json_value_to_tag_value(&req->tags[i].value, req->tags[i].t,
-                                    &cmd.tags[i].value)) {
-            plog_error(plugin, "invalid tag value type: %d", req->tags[i].t);
-            free(cmd.tags);
-            return -1;
-        }
-    }
-
-    if (0 != neu_plugin_op(plugin, header, &cmd)) {
-        plog_error(plugin, "neu_plugin_op(NEU_REQ_WRITE_TAGS) fail");
-        free(cmd.tags);
-        return -1;
-    }
-
-    req->node  = NULL; // ownership moved
-    req->group = NULL; // ownership moved
-
-    return 0;
-}
-
 static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic,
                        uint8_t *payload, uint32_t len, void *data)
 {
@@ -310,97 +180,6 @@ static inline int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
 
 void handle_write_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_write_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_write(json_str, &req);
-    if (0 != rv) {
-        plog_error(plugin, "neu_json_decode_write fail");
-        neu_json_decode_mqtt_req_free(mqtt);
-        free(json_str);
-        return;
-    }
-
-    if (req->singular) {
-        rv = send_write_tag_req(plugin, mqtt, &req->single);
-    } else {
-        rv = send_write_tags_req(plugin, mqtt, &req->plural);
-    }
-    if (0 != rv) {
-        neu_json_decode_mqtt_req_free(mqtt);
-    }
-
-    neu_json_decode_write_free(req);
-    free(json_str);
-}
-
-int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
-                          neu_resp_error_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_write_resp_json(plugin, mqtt_json, data);
-    if (NULL == json_str) {
-        plog_error(plugin, "generate write resp json fail, uuid:%s",
-                   mqtt_json->uuid);
-        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:
-    neu_json_decode_mqtt_req_free(mqtt_json);
-    return rv;
-}
-
-void handle_write_batch_req(neu_mqtt_qos_e qos, const char *topic,
-                            const uint8_t *payload, uint32_t len, void *data)
 {
     neu_plugin_t *plugin = data;
 
@@ -422,34 +201,77 @@ void handle_write_batch_req(neu_mqtt_qos_e qos, const char *topic,
 
     neu_reqresp_head_t header = {
         .ctx  = ctx,
-        .type = NEU_REQ_WRITE_BATCH,
+        .type = ctx->type,
     };
 
-    neu_reqresp_write_batch_t cmd = {
-        .batch = ctx->batch,
-    };
+    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(NEU_REQ_WRITE_BATCH) fail");
+    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_batch_resp(neu_plugin_t *plugin, xinao_ctx_t *ctx)
+int handle_write_resp(neu_plugin_t *plugin, xinao_ctx_t *ctx, void *data)
 {
-    char *json_str = xinao_ctx_gen_resp_json(ctx);
-    xinao_ctx_free(ctx);
+    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 write batch resp json fail");
-        return NEU_ERR_EINTERNAL;
+        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;
-    int            rv = publish(plugin, qos, topic, json_str, strlen(json_str));
-    json_str          = NULL;
+    rv       = publish(plugin, qos, topic, json_str, strlen(json_str));
+    json_str = NULL;
+
+end:
+    xinao_ctx_free(ctx);
     return rv;
 }
 
diff --git a/plugins/mqtt-xinao/mqtt_handle.h b/plugins/mqtt-xinao/mqtt_handle.h
index b383cb0e..e0af3500 100644
--- a/plugins/mqtt-xinao/mqtt_handle.h
+++ b/plugins/mqtt-xinao/mqtt_handle.h
@@ -32,12 +32,7 @@ extern "C" {
 void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
                       const uint8_t *payload, uint32_t len, void *data);
 
-int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
-                          neu_resp_error_t *data);
-
-void handle_write_batch_req(neu_mqtt_qos_e qos, const char *topic,
-                            const uint8_t *payload, uint32_t len, void *data);
-int  handle_write_batch_resp(neu_plugin_t *plugin, xinao_ctx_t *ctx);
+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);
diff --git a/plugins/mqtt-xinao/mqtt_plugin.c b/plugins/mqtt-xinao/mqtt_plugin.c
index c2f25982..3a14eca8 100644
--- a/plugins/mqtt-xinao/mqtt_plugin.c
+++ b/plugins/mqtt-xinao/mqtt_plugin.c
@@ -221,7 +221,7 @@ static int subscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
     if (0 !=
         neu_mqtt_client_subscribe(plugin->client, config->qos,
                                   config->write_req_topic, plugin,
-                                  handle_write_batch_req)) {
+                                  handle_write_req)) {
         plog_error(plugin, "subscribe [%s] fail",
                    plugin->config.write_req_topic);
         return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
@@ -360,10 +360,10 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
 
     switch (head->type) {
     case NEU_RESP_ERROR:
-        // error = handle_write_response(plugin, head->ctx, data);
+        error = handle_write_resp(plugin, head->ctx, data);
         break;
     case NEU_RESP_WRITE_BATCH:
-        error = handle_write_batch_resp(plugin, head->ctx);
+        error = handle_write_resp(plugin, head->ctx, data);
         break;
     case NEU_RESP_READ_GROUP:
         error = handle_read_response(plugin, head->ctx, data);
diff --git a/plugins/mqtt-xinao/xinao.c b/plugins/mqtt-xinao/xinao.c
index 39f15d47..17a836b7 100644
--- a/plugins/mqtt-xinao/xinao.c
+++ b/plugins/mqtt-xinao/xinao.c
@@ -66,48 +66,70 @@ static inline int json_value_to_tag_value(neu_dvalue_t *          value,
     return 0;
 }
 
-int neu_json_decode_write_batch_req_json(neu_plugin_t *plugin, void *json_root,
-                                         neu_write_batch_t **batch_p,
-                                         index_map_t *       map_p)
+int neu_json_decode_xinao_write_tag_req_json(neu_plugin_t *    plugin,
+                                             void *            json_root,
+                                             xinao_wtag_ctx_t *ctx)
 {
-    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`");
+    json_t *data = json_object_get(json_root, "data");
+    if (NULL == data || !json_is_object(data)) {
+        plog_error(plugin, "no `data` object");
         return -1;
     }
 
-    if (NULL == pKey || !json_is_string(pKey)) {
-        plog_error(plugin, "no valid key `pKey`");
+    json_t *node_name = json_object_get(data, "sysId");
+    if (NULL == node_name || !json_is_string(node_name)) {
+        plog_error(plugin, "data no valid key `sysId`");
         return -1;
     }
 
-    if (NULL == sn || !json_is_string(sn)) {
-        plog_error(plugin, "no valid key `sn`");
+    json_t *group_name = json_object_get(data, "dev");
+    if (NULL == group_name || !json_is_string(group_name)) {
+        plog_error(plugin, "data no valid key `dev`");
         return -1;
     }
 
-    if (NULL == type || !json_is_string(type) ||
-        0 != strcmp("algset", json_string_value(type))) {
-        plog_error(plugin, "no valid key `type`");
+    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(node_name), json_string_value(group_name),
+                   json_string_value(tag));
         return -1;
     }
 
-    if (NULL == seq || !json_is_integer(seq)) {
-        plog_error(plugin, "no valid key `seq`");
+    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;
     }
 
-    if (NULL == ts || !json_is_integer(ts)) {
-        plog_error(plugin, "no valid key `ts`");
+    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");
@@ -261,8 +283,9 @@ int neu_json_decode_write_batch_req_json(neu_plugin_t *plugin, void *json_root,
         }
     }
 
-    *batch_p = batch;
-    *map_p   = map;
+    ctx->type        = NEU_REQ_WRITE_BATCH;
+    ctx->batch.batch = batch;
+    ctx->map         = map;
     return 0;
 
 error:
@@ -272,6 +295,78 @@ error:
     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)
 {
@@ -281,26 +376,40 @@ xinao_ctx_t *xinao_ctx_from_json_buf(neu_plugin_t *plugin, char *buf,
         return NULL;
     }
 
-    xinao_ctx_t *ctx = calloc(1, sizeof(*ctx));
-    if (NULL == ctx) {
+    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;
     }
 
-    if (0 !=
-        neu_json_decode_write_batch_req_json(plugin, root, &ctx->batch,
-                                             &ctx->map)) {
-        plog_error(plugin, "decode batch req json fail");
-        neu_json_decode_free(root);
-        free(ctx);
-        return NULL;
-    }
-
     ctx->root = root;
     return ctx;
 }
 
-char *xinao_ctx_gen_resp_json(xinao_ctx_t *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);
@@ -317,8 +426,9 @@ char *xinao_ctx_gen_resp_json(xinao_ctx_t *ctx)
         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->drivers[n->ind];
-        neu_write_batch_group_t * batch_group  = &batch_driver->groups[g->ind];
+        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);
diff --git a/plugins/mqtt-xinao/xinao.h b/plugins/mqtt-xinao/xinao.h
index b6ad3659..787affbf 100644
--- a/plugins/mqtt-xinao/xinao.h
+++ b/plugins/mqtt-xinao/xinao.h
@@ -48,9 +48,21 @@ typedef struct {
 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;
-    index_map_t        map;
-    neu_write_batch_t *batch;
 } xinao_ctx_t;
 
 static inline void index_map_free(index_map_t *map)
@@ -101,20 +113,33 @@ static inline int index_map_add(index_map_t *map, const char *node,
     return 0;
 }
 
-int neu_json_decode_write_batch_req_json(neu_plugin_t *plugin, void *json_root,
-                                         neu_write_batch_t **batch_p,
-                                         index_map_t *       map_p);
+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_resp_json(xinao_ctx_t *ctx);
+
+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);
-        index_map_free(&ctx->map);
-        neu_write_batch_free(ctx->batch);
+        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);
     }
 }