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

refactor: reduce write tags message size

Browse Source
This commit is contained in:
eeff
2023-11-16 00:18:03 +08:00
parent 297b9b151d
commit be13500d24
7 changed files with 42 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/msg.h
+9 -2
View File
@@ -836,13 +836,20 @@ static inline UT_icd *neu_resp_tag_value_meta_icd()
}
typedef struct neu_req_write_tags {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
char *driver;
char *group;
int n_tag;
neu_resp_tag_value_t *tags;
} neu_req_write_tags_t;
static inline void neu_req_write_tags_fini(neu_req_write_tags_t *req)
{
free(req->driver);
free(req->group);
free(req->tags);
}
typedef struct {
char group[NEU_GROUP_NAME_LEN];
plugins/mqtt/mqtt_handle.c
+7 -4
View File
@@ -227,10 +227,10 @@ static int send_write_tags_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
};
neu_req_write_tags_t cmd = { 0 };
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));
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;
}
@@ -275,6 +275,9 @@ static int send_write_tags_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
return -1;
}
req->node = NULL; // ownership moved
req->group = NULL; // ownership moved
return 0;
}
plugins/restful/rw_handle.c
+7 -4
View File
@@ -162,10 +162,12 @@ void handle_write_tags(nng_aio *aio)
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));
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));
req->node = NULL; // ownership moved
req->group = NULL; // ownership moved
for (int i = 0; i < cmd.n_tag; i++) {
strcpy(cmd.tags[i].tag, req->tags[i].tag);
@@ -204,6 +206,7 @@ void handle_write_tags(nng_aio *aio)
int ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
neu_req_write_tags_fini(&cmd);
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
src/adapter/adapter.c
+7 -2
View File
@@ -435,12 +435,16 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
strcpy(pheader->receiver, cmd->driver);
break;
}
case NEU_REQ_WRITE_TAGS:
case NEU_REQ_WRITE_GTAGS: {
case NEU_REQ_WRITE_TAGS: {
neu_req_write_tags_t *cmd = (neu_req_write_tags_t *) data;
strcpy(pheader->receiver, cmd->driver);
break;
}
case NEU_REQ_WRITE_GTAGS: {
neu_req_write_gtags_t *cmd = (neu_req_write_gtags_t *) data;
strcpy(pheader->receiver, cmd->driver);
break;
}
case NEU_REQ_DEL_NODE: {
neu_req_del_node_t *cmd = (neu_req_del_node_t *) data;
strcpy(pheader->receiver, cmd->node);
@@ -723,6 +727,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_resp_error_t error = { 0 };
if (adapter->module->type != NEU_NA_TYPE_DRIVER) {
neu_req_write_tags_fini((neu_req_write_tags_t *) &header[1]);
error.error = NEU_ERR_GROUP_NOT_ALLOW;
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
src/adapter/driver/driver.c
+6 -18
View File
@@ -109,6 +109,8 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
if (NEU_REQ_WRITE_TAG == req->type) {
neu_req_write_tag_fini((neu_req_write_tag_t *) &req[1]);
} else if (NEU_REQ_WRITE_TAGS == req->type) {
neu_req_write_tags_fini((neu_req_write_tags_t *) &req[1]);
}
req->type = NEU_RESP_ERROR;
@@ -588,25 +590,18 @@ static void cal_decimal(neu_type_e tag_type, neu_type_e value_type,
}
}
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) {
free_tags(cmd);
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) {
free_tags(cmd);
driver->adapter.cb_funs.driver.write_response(
&driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS);
return;
@@ -615,11 +610,8 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
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);
driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
NEU_ERR_GROUP_NOT_EXIST);
return;
}
@@ -654,11 +646,8 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
}
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);
driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
NEU_ERR_TAG_NOT_EXIST);
utarray_foreach(tags, neu_plugin_tag_value_t *, tv)
{
neu_tag_free(tv->tag);
@@ -672,7 +661,6 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
wtag.req = (void *) req;
wtag.tvs = tags;
free_tags(cmd);
store_write_tag(g, &wtag);
}
src/core/manager.c
+2
View File
@@ -1204,6 +1204,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_req_read_group_fini((neu_req_read_group_t *) &header[1]);
} else if (NEU_REQ_WRITE_TAG == header->type) {
neu_req_write_tag_fini((neu_req_write_tag_t *) &header[1]);
} else if (NEU_REQ_WRITE_TAGS == header->type) {
neu_req_write_tags_fini((neu_req_write_tags_t *) &header[1]);
}
neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
header->type = NEU_RESP_ERROR;
src/parser/neu_json_rw.c
+4 -1
View File
@@ -306,14 +306,17 @@ static int decode_write_tags_req_json(void * json_obj,
req->tags[i].value = v_elems[1].v;
if (ret != 0) {
while (i-- >= 0) {
for (; i >= 0; --i) {
free(req->tags[i].tag);
if (NEU_JSON_STR == req->tags[i].t) {
free(req->tags[i].value.val_str);
}
}
free(req->node);
free(req->group);
free(req->tags);
req->tags = NULL;
return ret;
}
}