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

feat(core): support write batch

Browse Source
This commit is contained in:
eeff
2024-05-11 13:00:17 +08:00
parent add832f677
commit 1b60bed1e6
6 changed files with 157 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/errcodes.h
+1
View File
@@ -135,6 +135,7 @@ typedef enum {
NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS = 3017,
NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC = 3018,
NEU_ERR_PLUGIN_TYPE_NOT_SUPPORT = 3019,
NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_BATCH = 3021,
NEU_ERR_MQTT_FAILURE = 4000,
NEU_ERR_MQTT_NO_CERTFILESET = 4001,
src/adapter/adapter.c
+19
View File
@@ -707,6 +707,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQRESP_NODES_STATE:
case NEU_REQ_PRGFILE_PROCESS:
case NEU_RESP_PRGFILE_PROCESS:
case NEU_RESP_WRITE_BATCH:
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
neu_msg_free(msg);
@@ -803,6 +804,24 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
break;
}
case NEU_REQ_WRITE_BATCH: {
neu_reqresp_write_batch_t *cmd =
(neu_reqresp_write_batch_t *) &header[1];
neu_write_batch_t * batch = cmd->batch;
neu_write_batch_driver_t *driver = &batch->drivers[cmd->index_];
if (adapter->module->type != NEU_NA_TYPE_DRIVER) {
driver->error = NEU_ERR_GROUP_NOT_ALLOW;
header->type = NEU_RESP_WRITE_BATCH;
neu_msg_exchange(header);
reply(adapter, header, cmd);
} else {
neu_adapter_driver_write_batch((neu_adapter_driver_t *) adapter,
header);
}
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 };
src/adapter/driver/driver.c
+83
View File
@@ -162,6 +162,12 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
neu_req_write_gtags_fini((neu_req_write_gtags_t *) &req[1]);
}
if (NEU_REQ_WRITE_BATCH == req->type) {
req->type = NEU_RESP_WRITE_BATCH;
adapter->cb_funs.response(adapter, req, &req[1]);
return;
}
req->type = NEU_RESP_ERROR;
nlog_notice("write tag response <%p>", req->ctx);
@@ -979,6 +985,83 @@ void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver,
store_write_tag(first_g, &wtag);
}
void neu_adapter_driver_write_batch(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req)
{
neu_reqresp_write_batch_t *cmd = (neu_reqresp_write_batch_t *) &req[1];
neu_write_batch_driver_t * batch_driver = &cmd->batch->drivers[cmd->index_];
if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) {
batch_driver->error = NEU_ERR_PLUGIN_NOT_RUNNING;
driver->adapter.cb_funs.driver.write_response(
&driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING);
return;
}
if (driver->adapter.module->intf_funs->driver.write_batch == NULL) {
batch_driver->error = NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_BATCH;
driver->adapter.cb_funs.driver.write_response(
&driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS);
return;
}
bool need_write = false;
for (int i = 0; i < batch_driver->n_group; i++) {
neu_write_batch_group_t *batch_group = &batch_driver->groups[i];
group_t * g = find_group(driver, batch_group->group);
if (NULL == g) {
// mark all group tags error
batch_group->error = NEU_ERR_GROUP_NOT_EXIST;
continue;
}
for (int j = 0; j < batch_group->n_tag; j++) {
neu_datatag_t *tag =
neu_group_find_tag(g->group, batch_group->tags[j].name);
if (NULL == tag) {
neu_write_batch_group_set_error(batch_group, j,
NEU_ERR_TAG_NOT_EXIST);
continue;
}
if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_WRITE) ||
neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC)) {
neu_write_batch_group_set_error(
batch_group, j, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE);
continue;
}
if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
if (batch_group->tags[j].value.type == NEU_TYPE_INT64) {
batch_group->tags[j].value.value.d64 =
(double) batch_group->tags[j].value.value.i64;
}
}
if (tag->decimal != 0) {
cal_decimal(tag->type, batch_group->tags[j].value.type,
&batch_group->tags[j].value.value, tag->decimal);
}
fix_value(tag, batch_group->tags[j].value.type,
&batch_group->tags[j].value);
batch_group->tags[j].tag = tag;
need_write = true;
}
}
if (!need_write) {
// nothing to request plugin write
driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, 0);
return;
}
driver->adapter.module->intf_funs->driver.write_batch(
driver->adapter.plugin, req, batch_driver);
}
void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req)
{
src/adapter/driver/driver_internal.h
+2
View File
@@ -42,6 +42,8 @@ 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);
void neu_adapter_driver_write_batch(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/base/msg.c
+4 -2
View File
@@ -26,5 +26,7 @@ void neu_msg_gen(neu_reqresp_head_t *header, void *data)
{
size_t data_size = neu_reqresp_size(header->type);
assert(header->len >= sizeof(neu_reqresp_head_t) + data_size);
memcpy((uint8_t *) &header[1], data, data_size);
}
if (data && &header[1] != data) {
memcpy((uint8_t *) &header[1], data, data_size);
}
}
src/core/manager.c
+48
View File
@@ -1058,6 +1058,54 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case NEU_REQ_WRITE_BATCH: {
neu_reqresp_write_batch_t *cmd =
(neu_reqresp_write_batch_t *) &header[1];
neu_write_batch_t *batch = cmd->batch;
for (int i = 0; i < batch->n_driver; ++i) {
const char *driver = batch->drivers[i].driver;
if (NULL == neu_node_manager_find(manager->node_manager, driver)) {
// mark errors and done
batch->n_done_ += 1;
batch->drivers[i].error = NEU_ERR_NODE_NOT_EXIST;
continue;
}
cmd->index_ = i;
if (i + 1 == batch->n_driver) {
forward_msg(manager, header, driver); // save a copy
} else {
forward_msg_copy(manager, header, driver);
}
}
if (batch->n_done_ == batch->n_driver) {
// all done, reply response
header->type = NEU_RESP_WRITE_BATCH;
neu_msg_exchange(header);
forward_msg(manager, header, header->receiver);
}
break;
}
case NEU_RESP_WRITE_BATCH: {
neu_reqresp_write_batch_t *cmd =
(neu_reqresp_write_batch_t *) &header[1];
neu_write_batch_t *batch = cmd->batch;
batch->n_done_ += 1;
if (batch->n_done_ == batch->n_driver) {
// all done, reply response
header->type = NEU_RESP_WRITE_BATCH;
forward_msg(manager, header, header->receiver);
} else {
neu_msg_free(msg);
}
break;
}
case NEU_REQ_PRGFILE_PROCESS:
case NEU_REQ_PRGFILE_UPLOAD:
case NEU_REQ_GET_NODE_STATE: {