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

refactor: reduce write tag messages size

Browse Source
This commit is contained in:
eeff
2023-11-15 18:20:43 +08:00
parent 14e45c8ccc
commit cddd225c84
7 changed files with 50 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/msg.h
+10 -3
View File
@@ -782,12 +782,19 @@ static inline void neu_req_read_group_fini(neu_req_read_group_t *req)
}
typedef struct neu_req_write_tag {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
char tag[NEU_TAG_NAME_LEN];
char * driver;
char * group;
char * tag;
neu_dvalue_t value;
} neu_req_write_tag_t;
static inline void neu_req_write_tag_fini(neu_req_write_tag_t *req)
{
free(req->driver);
free(req->group);
free(req->tag);
}
typedef struct neu_resp_tag_value {
char tag[NEU_TAG_NAME_LEN];
neu_dvalue_t value;
plugins/ekuiper/read_write.c
+8 -3
View File
@@ -123,9 +123,9 @@ int write_data(neu_plugin_t *plugin, json_write_req_t *write_req)
header.ctx = NULL;
header.type = NEU_REQ_WRITE_TAG;
strcpy(cmd.driver, write_req->node_name);
strcpy(cmd.group, write_req->group_name);
strcpy(cmd.tag, write_req->tag_name);
cmd.driver = write_req->node_name;
cmd.group = write_req->group_name;
cmd.tag = write_req->tag_name;
switch (write_req->t) {
case NEU_JSON_INT:
@@ -156,5 +156,10 @@ int write_data(neu_plugin_t *plugin, json_write_req_t *write_req)
}
ret = neu_plugin_op(plugin, header, &cmd);
if (0 == ret) {
write_req->node_name = NULL; // ownership moved
write_req->group_name = NULL; // ownership moved
write_req->tag_name = NULL; // ownership moved
}
return ret;
}
plugins/mqtt/mqtt_handle.c
+6 -3
View File
@@ -164,9 +164,9 @@ static int send_write_tag_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
header.ctx = mqtt;
header.type = NEU_REQ_WRITE_TAG;
strcpy(cmd.driver, req->node);
strcpy(cmd.group, req->group);
strcpy(cmd.tag, req->tag);
cmd.driver = req->node;
cmd.group = req->group;
cmd.tag = req->tag;
switch (req->t) {
case NEU_JSON_INT:
@@ -201,6 +201,9 @@ static int send_write_tag_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
return -1;
}
req->node = NULL; // ownership moved
req->group = NULL; // ownership moved
req->tag = NULL; // ownership moved
return 0;
}
plugins/restful/rw_handle.c
+7 -3
View File
@@ -90,9 +90,12 @@ void handle_write(nng_aio *aio)
header.ctx = aio;
header.type = NEU_REQ_WRITE_TAG;
strcpy(cmd.driver, req->node);
strcpy(cmd.group, req->group);
strcpy(cmd.tag, req->tag);
cmd.driver = req->node;
cmd.group = req->group;
cmd.tag = req->tag;
req->node = NULL; // ownership moved
req->group = NULL; // ownership moved
req->tag = NULL; // ownership moved
switch (req->t) {
case NEU_JSON_INT:
@@ -124,6 +127,7 @@ void handle_write(nng_aio *aio)
int ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
neu_req_write_tag_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
@@ -420,13 +420,17 @@ 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_TAG: {
neu_req_write_tag_t *cmd = (neu_req_write_tag_t *) data;
strcpy(pheader->receiver, cmd->driver);
break;
}
case NEU_REQ_WRITE_TAGS:
case NEU_REQ_WRITE_GTAGS: {
neu_req_write_tags_t *cmd = (neu_req_write_tags_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);
@@ -688,6 +692,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_adapter_driver_write_tag((neu_adapter_driver_t *) adapter,
msg_dump);
} else {
neu_req_write_tag_fini((neu_req_write_tag_t *) &header[1]);
error.error = NEU_ERR_GROUP_NOT_ALLOW;
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
src/adapter/driver/driver.c
+10 -13
View File
@@ -110,6 +110,10 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
neu_reqresp_head_t *req = (neu_reqresp_head_t *) r;
neu_resp_error_t nerror = { .error = error };
if (NEU_REQ_WRITE_TAG == req->type) {
neu_req_write_tag_fini((neu_req_write_tag_t *) &req[1]);
}
req->type = NEU_RESP_ERROR;
nlog_notice("write tag response <%p>", req->ctx);
@@ -809,20 +813,15 @@ void neu_adapter_driver_write_tag(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;
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;
}
neu_datatag_t *tag = neu_group_find_tag(g->group, cmd->tag);
if (tag == NULL) {
neu_resp_error_t error = { .error = NEU_ERR_TAG_NOT_EXIST };
req->type = NEU_RESP_ERROR;
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);
} else {
if ((tag->attribute & NEU_ATTRIBUTE_WRITE) != NEU_ATTRIBUTE_WRITE) {
driver->adapter.cb_funs.driver.write_response(
@@ -849,10 +848,8 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
neu_tag_set_static_value(tag, &cmd->value.value);
neu_group_update_tag(g->group, tag);
adapter_storage_update_tag_value(cmd->driver, cmd->group, tag);
neu_resp_error_t error = { .error = NEU_ERR_SUCCESS };
req->type = NEU_RESP_ERROR;
driver->adapter.cb_funs.response(&driver->adapter, req, &error);
free(req);
driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
NEU_ERR_SUCCESS);
} else {
to_be_write_tag_t wtag = { 0 };
wtag.single = true;
src/core/manager.c
+2
View File
@@ -826,6 +826,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
NULL) {
if (NEU_REQ_READ_GROUP == header->type) {
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]);
}
neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
header->type = NEU_RESP_ERROR;