mirror of
https://github.com/emqx/neuron.git
synced 2026-10-29 15:09:11 +08:00
feat(xinao): support write both tag and batch
This commit is contained in:
eeff
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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);
|
||||
|
||||
+149
-39
@@ -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);
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user