From fadca0f2c99714aa1034b4a6fbd3b92187e90ecd Mon Sep 17 00:00:00 2001 From: fengzero Date: Fri, 9 Jun 2023 02:19:24 +0000 Subject: [PATCH] feat(core): support writing multiple tag in one request --- include/neuron/adapter.h | 12 + include/neuron/errcodes.h | 1 + include/neuron/json/neu_json_rw.h | 16 ++ include/neuron/plugin.h | 8 + plugins/modbus/modbus_qh_tcp.c | 1 + plugins/modbus/modbus_rtu.c | 1 + plugins/modbus/modbus_tcp.c | 1 + plugins/restful/handle.c | 9 + plugins/restful/rw_handle.c | 96 +++++-- plugins/restful/rw_handle.h | 1 + src/adapter/adapter.c | 20 ++ src/adapter/driver/driver.c | 386 ++++++++++++++++++--------- src/adapter/driver/driver_internal.h | 2 + src/core/manager.c | 1 + src/parser/neu_json_rw.c | 90 +++++++ src/utils/http.c | 1 + 16 files changed, 499 insertions(+), 147 deletions(-) diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 3cd9e26a..c017bb58 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -43,6 +43,7 @@ typedef enum neu_reqresp_type { NEU_REQ_READ_GROUP, NEU_RESP_READ_GROUP, NEU_REQ_WRITE_TAG, + NEU_REQ_WRITE_TAGS, NEU_REQ_SUBSCRIBE_GROUP, NEU_REQ_UNSUBSCRIBE_GROUP, @@ -144,6 +145,7 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQ_READ_GROUP] = "NEU_REQ_READ_GROUP", [NEU_RESP_READ_GROUP] = "NEU_RESP_READ_GROUP", [NEU_REQ_WRITE_TAG] = "NEU_REQ_WRITE_TAG", + [NEU_REQ_WRITE_TAGS] = "NEU_REQ_WRITE_TAGS", [NEU_REQ_SUBSCRIBE_GROUP] = "NEU_REQ_SUBSCRIBE_GROUP", [NEU_REQ_UNSUBSCRIBE_GROUP] = "NEU_REQ_UNSUBSCRIBE_GROUP", @@ -634,6 +636,14 @@ typedef struct neu_resp_tag_value { neu_dvalue_t value; } neu_resp_tag_value_t; +typedef struct neu_req_write_tags { + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; + + int n_tag; + neu_resp_tag_value_t *tags; +} neu_req_write_tags_t; + typedef struct { char driver[NEU_NODE_NAME_LEN]; char group[NEU_GROUP_NAME_LEN]; @@ -783,6 +793,8 @@ typedef struct adapter_callbacks { const char *tag, neu_dvalue_t value); void (*write_response)(neu_adapter_t *adapter, void *req, int error); + void (*update_im)(neu_adapter_t *adapter, const char *group, + const char *tag, neu_dvalue_t value); } driver; }; } adapter_callbacks_t; diff --git a/include/neuron/errcodes.h b/include/neuron/errcodes.h index 6d83e2bf..742d183b 100644 --- a/include/neuron/errcodes.h +++ b/include/neuron/errcodes.h @@ -117,6 +117,7 @@ typedef enum { NEU_ERR_PLUGIN_NOT_FOUND = 3014, NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE = 3015, NEU_ERR_PLUGIN_NOT_SUPPORT_TEMPLATE = 3016, + NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS = 3017, NEU_ERR_MQTT_FAILURE = 4000, NEU_ERR_MQTT_NO_CERTFILESET = 4001, diff --git a/include/neuron/json/neu_json_rw.h b/include/neuron/json/neu_json_rw.h index 24a79bc3..b5cb41cb 100644 --- a/include/neuron/json/neu_json_rw.h +++ b/include/neuron/json/neu_json_rw.h @@ -64,6 +64,22 @@ typedef struct { int neu_json_decode_write_req(char *buf, neu_json_write_req_t **result); void neu_json_decode_write_req_free(neu_json_write_req_t *req); +typedef struct { + char * tag; + enum neu_json_type t; + union neu_json_value value; +} neu_json_write_tags_elem_t; + +typedef struct { + char * group; + char * node; + int n_tag; + neu_json_write_tags_elem_t *tags; +} neu_json_write_tags_req_t; +int neu_json_decode_write_tags_req(char * buf, + neu_json_write_tags_req_t **result); +void neu_json_decode_write_tags_req_free(neu_json_write_tags_req_t *req); + typedef struct { char *group; char *node; diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index 5abdde59..f46e5abc 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -64,6 +64,11 @@ struct neu_plugin_group { typedef int (*neu_plugin_tag_validator_t)(const neu_datatag_t *tag); +typedef struct { + neu_datatag_t *tag; + neu_value_u value; +} neu_plugin_tag_value_t; + typedef struct neu_plugin_intf_funs { neu_plugin_t *(*open)(void); int (*close)(neu_plugin_t *plugin); @@ -81,6 +86,9 @@ typedef struct neu_plugin_intf_funs { int (*group_timer)(neu_plugin_t *plugin, neu_plugin_group_t *group); int (*write_tag)(neu_plugin_t *plugin, void *req, neu_datatag_t *tag, neu_value_u value); + int (*write_tags)( + neu_plugin_t *plugin, void *req, + UT_array *tag_values); // UT_array {neu_datatag_t, neu_value_u} neu_plugin_tag_validator_t tag_validator; } driver; }; diff --git a/plugins/modbus/modbus_qh_tcp.c b/plugins/modbus/modbus_qh_tcp.c index df1c4b1b..0b92012e 100644 --- a/plugins/modbus/modbus_qh_tcp.c +++ b/plugins/modbus/modbus_qh_tcp.c @@ -57,6 +57,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = { .driver.group_timer = driver_group_timer, .driver.write_tag = driver_write, .driver.tag_validator = driver_tag_validator, + .driver.write_tags = NULL, }; const neu_plugin_module_t neu_plugin_module = { diff --git a/plugins/modbus/modbus_rtu.c b/plugins/modbus/modbus_rtu.c index 6d97da9d..804ea9c6 100644 --- a/plugins/modbus/modbus_rtu.c +++ b/plugins/modbus/modbus_rtu.c @@ -57,6 +57,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = { .driver.group_timer = driver_group_timer, .driver.write_tag = driver_write, .driver.tag_validator = driver_tag_validator, + .driver.write_tags = NULL, }; const neu_plugin_module_t neu_plugin_module = { diff --git a/plugins/modbus/modbus_tcp.c b/plugins/modbus/modbus_tcp.c index fad07bf1..5e164ea0 100644 --- a/plugins/modbus/modbus_tcp.c +++ b/plugins/modbus/modbus_tcp.c @@ -57,6 +57,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = { .driver.group_timer = driver_group_timer, .driver.write_tag = driver_write, .driver.tag_validator = driver_tag_validator, + .driver.write_tags = NULL, }; const neu_plugin_module_t neu_plugin_module = { diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index f12c1b4f..c086e03a 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -74,6 +74,9 @@ static struct neu_http_handler cors_handler[] = { { .url = "/api/v2/write", }, + { + .url = "/api/v2/write/tags", + }, { .url = "/api/v2/subscribe", }, @@ -277,6 +280,12 @@ static struct neu_http_handler rest_handlers[] = { .url = "/api/v2/write", .value.handler = handle_write, }, + { + .method = NEU_HTTP_METHOD_POST, + .type = NEU_HTTP_HANDLER_FUNCTION, + .url = "/api/v2/write/tags", + .value.handler = handle_write_tags, + }, { .method = NEU_HTTP_METHOD_POST, .type = NEU_HTTP_HANDLER_FUNCTION, diff --git a/plugins/restful/rw_handle.c b/plugins/restful/rw_handle.c index 8c5b0a8d..62255a86 100644 --- a/plugins/restful/rw_handle.c +++ b/plugins/restful/rw_handle.c @@ -61,6 +61,15 @@ void handle_write(nng_aio *aio) neu_reqresp_head_t header = { 0 }; neu_req_write_tag_t cmd = { 0 }; + if (req->t == NEU_JSON_STR && + strlen(req->value.val_str) >= NEU_VALUE_SIZE) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, { + neu_http_response(aio, NEU_ERR_STRING_TOO_LONG, + result_error); + }); + return; + } + header.ctx = aio; header.type = NEU_REQ_WRITE_TAG; @@ -74,13 +83,8 @@ void handle_write(nng_aio *aio) cmd.value.value.u64 = req->value.val_int; break; case NEU_JSON_STR: - if (strlen(req->value.val_str) >= NEU_VALUE_SIZE) { - cmd.value.type = NEU_TYPE_ERROR; - cmd.value.value.i32 = NEU_ERR_STRING_TOO_LONG; - } else { - cmd.value.type = NEU_TYPE_STRING; - strcpy(cmd.value.value.str, req->value.val_str); - } + cmd.value.type = NEU_TYPE_STRING; + strcpy(cmd.value.value.str, req->value.val_str); break; case NEU_JSON_DOUBLE: cmd.value.type = NEU_TYPE_DOUBLE; @@ -95,19 +99,77 @@ void handle_write(nng_aio *aio) break; } - if (cmd.value.type == NEU_TYPE_ERROR) { - NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, { - neu_http_response(aio, NEU_ERR_STRING_TOO_LONG, - result_error); + int ret = neu_plugin_op(plugin, header, &cmd); + if (ret != 0) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { + neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); }); - } else { - int ret = neu_plugin_op(plugin, header, &cmd); - if (ret != 0) { - NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { - neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); - }); + } + }) +} + +void handle_write_tags(nng_aio *aio) +{ + neu_plugin_t *plugin = neu_rest_get_plugin(); + + NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT( + aio, neu_json_write_tags_req_t, neu_json_decode_write_tags_req, { + neu_reqresp_head_t header = { 0 }; + neu_req_write_tags_t cmd = { 0 }; + + 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) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, { + neu_http_response(aio, NEU_ERR_STRING_TOO_LONG, + result_error); + }); + return; + } } } + + header.ctx = aio; + header.type = NEU_REQ_WRITE_TAGS; + + strcpy(cmd.driver, req->node); + strcpy(cmd.group, req->group); + cmd.n_tag = req->n_tag; + cmd.tags = calloc(cmd.n_tag, sizeof(neu_resp_tag_value_t)); + + for (int i = 0; i < cmd.n_tag; i++) { + strcpy(cmd.tags[i].tag, req->tags[i].tag); + switch (req->tags[i].t) { + case NEU_JSON_INT: + cmd.tags[i].value.type = NEU_TYPE_INT64; + cmd.tags[i].value.value.u64 = req->tags[i].value.val_int; + break; + case NEU_JSON_STR: + cmd.tags[i].value.type = NEU_TYPE_STRING; + strcpy(cmd.tags[i].value.value.str, + req->tags[i].value.val_str); + break; + case NEU_JSON_DOUBLE: + cmd.tags[i].value.type = NEU_TYPE_DOUBLE; + cmd.tags[i].value.value.d64 = req->tags[i].value.val_double; + break; + case NEU_JSON_BOOL: + cmd.tags[i].value.type = NEU_TYPE_BOOL; + cmd.tags[i].value.value.boolean = + req->tags[i].value.val_bool; + break; + default: + assert(false); + break; + } + } + + int ret = neu_plugin_op(plugin, header, &cmd); + if (ret != 0) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { + neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); + }); + } }) } diff --git a/plugins/restful/rw_handle.h b/plugins/restful/rw_handle.h index e8a399a1..b3c27ed1 100644 --- a/plugins/restful/rw_handle.h +++ b/plugins/restful/rw_handle.h @@ -28,6 +28,7 @@ void handle_rw_init(); void handle_rw_uninit(); void handle_read(nng_aio *aio); void handle_write(nng_aio *aio); +void handle_write_tags(nng_aio *aio); void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp); #endif \ No newline at end of file diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 71d556f0..1b7ec23d 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -330,6 +330,7 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, strcpy(header.sender, adapter->name); switch (header.type) { case NEU_REQ_READ_GROUP: + case NEU_REQ_WRITE_TAGS: case NEU_REQ_WRITE_TAG: { neu_req_read_group_t *cmd = (neu_req_read_group_t *) data; strcpy(header.receiver, cmd->driver); @@ -512,6 +513,22 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } + case NEU_REQ_WRITE_TAGS: { + neu_resp_error_t error = { 0 }; + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + error.error = NEU_ERR_GROUP_NOT_ALLOW; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(adapter, header, &error); + } else { + void *msg_dump = calloc(nng_msg_len(msg), 1); + memcpy(msg_dump, header, nng_msg_len(msg)); + neu_adapter_driver_write_tags((neu_adapter_driver_t *) adapter, + msg_dump); + } + 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 }; @@ -1436,6 +1453,9 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) case NEU_REQ_WRITE_TAG: data_size = sizeof(neu_req_write_tag_t); break; + case NEU_REQ_WRITE_TAGS: + data_size = sizeof(neu_req_write_tags_t); + break; case NEU_RESP_READ_GROUP: data_size = sizeof(neu_resp_read_group_t); break; diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 455676d8..71708dd7 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -39,8 +39,10 @@ #include "tag.h" typedef struct to_be_write_tag { + bool single; neu_datatag_t *tag; neu_value_u value; + UT_array * tvs; void * req; } to_be_write_tag_t; @@ -96,6 +98,15 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error) free(req); } +static void update_im(neu_adapter_t *adapter, const char *group, + const char *tag, neu_dvalue_t value) +{ + (void) adapter; + (void) group; + (void) tag; + (void) value; +} + static void update(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value) { @@ -146,6 +157,7 @@ neu_adapter_driver_t *neu_adapter_driver_create() driver->driver_events = neu_event_new(); driver->adapter.cb_funs.driver.update = update; driver->adapter.cb_funs.driver.write_response = write_response; + driver->adapter.cb_funs.driver.update_im = update_im; return driver; } @@ -286,6 +298,223 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, driver->adapter.cb_funs.response(&driver->adapter, req, &resp); } +static void fix_value(neu_datatag_t *tag, neu_type_e value_type, + neu_dvalue_t *value) +{ + switch (tag->type) { + case NEU_TYPE_BOOL: + case NEU_TYPE_STRING: + break; + case NEU_TYPE_BIT: + value->type = NEU_TYPE_BIT; + value->value.u8 = (uint8_t) value->value.u64; + break; + case NEU_TYPE_UINT8: + case NEU_TYPE_INT8: + value->type = NEU_TYPE_UINT8; + value->value.u8 = (uint8_t) value->value.u64; + break; + case NEU_TYPE_INT16: + case NEU_TYPE_UINT16: + case NEU_TYPE_WORD: + value->type = NEU_TYPE_UINT16; + value->value.u16 = (uint16_t) value->value.u64; + if (tag->option.value16.endian == NEU_DATATAG_ENDIAN_B16) { + value->value.u16 = htons(value->value.u16); + } + break; + case NEU_TYPE_UINT32: + case NEU_TYPE_INT32: + case NEU_TYPE_DWORD: + value->type = NEU_TYPE_UINT32; + value->value.u32 = (uint32_t) value->value.u64; + switch (tag->option.value32.endian) { + case NEU_DATATAG_ENDIAN_LB32: + neu_ntohs_p((uint16_t *) value->value.bytes); + neu_ntohs_p((uint16_t *) value->value.bytes + 2); + break; + case NEU_DATATAG_ENDIAN_BB32: + value->value.u32 = htonl(value->value.u32); + break; + case NEU_DATATAG_ENDIAN_BL32: + value->value.u32 = htonl(value->value.u32); + neu_ntohs_p((uint16_t *) value->value.bytes); + neu_ntohs_p((uint16_t *) value->value.bytes + 2); + break; + case NEU_DATATAG_ENDIAN_LL32: + default: + break; + } + break; + case NEU_TYPE_FLOAT: + if (value_type != NEU_TYPE_FLOAT) { + value->type = NEU_TYPE_FLOAT; + value->value.f32 = (float) value->value.d64; + } + + switch (tag->option.value32.endian) { + case NEU_DATATAG_ENDIAN_LB32: + neu_ntohs_p((uint16_t *) value->value.bytes); + neu_ntohs_p((uint16_t *) value->value.bytes + 2); + break; + case NEU_DATATAG_ENDIAN_BB32: + value->value.u32 = htonl(value->value.u32); + break; + case NEU_DATATAG_ENDIAN_BL32: + value->value.u32 = htonl(value->value.u32); + neu_ntohs_p((uint16_t *) value->value.bytes); + neu_ntohs_p((uint16_t *) value->value.bytes + 2); + break; + case NEU_DATATAG_ENDIAN_LL32: + default: + break; + } + + break; + case NEU_TYPE_DOUBLE: + case NEU_TYPE_UINT64: + case NEU_TYPE_INT64: + case NEU_TYPE_LWORD: + switch (tag->option.value64.endian) { + case NEU_DATATAG_ENDIAN_B64: + value->value.u64 = neu_ntohll(value->value.u64); + break; + case NEU_DATATAG_ENDIAN_L64: + default: + break; + } + break; + default: + assert(false); + break; + } +} + +static void cal_decimal(neu_type_e tag_type, neu_type_e value_type, + neu_value_u *value, double decimal) +{ + switch (tag_type) { + case NEU_TYPE_INT8: + case NEU_TYPE_UINT8: + case NEU_TYPE_INT16: + case NEU_TYPE_UINT16: + case NEU_TYPE_INT32: + case NEU_TYPE_UINT32: + case NEU_TYPE_INT64: + case NEU_TYPE_UINT64: + case NEU_TYPE_WORD: + case NEU_TYPE_LWORD: + case NEU_TYPE_DWORD: + if (value_type == NEU_TYPE_INT64 || NEU_TYPE_LWORD) { + value->u64 = value->u64 / decimal; + } else if (value_type == NEU_TYPE_DOUBLE) { + value->u64 = (uint64_t) round(value->d64 / decimal); + } + break; + case NEU_TYPE_FLOAT: + case NEU_TYPE_DOUBLE: + if (value_type == NEU_TYPE_INT64) { + value->d64 = (double) (value->d64 / decimal); + } else if (value_type == NEU_TYPE_DOUBLE) { + value->d64 = value->d64 / decimal; + } else if (value_type == NEU_TYPE_FLOAT) { + value->f32 = value->f32 / decimal; + } + break; + default: + break; + } +} + +static void free_tags(neu_req_write_tags_t *cmd) +{ + free(cmd->tags); +} + +void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver, + neu_reqresp_head_t * req) +{ + neu_req_write_tags_t *cmd = (neu_req_write_tags_t *) &req[1]; + + if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { + driver->adapter.cb_funs.driver.write_response( + &driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING); + free_tags(cmd); + return; + } + + if (driver->adapter.module->intf_funs->driver.write_tags == NULL) { + free_tags(cmd); + driver->adapter.cb_funs.driver.write_response( + &driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS); + return; + } + + group_t *g = find_group(driver, cmd->group); + + if (g == NULL) { + neu_resp_error_t error = { .error = NEU_ERR_GROUP_NOT_EXIST }; + req->type = NEU_RESP_ERROR; + free_tags(cmd); + driver->adapter.cb_funs.response(&driver->adapter, req, &error); + free(req); + return; + } + + UT_array *tags = NULL; + UT_icd icd = { sizeof(neu_plugin_tag_value_t), NULL, NULL, NULL }; + utarray_new(tags, &icd); + for (int i = 0; i < cmd->n_tag; i++) { + neu_plugin_tag_value_t tv = { 0 }; + + neu_datatag_t *tag = neu_group_find_tag(g->group, cmd->tags[i].tag); + if (tag != NULL && neu_tag_attribute_test(tag, NEU_ATTRIBUTE_WRITE) && + neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) == false) { + tv.tag = neu_tag_dup(tag); + + if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) { + if (cmd->tags[i].value.type == NEU_TYPE_INT64) { + tv.value.d64 = (double) tv.value.u64; + } + } + if (tag->decimal != 0) { + cal_decimal(tag->type, cmd->tags[i].value.type, + &cmd->tags[i].value.value, tag->decimal); + } + fix_value(tag, cmd->tags[i].value.type, &cmd->tags[i].value); + + tv.value = cmd->tags[i].value.value; + utarray_push_back(tags, &tv); + } else { + if (tag != NULL) { + neu_tag_free(tag); + } + } + } + + if (utarray_len(tags) != (unsigned int) cmd->n_tag) { + neu_resp_error_t error = { .error = NEU_ERR_TAG_NOT_EXIST }; + req->type = NEU_RESP_ERROR; + free_tags(cmd); + driver->adapter.cb_funs.response(&driver->adapter, req, &error); + free(req); + utarray_foreach(tags, neu_plugin_tag_value_t *, tv) + { + neu_tag_free(tv->tag); + } + utarray_free(tags); + return; + } + + to_be_write_tag_t wtag = { 0 }; + wtag.single = false; + wtag.req = (void *) req; + wtag.tvs = tags; + + store_write_tag(g, &wtag); + free_tags(cmd); +} + void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, neu_reqresp_head_t * req) { @@ -327,133 +556,11 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, } if (tag->decimal != 0) { - switch (tag->type) { - case NEU_TYPE_INT8: - case NEU_TYPE_UINT8: - case NEU_TYPE_INT16: - case NEU_TYPE_UINT16: - case NEU_TYPE_INT32: - case NEU_TYPE_UINT32: - case NEU_TYPE_INT64: - case NEU_TYPE_UINT64: - case NEU_TYPE_WORD: - case NEU_TYPE_LWORD: - case NEU_TYPE_DWORD: - if (cmd->value.type == NEU_TYPE_INT64 || - cmd->value.type == NEU_TYPE_LWORD) { - cmd->value.value.u64 = cmd->value.value.u64 / tag->decimal; - } else if (cmd->value.type == NEU_TYPE_DOUBLE) { - cmd->value.value.u64 = - (uint64_t) round(cmd->value.value.d64 / tag->decimal); - } - break; - case NEU_TYPE_FLOAT: - case NEU_TYPE_DOUBLE: - if (cmd->value.type == NEU_TYPE_INT64) { - cmd->value.value.d64 = - (double) (cmd->value.value.d64 / tag->decimal); - } else if (cmd->value.type == NEU_TYPE_DOUBLE) { - cmd->value.value.d64 = cmd->value.value.d64 / tag->decimal; - } else if (cmd->value.type == NEU_TYPE_FLOAT) { - cmd->value.value.f32 = cmd->value.value.f32 / tag->decimal; - } - break; - default: - break; - } + cal_decimal(tag->type, cmd->value.type, &cmd->value.value, + tag->decimal); } - switch (tag->type) { - case NEU_TYPE_BOOL: - case NEU_TYPE_STRING: - break; - case NEU_TYPE_BIT: - cmd->value.type = NEU_TYPE_BIT; - cmd->value.value.u8 = (uint8_t) cmd->value.value.u64; - break; - case NEU_TYPE_UINT8: - case NEU_TYPE_INT8: - cmd->value.type = NEU_TYPE_UINT8; - cmd->value.value.u8 = (uint8_t) cmd->value.value.u64; - break; - case NEU_TYPE_INT16: - case NEU_TYPE_UINT16: - case NEU_TYPE_WORD: - cmd->value.type = NEU_TYPE_UINT16; - cmd->value.value.u16 = (uint16_t) cmd->value.value.u64; - switch (tag->option.value16.endian) { - case NEU_DATATAG_ENDIAN_B16: - cmd->value.value.u16 = htons(cmd->value.value.u16); - break; - case NEU_DATATAG_ENDIAN_L16: - default: - break; - } - break; - case NEU_TYPE_UINT32: - case NEU_TYPE_INT32: - case NEU_TYPE_DWORD: - cmd->value.type = NEU_TYPE_UINT32; - cmd->value.value.u32 = (uint32_t) cmd->value.value.u64; - switch (tag->option.value32.endian) { - case NEU_DATATAG_ENDIAN_LB32: - neu_ntohs_p((uint16_t *) cmd->value.value.bytes); - neu_ntohs_p((uint16_t *) (cmd->value.value.bytes + 2)); - break; - case NEU_DATATAG_ENDIAN_BB32: - cmd->value.value.u32 = htonl(cmd->value.value.u32); - break; - case NEU_DATATAG_ENDIAN_BL32: - cmd->value.value.u32 = htonl(cmd->value.value.u32); - neu_ntohs_p((uint16_t *) cmd->value.value.bytes); - neu_ntohs_p((uint16_t *) (cmd->value.value.bytes + 2)); - break; - case NEU_DATATAG_ENDIAN_LL32: - default: - break; - } - break; - case NEU_TYPE_FLOAT: - if (NEU_TYPE_FLOAT != cmd->value.type) { - cmd->value.value.f32 = (float) cmd->value.value.d64; - cmd->value.type = NEU_TYPE_FLOAT; - } - - switch (tag->option.value32.endian) { - case NEU_DATATAG_ENDIAN_LB32: - neu_ntohs_p((uint16_t *) cmd->value.value.bytes); - neu_ntohs_p((uint16_t *) (cmd->value.value.bytes + 2)); - break; - case NEU_DATATAG_ENDIAN_BB32: - cmd->value.value.u32 = htonl(cmd->value.value.u32); - break; - case NEU_DATATAG_ENDIAN_BL32: - cmd->value.value.u32 = htonl(cmd->value.value.u32); - neu_ntohs_p((uint16_t *) cmd->value.value.bytes); - neu_ntohs_p((uint16_t *) (cmd->value.value.bytes + 2)); - break; - case NEU_DATATAG_ENDIAN_LL32: - default: - break; - } - break; - case NEU_TYPE_DOUBLE: - case NEU_TYPE_UINT64: - case NEU_TYPE_INT64: - case NEU_TYPE_LWORD: - switch (tag->option.value64.endian) { - case NEU_DATATAG_ENDIAN_B64: - cmd->value.value.u64 = neu_ntohll(cmd->value.value.u64); - break; - case NEU_DATATAG_ENDIAN_L64: - default: - break; - } - break; - default: - assert(false); - break; - } + fix_value(tag, cmd->value.type, &cmd->value); if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC)) { neu_driver_cache_update(g->driver->cache, g->name, tag->name, @@ -467,6 +574,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, free(req); } else { to_be_write_tag_t wtag = { 0 }; + wtag.single = true; wtag.req = (void *) req; wtag.value = cmd->value.value; wtag.tag = neu_tag_dup(tag); @@ -593,7 +701,15 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name) utarray_foreach(find->wt_tags, to_be_write_tag_t *, tag) { - neu_tag_free(tag->tag); + if (tag->single) { + neu_tag_free(tag->tag); + } else { + utarray_foreach(tag->tvs, neu_plugin_tag_value_t *, tv) + { + neu_tag_free(tv->tag); + } + utarray_free(tag->tvs); + } } utarray_free(find->static_tags); @@ -964,10 +1080,20 @@ static int read_callback(void *usr_data) pthread_mutex_lock(&group->wt_mtx); utarray_foreach(group->wt_tags, to_be_write_tag_t *, wtag) { - group->driver->adapter.module->intf_funs->driver.write_tag( - group->driver->adapter.plugin, (void *) wtag->req, wtag->tag, - wtag->value); - neu_tag_free(wtag->tag); + if (wtag->single) { + group->driver->adapter.module->intf_funs->driver.write_tag( + group->driver->adapter.plugin, (void *) wtag->req, wtag->tag, + wtag->value); + neu_tag_free(wtag->tag); + } else { + group->driver->adapter.module->intf_funs->driver.write_tags( + group->driver->adapter.plugin, (void *) wtag->req, wtag->tvs); + utarray_foreach(wtag->tvs, neu_plugin_tag_value_t *, tv) + { + neu_tag_free(tv->tag); + } + utarray_free(wtag->tvs); + } } utarray_clear(group->wt_tags); pthread_mutex_unlock(&group->wt_mtx); diff --git a/src/adapter/driver/driver_internal.h b/src/adapter/driver/driver_internal.h index e0e584d0..9dfb272b 100644 --- a/src/adapter/driver/driver_internal.h +++ b/src/adapter/driver/driver_internal.h @@ -38,6 +38,8 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, neu_reqresp_head_t * req); +void neu_adapter_driver_write_tags(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); diff --git a/src/core/manager.c b/src/core/manager.c index e358a294..3c84d036 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -786,6 +786,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_GET_NODE_SETTING: case NEU_REQ_READ_GROUP: case NEU_REQ_WRITE_TAG: + case NEU_REQ_WRITE_TAGS: case NEU_REQ_GET_NODE_STATE: case NEU_REQ_GET_TAG: case NEU_REQ_GET_NDRIVER_TAGS: diff --git a/src/parser/neu_json_rw.c b/src/parser/neu_json_rw.c index 8afce173..86ee90ef 100644 --- a/src/parser/neu_json_rw.c +++ b/src/parser/neu_json_rw.c @@ -181,6 +181,96 @@ void neu_json_decode_write_req_free(neu_json_write_req_t *req) free(req); } +int neu_json_decode_write_tags_req(char * buf, + neu_json_write_tags_req_t **result) +{ + int ret = 0; + void * json_obj = NULL; + neu_json_write_tags_req_t *req = + calloc(1, sizeof(neu_json_write_tags_req_t)); + if (req == NULL) { + return -1; + } + + json_obj = neu_json_decode_new(buf); + neu_json_elem_t req_elems[] = { + { + .name = "node", + .t = NEU_JSON_STR, + }, + { + .name = "group", + .t = NEU_JSON_STR, + }, + { + .name = "tags", + .t = NEU_JSON_OBJECT, + }, + }; + ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems), + req_elems); + if (ret != 0) { + free(req); + neu_json_decode_free(json_obj); + return -1; + } + + req->node = req_elems[0].v.val_str; + req->group = req_elems[1].v.val_str; + + req->n_tag = neu_json_decode_array_size_by_json(json_obj, "tags"); + if (req->n_tag <= 0) { + free(req); + neu_json_decode_free(json_obj); + return -1; + } + + req->tags = calloc(req->n_tag, sizeof(neu_json_write_tags_elem_t)); + for (int i = 0; i < req->n_tag; i++) { + neu_json_elem_t v_elems[] = { + { + .name = "tag", + .t = NEU_JSON_STR, + }, + { + .name = "value", + .t = NEU_JSON_VALUE, + }, + }; + + ret = neu_json_decode_array_by_json( + json_obj, "tags", i, NEU_JSON_ELEM_SIZE(v_elems), v_elems); + if (ret != 0) { + free(req->tags); + free(req); + neu_json_decode_free(json_obj); + } + + req->tags[i].tag = v_elems[0].v.val_str; + req->tags[i].t = v_elems[1].t; + req->tags[i].value = v_elems[1].v; + } + *result = req; + + neu_json_decode_free(json_obj); + return 0; +} + +void neu_json_decode_write_tags_req_free(neu_json_write_tags_req_t *req) +{ + free(req->group); + free(req->node); + + for (int i = 0; i < req->n_tag; i++) { + free(req->tags[i].tag); + if (req->tags[i].t == NEU_JSON_STR) { + free(req->tags[i].value.val_str); + } + } + free(req->tags); + free(req); +} + int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result) { int ret = 0; diff --git a/src/utils/http.c b/src/utils/http.c index f6fb942f..27ad6651 100644 --- a/src/utils/http.c +++ b/src/utils/http.c @@ -350,6 +350,7 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content) case NEU_ERR_COMMAND_EXECUTION_FAILED: case NEU_ERR_TEMPLATE_NOT_FOUND: case NEU_ERR_PLUGIN_NOT_FOUND: + case NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS: status = NNG_HTTP_STATUS_NOT_FOUND; break; case NEU_ERR_NODE_EXIST: