GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-24 10:46:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1614 from fengzeroz/main

Browse Source 
write multiple group
This commit is contained in:
fengzeroz
2023-10-27 10:04:30 +08:00
committed by GitHub
parent bdc6896aa9 ac0559d89b
commit 22745dbdf7
13 changed files with 444 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+16
View File
@@ -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];
include/neuron/json/json.h
+2
View File
@@ -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);
include/neuron/json/neu_json_rw.h
+18
View File
@@ -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;
plugins/restful/handle.c
+9
View File
@@ -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,
plugins/restful/rw_handle.c
+77
View File
@@ -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 };
plugins/restful/rw_handle.h
+1
View File
@@ -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
src/adapter/adapter.c
+20
View File
@@ -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;
src/adapter/driver/driver.c
+126 -2
View File
@@ -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,
src/adapter/driver/driver_internal.h
+2
View File
@@ -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);
src/connection/connection.c
+7 -2
View File
@@ -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: "
src/core/manager.c
+1
View File
@@ -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:
src/parser/neu_json_rw.c
+146 -1
View File
@@ -556,4 +556,149 @@ void neu_json_metas_to_json(neu_tag_meta_t *metas, int n_meta,
break;
}
}
}
}
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);
}
src/utils/json.c
+19
View File
@@ -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)
{