diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 2d067484..b5cd6996 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -47,6 +47,7 @@ typedef enum neu_reqresp_type { NEU_RESP_READ_GROUP, NEU_REQ_WRITE_TAG, NEU_REQ_WRITE_TAGS, + NEU_REQ_WRITE_GTAGS, NEU_REQ_SUBSCRIBE_GROUP, NEU_REQ_UNSUBSCRIBE_GROUP, @@ -154,6 +155,7 @@ static const char *neu_reqresp_type_string_t[] = { [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_WRITE_GTAGS] = "NEU_REQ_WRITE_GTAGS", [NEU_REQ_SUBSCRIBE_GROUP] = "NEU_REQ_SUBSCRIBE_GROUP", [NEU_REQ_UNSUBSCRIBE_GROUP] = "NEU_REQ_UNSUBSCRIBE_GROUP", @@ -810,6 +812,20 @@ typedef struct neu_req_write_tags { neu_resp_tag_value_t *tags; } neu_req_write_tags_t; +typedef struct { + char group[NEU_GROUP_NAME_LEN]; + + int n_tag; + neu_resp_tag_value_t *tags; +} neu_req_gtag_group_t; + +typedef struct { + char driver[NEU_NODE_NAME_LEN]; + + int n_group; + neu_req_gtag_group_t *groups; +} neu_req_write_gtags_t; + typedef struct { char driver[NEU_NODE_NAME_LEN]; char group[NEU_GROUP_NAME_LEN]; diff --git a/include/neuron/json/json.h b/include/neuron/json/json.h index aeca7768..9dcc62d6 100644 --- a/include/neuron/json/json.h +++ b/include/neuron/json/json.h @@ -78,6 +78,8 @@ void *neu_json_array(); int neu_json_decode_by_json(void *json, int size, neu_json_elem_t *ele); int neu_json_decode(char *buf, int size, neu_json_elem_t *ele); int neu_json_decode_array_size_by_json(void *json, char *child); +int neu_json_decode_array_elem(void *json, int index, int size, + neu_json_elem_t *ele); int neu_json_decode_array_by_json(void *json, char *name, int index, int size, neu_json_elem_t *ele); int neu_json_decode_array_size(char *buf, char *child); diff --git a/include/neuron/json/neu_json_rw.h b/include/neuron/json/neu_json_rw.h index d2a610d0..bd84c2c1 100644 --- a/include/neuron/json/neu_json_rw.h +++ b/include/neuron/json/neu_json_rw.h @@ -94,6 +94,24 @@ int neu_json_decode_write_tags_req_json(void * json_obj, 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; + int n_tag; + neu_json_write_tags_elem_t *tags; +} neu_json_write_gtags_elem_t; +typedef struct { + char *node; + + int n_group; + neu_json_write_gtags_elem_t *groups; +} neu_json_write_gtags_req_t; + +int neu_json_decode_write_gtags_req(char * buf, + neu_json_write_gtags_req_t **result); +int neu_json_decode_write_gtags_req_json(void * json_obj, + neu_json_write_gtags_req_t **result); +void neu_json_decode_write_gtags_req_free(neu_json_write_gtags_req_t *req); + typedef struct { union { neu_json_write_req_t single; diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index 095d983d..a26d1b85 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -77,6 +77,9 @@ static struct neu_http_handler cors_handler[] = { { .url = "/api/v2/write/tags", }, + { + .url = "/api/v2/write/gtags", + }, { .url = "/api/v2/subscribe", }, @@ -289,6 +292,12 @@ static struct neu_http_handler rest_handlers[] = { .url = "/api/v2/write/tags", .value.handler = handle_write_tags, }, + { + .method = NEU_HTTP_METHOD_POST, + .type = NEU_HTTP_HANDLER_FUNCTION, + .url = "/api/v2/write/gtags", + .value.handler = handle_write_gtags, + }, { .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 e46683b8..702068c0 100644 --- a/plugins/restful/rw_handle.c +++ b/plugins/restful/rw_handle.c @@ -204,6 +204,83 @@ void handle_write_tags(nng_aio *aio) }) } +static void trans(neu_json_write_gtags_req_t *req, neu_req_write_gtags_t *cmd) +{ + strcpy(cmd->driver, req->node); + cmd->n_group = req->n_group; + cmd->groups = calloc(cmd->n_group, sizeof(neu_req_gtag_group_t)); + for (int i = 0; i < cmd->n_group; i++) { + strcpy(cmd->groups[i].group, req->groups[i].group); + cmd->groups[i].n_tag = req->groups[i].n_tag; + cmd->groups[i].tags = + calloc(cmd->groups[i].n_tag, sizeof(neu_resp_tag_value_t)); + + for (int k = 0; k < cmd->groups[i].n_tag; k++) { + strcpy(cmd->groups[i].tags[k].tag, req->groups[i].tags[k].tag); + + switch (req->groups[i].tags[k].t) { + case NEU_JSON_INT: + cmd->groups[i].tags[k].value.type = NEU_TYPE_INT64; + cmd->groups[i].tags[k].value.value.u64 = + req->groups[i].tags[k].value.val_int; + break; + case NEU_JSON_STR: + cmd->groups[i].tags[k].value.type = NEU_TYPE_STRING; + strcpy(cmd->groups[i].tags[k].value.value.str, + req->groups[i].tags[k].value.val_str); + break; + case NEU_JSON_DOUBLE: + cmd->groups[i].tags[k].value.type = NEU_TYPE_DOUBLE; + cmd->groups[i].tags[k].value.value.d64 = + req->groups[i].tags[k].value.val_double; + break; + case NEU_JSON_BOOL: + cmd->groups[i].tags[k].value.type = NEU_TYPE_BOOL; + cmd->groups[i].tags[k].value.value.boolean = + req->groups[i].tags[k].value.val_bool; + break; + case NEU_JSON_BYTES: + cmd->groups[i].tags[k].value.type = NEU_TYPE_BYTES; + cmd->groups[i].tags[k].value.value.bytes.length = + req->groups[i].tags[k].value.val_bytes.length; + memcpy(cmd->groups[i].tags[k].value.value.bytes.bytes, + req->groups[i].tags[k].value.val_bytes.bytes, + req->groups[i].tags[k].value.val_bytes.length); + break; + default: + assert(false); + break; + } + } + } +} + +void handle_write_gtags(nng_aio *aio) +{ + neu_plugin_t *plugin = neu_rest_get_plugin(); + + NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT( + aio, neu_json_write_gtags_req_t, neu_json_decode_write_gtags_req, { + neu_reqresp_head_t header = { 0 }; + neu_req_write_gtags_t cmd = { 0 }; + + nng_http_req *nng_req = nng_aio_get_input(aio, 0); + nlog_notice("<%p> req %s %s", aio, nng_http_req_get_method(nng_req), + nng_http_req_get_uri(nng_req)); + header.ctx = aio; + header.type = NEU_REQ_WRITE_GTAGS; + + trans(req, &cmd); + + 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_read_resp(nng_aio *aio, neu_resp_read_group_t *resp) { neu_json_read_resp_t api_res = { 0 }; diff --git a/plugins/restful/rw_handle.h b/plugins/restful/rw_handle.h index b3c27ed1..b0c3494d 100644 --- a/plugins/restful/rw_handle.h +++ b/plugins/restful/rw_handle.h @@ -29,6 +29,7 @@ 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_write_gtags(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 af3c4688..c3d0e269 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -427,6 +427,7 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, switch (pheader->type) { case NEU_REQ_READ_GROUP: case NEU_REQ_WRITE_TAGS: + case NEU_REQ_WRITE_GTAGS: case NEU_REQ_WRITE_TAG: { neu_req_read_group_t *cmd = (neu_req_read_group_t *) data; strcpy(pheader->receiver, cmd->driver); @@ -716,6 +717,22 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } break; } + case NEU_REQ_WRITE_GTAGS: { + 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(header->len, 1); + memcpy(msg_dump, header, header->len); + neu_adapter_driver_write_gtags((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 }; @@ -1707,6 +1724,9 @@ void neu_msg_gen(neu_reqresp_head_t *header, void *data) case NEU_REQ_WRITE_TAGS: data_size = sizeof(neu_req_write_tags_t); break; + case NEU_REQ_WRITE_GTAGS: + data_size = sizeof(neu_req_write_gtags_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 06915227..eaf28fb6 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -75,6 +75,7 @@ struct neu_adapter_driver { neu_driver_cache_t *cache; neu_events_t * driver_events; + neu_events_t * driver_write_events; size_t tag_cnt; struct group *groups; @@ -266,6 +267,7 @@ neu_adapter_driver_t *neu_adapter_driver_create() driver->cache = neu_driver_cache_new(); driver->driver_events = neu_event_new(); + driver->driver_write_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; @@ -277,6 +279,7 @@ neu_adapter_driver_t *neu_adapter_driver_create() void neu_adapter_driver_destroy(neu_adapter_driver_t *driver) { neu_event_close(driver->driver_events); + neu_event_close(driver->driver_write_events); neu_driver_cache_destroy(driver->cache); } @@ -310,7 +313,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver) neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(el->group)); neu_adapter_del_timer((neu_adapter_t *) driver, el->report); neu_event_del_timer(driver->driver_events, el->read); - neu_event_del_timer(driver->driver_events, el->write); + neu_event_del_timer(driver->driver_write_events, el->write); if (el->grp.group_free != NULL) { el->grp.group_free(&el->grp); } @@ -663,6 +666,127 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver, free_tags(cmd); } +void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver, + neu_reqresp_head_t * req) +{ + neu_req_write_gtags_t *cmd = (neu_req_write_gtags_t *) &req[1]; + group_t * first_g = NULL; + + if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { + for (int i = 0; i < cmd->n_group; i++) { + free(cmd->groups[i].tags); + } + free(cmd->groups); + driver->adapter.cb_funs.driver.write_response( + &driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING); + return; + } + + if (driver->adapter.module->intf_funs->driver.write_tags == NULL) { + for (int i = 0; i < cmd->n_group; i++) { + free(cmd->groups[i].tags); + } + free(cmd->groups); + driver->adapter.cb_funs.driver.write_response( + &driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS); + return; + } + + for (int i = 0; i < cmd->n_group; i++) { + group_t *g = find_group(driver, cmd->groups[i].group); + + if (g == NULL) { + neu_resp_error_t error = { .error = NEU_ERR_GROUP_NOT_EXIST }; + req->type = NEU_RESP_ERROR; + for (int x = 0; x < cmd->n_group; x++) { + free(cmd->groups[x].tags); + } + free(cmd->groups); + driver->adapter.cb_funs.response(&driver->adapter, req, &error); + free(req); + return; + } + + if (first_g == NULL) { + first_g = g; + } + } + + 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_group; i++) { + group_t *g = find_group(driver, cmd->groups[i].group); + + for (int k = 0; k < cmd->groups[i].n_tag; k++) { + neu_plugin_tag_value_t tv = { 0 }; + + neu_datatag_t *tag = + neu_group_find_tag(g->group, cmd->groups[i].tags[k].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->groups[i].tags[k].value.type == NEU_TYPE_INT64) { + tv.value.d64 = (double) tv.value.u64; + } + } + if (tag->decimal != 0) { + cal_decimal(tag->type, cmd->groups[i].tags[k].value.type, + &cmd->groups[i].tags[k].value.value, + tag->decimal); + } + fix_value(tag, cmd->groups[i].tags[k].value.type, + &cmd->groups[i].tags[k].value); + + tv.value = cmd->groups[i].tags[k].value.value; + utarray_push_back(tags, &tv); + } + if (tag != NULL) { + neu_tag_free(tag); + } + } + } + + uint32_t n_tag = 0; + for (int i = 0; i < cmd->n_group; i++) { + n_tag += cmd->groups[i].n_tag; + } + + if (utarray_len(tags) != (unsigned int) n_tag) { + neu_resp_error_t error = { .error = NEU_ERR_TAG_NOT_EXIST }; + req->type = NEU_RESP_ERROR; + for (int i = 0; i < cmd->n_group; i++) { + free(cmd->groups[i].tags); + } + free(cmd->groups); + 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(first_g, &wtag); + for (int i = 0; i < cmd->n_group; i++) { + free(cmd->groups[i].tags); + } + free(cmd->groups); +} + void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, neu_reqresp_head_t * req) { @@ -788,7 +912,7 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name, param.second = 0; param.millisecond = 3; param.cb = write_callback; - find->write = neu_event_add_timer(driver->driver_events, param); + find->write = neu_event_add_timer(driver->driver_write_events, param); REGISTER_GROUP_METRIC(&driver->adapter, find->name, NEU_METRIC_GROUP_TAGS_TOTAL, diff --git a/src/adapter/driver/driver_internal.h b/src/adapter/driver/driver_internal.h index c7cfbb2a..a91ca58a 100644 --- a/src/adapter/driver/driver_internal.h +++ b/src/adapter/driver/driver_internal.h @@ -40,6 +40,8 @@ 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); +void neu_adapter_driver_write_gtags(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/connection/connection.c b/src/connection/connection.c index 73639d43..f8cf1f12 100644 --- a/src/connection/connection.c +++ b/src/connection/connection.c @@ -348,14 +348,19 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len) } } else { if (rc == -1 && errno == EAGAIN) { - if (retry > 5) { + if (retry > 50) { zlog_error(conn->param.log, "conn fd: %d, send buf len: %zd, ret: %zd, " "errno: %s(%d)", conn->fd, len, ret, strerror(errno), errno); break; } else { - sleep(1); + struct timespec t1 = { + .tv_sec = 0, + .tv_nsec = 1000 * 1000 * 10, + }; + struct timespec t2 = { 0 }; + nanosleep(&t1, &t2); retry++; zlog_warn(conn->param.log, "not all data send, retry: %d, ret: " diff --git a/src/core/manager.c b/src/core/manager.c index 8f39b064..6eac1c4d 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -1190,6 +1190,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_READ_GROUP: case NEU_REQ_WRITE_TAG: case NEU_REQ_WRITE_TAGS: + case NEU_REQ_WRITE_GTAGS: case NEU_REQ_GET_TAG: case NEU_REQ_GET_NDRIVER_TAGS: case NEU_REQ_NODE_CTL: diff --git a/src/parser/neu_json_rw.c b/src/parser/neu_json_rw.c index 5fdfddd6..777df71b 100644 --- a/src/parser/neu_json_rw.c +++ b/src/parser/neu_json_rw.c @@ -556,4 +556,149 @@ void neu_json_metas_to_json(neu_tag_meta_t *metas, int n_meta, break; } } -} \ No newline at end of file +} + +int neu_json_decode_write_gtags_req(char * buf, + neu_json_write_gtags_req_t **result) +{ + void *json_obj = neu_json_decode_new(buf); + if (NULL == json_obj) { + return -1; + } + + int ret = neu_json_decode_write_gtags_req_json(json_obj, result); + neu_json_decode_free(json_obj); + return ret; +} + +#include "utils/log.h" + +static int decode_write_gtags_req_json(void * json_obj, + neu_json_write_gtags_req_t *req) +{ + int ret = 0; + + neu_json_elem_t req_elems[] = { + { + .name = "node", + .t = NEU_JSON_STR, + }, + { + .name = "groups", + .t = NEU_JSON_OBJECT, + }, + }; + ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems), + req_elems); + if (ret != 0) { + return -1; + } + + req->node = req_elems[0].v.val_str; + + req->n_group = neu_json_decode_array_size_by_json(json_obj, "groups"); + if (req->n_group <= 0) { + return -1; + } + + req->groups = calloc(req->n_group, sizeof(neu_json_write_gtags_elem_t)); + for (int i = 0; i < req->n_group; i++) { + neu_json_elem_t g_elems[] = { + { + .name = "group", + .t = NEU_JSON_STR, + }, + { + .name = "tags", + .t = NEU_JSON_OBJECT, + }, + }; + + ret = neu_json_decode_array_by_json( + json_obj, "groups", i, NEU_JSON_ELEM_SIZE(g_elems), g_elems); + + req->groups[i].group = g_elems[0].v.val_str; + req->groups[i].n_tag = json_array_size(g_elems[1].v.val_object); + if (req->groups[i].n_tag <= 0) { + continue; + } + + req->groups[i].tags = + calloc(req->groups[i].n_tag, sizeof(neu_json_write_tags_elem_t)); + + for (int k = 0; k < req->groups[i].n_tag; k++) { + neu_json_elem_t v_elems[] = { + { + .name = "tag", + .t = NEU_JSON_STR, + }, + { + .name = "value", + .t = NEU_JSON_VALUE, + }, + }; + + ret = neu_json_decode_array_elem(g_elems[1].v.val_object, k, + NEU_JSON_ELEM_SIZE(v_elems), + v_elems); + if (ret != 0) { + for (int x = i; x >= 0; x--) { + free(req->groups[x].tags); + } + free(req->groups); + return -1; + } + req->groups[i].tags[k].tag = v_elems[0].v.val_str; + req->groups[i].tags[k].t = v_elems[1].t; + req->groups[i].tags[k].value = v_elems[1].v; + } + } + + return 0; +} + +int neu_json_decode_write_gtags_req_json(void * json_obj, + neu_json_write_gtags_req_t **result) +{ + neu_json_write_gtags_req_t *req = + calloc(1, sizeof(neu_json_write_gtags_req_t)); + if (req == NULL) { + return -1; + } + + int ret = decode_write_gtags_req_json(json_obj, req); + if (0 == ret) { + *result = req; + } else { + free(req); + } + + return ret; +} + +void neu_json_decode_write_gtags_req_free(neu_json_write_gtags_req_t *req) +{ + free(req->node); + + for (int i = 0; i < req->n_group; i++) { + free(req->groups[i].group); + + for (int k = 0; k < req->groups[i].n_tag; k++) { + free(req->groups[i].tags[k].tag); + + if (req->groups[i].tags[k].t == NEU_JSON_STR) { + free(req->groups[i].tags[k].value.val_str); + } + if (req->groups[i].tags[k].t == NEU_JSON_BYTES && + req->groups[i].tags[k].value.val_bytes.length > 0) { + free(req->groups[i].tags[k].value.val_bytes.bytes); + } + } + if (req->groups[i].n_tag > 0) { + free(req->groups[i].tags); + } + } + + free(req->groups); + free(req); +} diff --git a/src/utils/json.c b/src/utils/json.c index 66fb94b9..f8b44f98 100644 --- a/src/utils/json.c +++ b/src/utils/json.c @@ -272,6 +272,25 @@ int neu_json_decode_array(char *buf, char *name, int index, int size, return 0; } +int neu_json_decode_array_elem(void *json, int index, int size, + neu_json_elem_t *ele) +{ + json_t *child = NULL; + + child = json_array_get(json, index); + if (child == NULL) { + return -1; + } + + for (int i = 0; i < size; i++) { + if (decode_object(child, &ele[i]) == -1) { + return -1; + } + } + + return 0; +} + int neu_json_decode_array_by_json(void *json, char *name, int index, int size, neu_json_elem_t *ele) {