mirror of
https://github.com/emqx/neuron.git
synced 2026-10-17 04:40:42 +08:00
refactor: reduce read group message size
This commit is contained in:
+13
-5
@@ -793,11 +793,17 @@ static inline void neu_req_inst_templates_fini(neu_req_inst_templates_t *req)
|
||||
}
|
||||
|
||||
typedef struct neu_req_read_group {
|
||||
char driver[NEU_NODE_NAME_LEN];
|
||||
char group[NEU_GROUP_NAME_LEN];
|
||||
bool sync;
|
||||
char *driver;
|
||||
char *group;
|
||||
bool sync;
|
||||
} neu_req_read_group_t;
|
||||
|
||||
static inline void neu_req_read_group_fini(neu_req_read_group_t *req)
|
||||
{
|
||||
free(req->driver);
|
||||
free(req->group);
|
||||
}
|
||||
|
||||
typedef struct neu_req_write_tag {
|
||||
char driver[NEU_NODE_NAME_LEN];
|
||||
char group[NEU_GROUP_NAME_LEN];
|
||||
@@ -845,8 +851,8 @@ typedef struct {
|
||||
} neu_req_write_gtags_t;
|
||||
|
||||
typedef struct {
|
||||
char driver[NEU_NODE_NAME_LEN];
|
||||
char group[NEU_GROUP_NAME_LEN];
|
||||
char *driver;
|
||||
char *group;
|
||||
|
||||
UT_array *tags; // neu_resp_tag_value_meta_t
|
||||
} neu_resp_read_group_t;
|
||||
@@ -859,6 +865,8 @@ static inline void neu_resp_read_free(neu_resp_read_group_t *resp)
|
||||
free(tag_value->value.value.ptr.ptr);
|
||||
}
|
||||
}
|
||||
free(resp->driver);
|
||||
free(resp->group);
|
||||
utarray_free(resp->tags);
|
||||
}
|
||||
|
||||
|
||||
@@ -138,10 +138,13 @@ static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
|
||||
header.ctx = mqtt;
|
||||
header.type = NEU_REQ_READ_GROUP;
|
||||
neu_req_read_group_t cmd = { 0 };
|
||||
strcpy(cmd.driver, req->node);
|
||||
strcpy(cmd.group, req->group);
|
||||
cmd.sync = req->sync;
|
||||
cmd.driver = req->node;
|
||||
cmd.group = req->group;
|
||||
cmd.sync = req->sync;
|
||||
req->node = NULL; // ownership moved
|
||||
req->group = NULL; // ownership moved
|
||||
if (0 != neu_plugin_op(plugin, header, &cmd)) {
|
||||
neu_req_read_group_fini(&cmd);
|
||||
plog_error(plugin, "neu_plugin_op(NEU_REQ_READ_GROUP) fail");
|
||||
return -1;
|
||||
}
|
||||
|
||||
@@ -41,11 +41,14 @@ void handle_read(nng_aio *aio)
|
||||
header.ctx = aio;
|
||||
header.type = NEU_REQ_READ_GROUP;
|
||||
|
||||
strcpy(cmd.driver, req->node);
|
||||
strcpy(cmd.group, req->group);
|
||||
cmd.sync = req->sync;
|
||||
ret = neu_plugin_op(plugin, header, &cmd);
|
||||
cmd.driver = req->node;
|
||||
cmd.group = req->group;
|
||||
cmd.sync = req->sync;
|
||||
req->node = NULL;
|
||||
req->group = NULL;
|
||||
ret = neu_plugin_op(plugin, header, &cmd);
|
||||
if (ret != 0) {
|
||||
neu_req_read_group_fini(&cmd);
|
||||
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
|
||||
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
|
||||
});
|
||||
|
||||
@@ -425,11 +425,15 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
|
||||
|
||||
strcpy(pheader->sender, adapter->name);
|
||||
switch (pheader->type) {
|
||||
case NEU_REQ_READ_GROUP:
|
||||
case NEU_REQ_READ_GROUP: {
|
||||
neu_req_read_group_t *cmd = (neu_req_read_group_t *) data;
|
||||
strcpy(pheader->receiver, cmd->driver);
|
||||
break;
|
||||
}
|
||||
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;
|
||||
neu_req_write_tag_t *cmd = (neu_req_write_tag_t *) data;
|
||||
strcpy(pheader->receiver, cmd->driver);
|
||||
break;
|
||||
}
|
||||
@@ -685,6 +689,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
|
||||
neu_adapter_driver_read_group((neu_adapter_driver_t *) adapter,
|
||||
header);
|
||||
} else {
|
||||
neu_req_read_group_fini((neu_req_read_group_t *) &header[1]);
|
||||
error.error = NEU_ERR_GROUP_NOT_ALLOW;
|
||||
header->type = NEU_RESP_ERROR;
|
||||
neu_msg_exchange(header);
|
||||
|
||||
@@ -390,6 +390,7 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
|
||||
if (g == NULL) {
|
||||
neu_resp_error_t error = { .error = NEU_ERR_GROUP_NOT_EXIST };
|
||||
req->type = NEU_RESP_ERROR;
|
||||
neu_req_read_group_fini(cmd);
|
||||
driver->adapter.cb_funs.response(&driver->adapter, req, &error);
|
||||
return;
|
||||
}
|
||||
@@ -442,10 +443,13 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
|
||||
driver->cache, cmd->group, tags, resp.tags);
|
||||
}
|
||||
|
||||
strcpy(resp.driver, cmd->driver);
|
||||
strcpy(resp.group, cmd->group);
|
||||
resp.driver = cmd->driver;
|
||||
resp.group = cmd->group;
|
||||
cmd->driver = NULL; // ownership moved
|
||||
cmd->group = NULL; // ownership moved
|
||||
|
||||
utarray_free(tags);
|
||||
neu_req_read_group_fini(cmd);
|
||||
|
||||
req->type = NEU_RESP_READ_GROUP;
|
||||
driver->adapter.cb_funs.response(&driver->adapter, req, &resp);
|
||||
|
||||
@@ -1200,6 +1200,9 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
|
||||
case NEU_REQ_ADD_GROUP: {
|
||||
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
|
||||
NULL) {
|
||||
if (NEU_REQ_READ_GROUP == header->type) {
|
||||
neu_req_read_group_fini((neu_req_read_group_t *) &header[1]);
|
||||
}
|
||||
neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
|
||||
header->type = NEU_RESP_ERROR;
|
||||
neu_msg_exchange(header);
|
||||
|
||||
Reference in New Issue
Block a user