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

feat(core): support writing multiple tag in one request

Browse Source
This commit is contained in:
fengzero
2023-06-12 06:14:28 +00:00
parent 685d756a9f
commit fadca0f2c9
16 changed files with 499 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+12
View File
@@ -43,6 +43,7 @@ typedef enum neu_reqresp_type {
NEU_REQ_READ_GROUP,
NEU_RESP_READ_GROUP,
NEU_REQ_WRITE_TAG,
NEU_REQ_WRITE_TAGS,
NEU_REQ_SUBSCRIBE_GROUP,
NEU_REQ_UNSUBSCRIBE_GROUP,
@@ -144,6 +145,7 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_READ_GROUP] = "NEU_REQ_READ_GROUP",
[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_SUBSCRIBE_GROUP] = "NEU_REQ_SUBSCRIBE_GROUP",
[NEU_REQ_UNSUBSCRIBE_GROUP] = "NEU_REQ_UNSUBSCRIBE_GROUP",
@@ -634,6 +636,14 @@ typedef struct neu_resp_tag_value {
neu_dvalue_t value;
} neu_resp_tag_value_t;
typedef struct neu_req_write_tags {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
int n_tag;
neu_resp_tag_value_t *tags;
} neu_req_write_tags_t;
typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
@@ -783,6 +793,8 @@ typedef struct adapter_callbacks {
const char *tag, neu_dvalue_t value);
void (*write_response)(neu_adapter_t *adapter, void *req,
int error);
void (*update_im)(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value);
} driver;
};
} adapter_callbacks_t;
include/neuron/errcodes.h
+1
View File
@@ -117,6 +117,7 @@ typedef enum {
NEU_ERR_PLUGIN_NOT_FOUND = 3014,
NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE = 3015,
NEU_ERR_PLUGIN_NOT_SUPPORT_TEMPLATE = 3016,
NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS = 3017,
NEU_ERR_MQTT_FAILURE = 4000,
NEU_ERR_MQTT_NO_CERTFILESET = 4001,
include/neuron/json/neu_json_rw.h
+16
View File
@@ -64,6 +64,22 @@ typedef struct {
int neu_json_decode_write_req(char *buf, neu_json_write_req_t **result);
void neu_json_decode_write_req_free(neu_json_write_req_t *req);
typedef struct {
char * tag;
enum neu_json_type t;
union neu_json_value value;
} neu_json_write_tags_elem_t;
typedef struct {
char * group;
char * node;
int n_tag;
neu_json_write_tags_elem_t *tags;
} neu_json_write_tags_req_t;
int neu_json_decode_write_tags_req(char * buf,
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;
char *node;
include/neuron/plugin.h
+8
View File
@@ -64,6 +64,11 @@ struct neu_plugin_group {
typedef int (*neu_plugin_tag_validator_t)(const neu_datatag_t *tag);
typedef struct {
neu_datatag_t *tag;
neu_value_u value;
} neu_plugin_tag_value_t;
typedef struct neu_plugin_intf_funs {
neu_plugin_t *(*open)(void);
int (*close)(neu_plugin_t *plugin);
@@ -81,6 +86,9 @@ typedef struct neu_plugin_intf_funs {
int (*group_timer)(neu_plugin_t *plugin, neu_plugin_group_t *group);
int (*write_tag)(neu_plugin_t *plugin, void *req,
neu_datatag_t *tag, neu_value_u value);
int (*write_tags)(
neu_plugin_t *plugin, void *req,
UT_array *tag_values); // UT_array {neu_datatag_t, neu_value_u}
neu_plugin_tag_validator_t tag_validator;
} driver;
};
plugins/modbus/modbus_qh_tcp.c
+1
View File
@@ -57,6 +57,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
.driver.group_timer = driver_group_timer,
.driver.write_tag = driver_write,
.driver.tag_validator = driver_tag_validator,
.driver.write_tags = NULL,
};
const neu_plugin_module_t neu_plugin_module = {
plugins/modbus/modbus_rtu.c
+1
View File
@@ -57,6 +57,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
.driver.group_timer = driver_group_timer,
.driver.write_tag = driver_write,
.driver.tag_validator = driver_tag_validator,
.driver.write_tags = NULL,
};
const neu_plugin_module_t neu_plugin_module = {
plugins/modbus/modbus_tcp.c
+1
View File
@@ -57,6 +57,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
.driver.group_timer = driver_group_timer,
.driver.write_tag = driver_write,
.driver.tag_validator = driver_tag_validator,
.driver.write_tags = NULL,
};
const neu_plugin_module_t neu_plugin_module = {
plugins/restful/handle.c
+9
View File
@@ -74,6 +74,9 @@ static struct neu_http_handler cors_handler[] = {
{
.url = "/api/v2/write",
},
{
.url = "/api/v2/write/tags",
},
{
.url = "/api/v2/subscribe",
},
@@ -277,6 +280,12 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/write",
.value.handler = handle_write,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/write/tags",
.value.handler = handle_write_tags,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
plugins/restful/rw_handle.c
+79 -17
View File
@@ -61,6 +61,15 @@ void handle_write(nng_aio *aio)
neu_reqresp_head_t header = { 0 };
neu_req_write_tag_t cmd = { 0 };
if (req->t == NEU_JSON_STR &&
strlen(req->value.val_str) >= NEU_VALUE_SIZE) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, {
neu_http_response(aio, NEU_ERR_STRING_TOO_LONG,
result_error);
});
return;
}
header.ctx = aio;
header.type = NEU_REQ_WRITE_TAG;
@@ -74,13 +83,8 @@ void handle_write(nng_aio *aio)
cmd.value.value.u64 = req->value.val_int;
break;
case NEU_JSON_STR:
if (strlen(req->value.val_str) >= NEU_VALUE_SIZE) {
cmd.value.type = NEU_TYPE_ERROR;
cmd.value.value.i32 = NEU_ERR_STRING_TOO_LONG;
} else {
cmd.value.type = NEU_TYPE_STRING;
strcpy(cmd.value.value.str, req->value.val_str);
}
cmd.value.type = NEU_TYPE_STRING;
strcpy(cmd.value.value.str, req->value.val_str);
break;
case NEU_JSON_DOUBLE:
cmd.value.type = NEU_TYPE_DOUBLE;
@@ -95,19 +99,77 @@ void handle_write(nng_aio *aio)
break;
}
if (cmd.value.type == NEU_TYPE_ERROR) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, {
neu_http_response(aio, NEU_ERR_STRING_TOO_LONG,
result_error);
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);
});
} else {
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_write_tags(nng_aio *aio)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_write_tags_req_t, neu_json_decode_write_tags_req, {
neu_reqresp_head_t header = { 0 };
neu_req_write_tags_t cmd = { 0 };
for (int i = 0; i < req->n_tag; i++) {
if (req->tags[i].t == NEU_JSON_STR) {
if (strlen(req->tags[i].value.val_str) >= NEU_VALUE_SIZE) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, {
neu_http_response(aio, NEU_ERR_STRING_TOO_LONG,
result_error);
});
return;
}
}
}
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));
for (int i = 0; i < cmd.n_tag; i++) {
strcpy(cmd.tags[i].tag, req->tags[i].tag);
switch (req->tags[i].t) {
case NEU_JSON_INT:
cmd.tags[i].value.type = NEU_TYPE_INT64;
cmd.tags[i].value.value.u64 = req->tags[i].value.val_int;
break;
case NEU_JSON_STR:
cmd.tags[i].value.type = NEU_TYPE_STRING;
strcpy(cmd.tags[i].value.value.str,
req->tags[i].value.val_str);
break;
case NEU_JSON_DOUBLE:
cmd.tags[i].value.type = NEU_TYPE_DOUBLE;
cmd.tags[i].value.value.d64 = req->tags[i].value.val_double;
break;
case NEU_JSON_BOOL:
cmd.tags[i].value.type = NEU_TYPE_BOOL;
cmd.tags[i].value.value.boolean =
req->tags[i].value.val_bool;
break;
default:
assert(false);
break;
}
}
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);
});
}
})
}
plugins/restful/rw_handle.h
+1
View File
@@ -28,6 +28,7 @@ void handle_rw_init();
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_read_resp(nng_aio *aio, neu_resp_read_group_t *resp);
#endif
src/adapter/adapter.c
+20
View File
@@ -330,6 +330,7 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
strcpy(header.sender, adapter->name);
switch (header.type) {
case NEU_REQ_READ_GROUP:
case NEU_REQ_WRITE_TAGS:
case NEU_REQ_WRITE_TAG: {
neu_req_read_group_t *cmd = (neu_req_read_group_t *) data;
strcpy(header.receiver, cmd->driver);
@@ -512,6 +513,22 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case NEU_REQ_WRITE_TAGS: {
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(nng_msg_len(msg), 1);
memcpy(msg_dump, header, nng_msg_len(msg));
neu_adapter_driver_write_tags((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 };
@@ -1436,6 +1453,9 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_WRITE_TAG:
data_size = sizeof(neu_req_write_tag_t);
break;
case NEU_REQ_WRITE_TAGS:
data_size = sizeof(neu_req_write_tags_t);
break;
case NEU_RESP_READ_GROUP:
data_size = sizeof(neu_resp_read_group_t);
break;
src/adapter/driver/driver.c
+256 -130
View File
@@ -39,8 +39,10 @@
#include "tag.h"
typedef struct to_be_write_tag {
bool single;
neu_datatag_t *tag;
neu_value_u value;
UT_array * tvs;
void * req;
} to_be_write_tag_t;
@@ -96,6 +98,15 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
free(req);
}
static void update_im(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value)
{
(void) adapter;
(void) group;
(void) tag;
(void) value;
}
static void update(neu_adapter_t *adapter, const char *group, const char *tag,
neu_dvalue_t value)
{
@@ -146,6 +157,7 @@ neu_adapter_driver_t *neu_adapter_driver_create()
driver->driver_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;
return driver;
}
@@ -286,6 +298,223 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
driver->adapter.cb_funs.response(&driver->adapter, req, &resp);
}
static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
neu_dvalue_t *value)
{
switch (tag->type) {
case NEU_TYPE_BOOL:
case NEU_TYPE_STRING:
break;
case NEU_TYPE_BIT:
value->type = NEU_TYPE_BIT;
value->value.u8 = (uint8_t) value->value.u64;
break;
case NEU_TYPE_UINT8:
case NEU_TYPE_INT8:
value->type = NEU_TYPE_UINT8;
value->value.u8 = (uint8_t) value->value.u64;
break;
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
case NEU_TYPE_WORD:
value->type = NEU_TYPE_UINT16;
value->value.u16 = (uint16_t) value->value.u64;
if (tag->option.value16.endian == NEU_DATATAG_ENDIAN_B16) {
value->value.u16 = htons(value->value.u16);
}
break;
case NEU_TYPE_UINT32:
case NEU_TYPE_INT32:
case NEU_TYPE_DWORD:
value->type = NEU_TYPE_UINT32;
value->value.u32 = (uint32_t) value->value.u64;
switch (tag->option.value32.endian) {
case NEU_DATATAG_ENDIAN_LB32:
neu_ntohs_p((uint16_t *) value->value.bytes);
neu_ntohs_p((uint16_t *) value->value.bytes + 2);
break;
case NEU_DATATAG_ENDIAN_BB32:
value->value.u32 = htonl(value->value.u32);
break;
case NEU_DATATAG_ENDIAN_BL32:
value->value.u32 = htonl(value->value.u32);
neu_ntohs_p((uint16_t *) value->value.bytes);
neu_ntohs_p((uint16_t *) value->value.bytes + 2);
break;
case NEU_DATATAG_ENDIAN_LL32:
default:
break;
}
break;
case NEU_TYPE_FLOAT:
if (value_type != NEU_TYPE_FLOAT) {
value->type = NEU_TYPE_FLOAT;
value->value.f32 = (float) value->value.d64;
}
switch (tag->option.value32.endian) {
case NEU_DATATAG_ENDIAN_LB32:
neu_ntohs_p((uint16_t *) value->value.bytes);
neu_ntohs_p((uint16_t *) value->value.bytes + 2);
break;
case NEU_DATATAG_ENDIAN_BB32:
value->value.u32 = htonl(value->value.u32);
break;
case NEU_DATATAG_ENDIAN_BL32:
value->value.u32 = htonl(value->value.u32);
neu_ntohs_p((uint16_t *) value->value.bytes);
neu_ntohs_p((uint16_t *) value->value.bytes + 2);
break;
case NEU_DATATAG_ENDIAN_LL32:
default:
break;
}
break;
case NEU_TYPE_DOUBLE:
case NEU_TYPE_UINT64:
case NEU_TYPE_INT64:
case NEU_TYPE_LWORD:
switch (tag->option.value64.endian) {
case NEU_DATATAG_ENDIAN_B64:
value->value.u64 = neu_ntohll(value->value.u64);
break;
case NEU_DATATAG_ENDIAN_L64:
default:
break;
}
break;
default:
assert(false);
break;
}
}
static void cal_decimal(neu_type_e tag_type, neu_type_e value_type,
neu_value_u *value, double decimal)
{
switch (tag_type) {
case NEU_TYPE_INT8:
case NEU_TYPE_UINT8:
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32:
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
case NEU_TYPE_WORD:
case NEU_TYPE_LWORD:
case NEU_TYPE_DWORD:
if (value_type == NEU_TYPE_INT64 || NEU_TYPE_LWORD) {
value->u64 = value->u64 / decimal;
} else if (value_type == NEU_TYPE_DOUBLE) {
value->u64 = (uint64_t) round(value->d64 / decimal);
}
break;
case NEU_TYPE_FLOAT:
case NEU_TYPE_DOUBLE:
if (value_type == NEU_TYPE_INT64) {
value->d64 = (double) (value->d64 / decimal);
} else if (value_type == NEU_TYPE_DOUBLE) {
value->d64 = value->d64 / decimal;
} else if (value_type == NEU_TYPE_FLOAT) {
value->f32 = value->f32 / decimal;
}
break;
default:
break;
}
}
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) {
driver->adapter.cb_funs.driver.write_response(
&driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING);
free_tags(cmd);
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;
}
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);
return;
}
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_tag; i++) {
neu_plugin_tag_value_t tv = { 0 };
neu_datatag_t *tag = neu_group_find_tag(g->group, cmd->tags[i].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->tags[i].value.type == NEU_TYPE_INT64) {
tv.value.d64 = (double) tv.value.u64;
}
}
if (tag->decimal != 0) {
cal_decimal(tag->type, cmd->tags[i].value.type,
&cmd->tags[i].value.value, tag->decimal);
}
fix_value(tag, cmd->tags[i].value.type, &cmd->tags[i].value);
tv.value = cmd->tags[i].value.value;
utarray_push_back(tags, &tv);
} else {
if (tag != NULL) {
neu_tag_free(tag);
}
}
}
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);
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(g, &wtag);
free_tags(cmd);
}
void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req)
{
@@ -327,133 +556,11 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
}
if (tag->decimal != 0) {
switch (tag->type) {
case NEU_TYPE_INT8:
case NEU_TYPE_UINT8:
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32:
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
case NEU_TYPE_WORD:
case NEU_TYPE_LWORD:
case NEU_TYPE_DWORD:
if (cmd->value.type == NEU_TYPE_INT64 ||
cmd->value.type == NEU_TYPE_LWORD) {
cmd->value.value.u64 = cmd->value.value.u64 / tag->decimal;
} else if (cmd->value.type == NEU_TYPE_DOUBLE) {
cmd->value.value.u64 =
(uint64_t) round(cmd->value.value.d64 / tag->decimal);
}
break;
case NEU_TYPE_FLOAT:
case NEU_TYPE_DOUBLE:
if (cmd->value.type == NEU_TYPE_INT64) {
cmd->value.value.d64 =
(double) (cmd->value.value.d64 / tag->decimal);
} else if (cmd->value.type == NEU_TYPE_DOUBLE) {
cmd->value.value.d64 = cmd->value.value.d64 / tag->decimal;
} else if (cmd->value.type == NEU_TYPE_FLOAT) {
cmd->value.value.f32 = cmd->value.value.f32 / tag->decimal;
}
break;
default:
break;
}
cal_decimal(tag->type, cmd->value.type, &cmd->value.value,
tag->decimal);
}
switch (tag->type) {
case NEU_TYPE_BOOL:
case NEU_TYPE_STRING:
break;
case NEU_TYPE_BIT:
cmd->value.type = NEU_TYPE_BIT;
cmd->value.value.u8 = (uint8_t) cmd->value.value.u64;
break;
case NEU_TYPE_UINT8:
case NEU_TYPE_INT8:
cmd->value.type = NEU_TYPE_UINT8;
cmd->value.value.u8 = (uint8_t) cmd->value.value.u64;
break;
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
case NEU_TYPE_WORD:
cmd->value.type = NEU_TYPE_UINT16;
cmd->value.value.u16 = (uint16_t) cmd->value.value.u64;
switch (tag->option.value16.endian) {
case NEU_DATATAG_ENDIAN_B16:
cmd->value.value.u16 = htons(cmd->value.value.u16);
break;
case NEU_DATATAG_ENDIAN_L16:
default:
break;
}
break;
case NEU_TYPE_UINT32:
case NEU_TYPE_INT32:
case NEU_TYPE_DWORD:
cmd->value.type = NEU_TYPE_UINT32;
cmd->value.value.u32 = (uint32_t) cmd->value.value.u64;
switch (tag->option.value32.endian) {
case NEU_DATATAG_ENDIAN_LB32:
neu_ntohs_p((uint16_t *) cmd->value.value.bytes);
neu_ntohs_p((uint16_t *) (cmd->value.value.bytes + 2));
break;
case NEU_DATATAG_ENDIAN_BB32:
cmd->value.value.u32 = htonl(cmd->value.value.u32);
break;
case NEU_DATATAG_ENDIAN_BL32:
cmd->value.value.u32 = htonl(cmd->value.value.u32);
neu_ntohs_p((uint16_t *) cmd->value.value.bytes);
neu_ntohs_p((uint16_t *) (cmd->value.value.bytes + 2));
break;
case NEU_DATATAG_ENDIAN_LL32:
default:
break;
}
break;
case NEU_TYPE_FLOAT:
if (NEU_TYPE_FLOAT != cmd->value.type) {
cmd->value.value.f32 = (float) cmd->value.value.d64;
cmd->value.type = NEU_TYPE_FLOAT;
}
switch (tag->option.value32.endian) {
case NEU_DATATAG_ENDIAN_LB32:
neu_ntohs_p((uint16_t *) cmd->value.value.bytes);
neu_ntohs_p((uint16_t *) (cmd->value.value.bytes + 2));
break;
case NEU_DATATAG_ENDIAN_BB32:
cmd->value.value.u32 = htonl(cmd->value.value.u32);
break;
case NEU_DATATAG_ENDIAN_BL32:
cmd->value.value.u32 = htonl(cmd->value.value.u32);
neu_ntohs_p((uint16_t *) cmd->value.value.bytes);
neu_ntohs_p((uint16_t *) (cmd->value.value.bytes + 2));
break;
case NEU_DATATAG_ENDIAN_LL32:
default:
break;
}
break;
case NEU_TYPE_DOUBLE:
case NEU_TYPE_UINT64:
case NEU_TYPE_INT64:
case NEU_TYPE_LWORD:
switch (tag->option.value64.endian) {
case NEU_DATATAG_ENDIAN_B64:
cmd->value.value.u64 = neu_ntohll(cmd->value.value.u64);
break;
case NEU_DATATAG_ENDIAN_L64:
default:
break;
}
break;
default:
assert(false);
break;
}
fix_value(tag, cmd->value.type, &cmd->value);
if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC)) {
neu_driver_cache_update(g->driver->cache, g->name, tag->name,
@@ -467,6 +574,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
free(req);
} else {
to_be_write_tag_t wtag = { 0 };
wtag.single = true;
wtag.req = (void *) req;
wtag.value = cmd->value.value;
wtag.tag = neu_tag_dup(tag);
@@ -593,7 +701,15 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
utarray_foreach(find->wt_tags, to_be_write_tag_t *, tag)
{
neu_tag_free(tag->tag);
if (tag->single) {
neu_tag_free(tag->tag);
} else {
utarray_foreach(tag->tvs, neu_plugin_tag_value_t *, tv)
{
neu_tag_free(tv->tag);
}
utarray_free(tag->tvs);
}
}
utarray_free(find->static_tags);
@@ -964,10 +1080,20 @@ static int read_callback(void *usr_data)
pthread_mutex_lock(&group->wt_mtx);
utarray_foreach(group->wt_tags, to_be_write_tag_t *, wtag)
{
group->driver->adapter.module->intf_funs->driver.write_tag(
group->driver->adapter.plugin, (void *) wtag->req, wtag->tag,
wtag->value);
neu_tag_free(wtag->tag);
if (wtag->single) {
group->driver->adapter.module->intf_funs->driver.write_tag(
group->driver->adapter.plugin, (void *) wtag->req, wtag->tag,
wtag->value);
neu_tag_free(wtag->tag);
} else {
group->driver->adapter.module->intf_funs->driver.write_tags(
group->driver->adapter.plugin, (void *) wtag->req, wtag->tvs);
utarray_foreach(wtag->tvs, neu_plugin_tag_value_t *, tv)
{
neu_tag_free(tv->tag);
}
utarray_free(wtag->tvs);
}
}
utarray_clear(group->wt_tags);
pthread_mutex_unlock(&group->wt_mtx);
src/adapter/driver/driver_internal.h
+2
View File
@@ -38,6 +38,8 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
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);
int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
uint32_t interval);
src/core/manager.c
+1
View File
@@ -786,6 +786,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_GET_NODE_SETTING:
case NEU_REQ_READ_GROUP:
case NEU_REQ_WRITE_TAG:
case NEU_REQ_WRITE_TAGS:
case NEU_REQ_GET_NODE_STATE:
case NEU_REQ_GET_TAG:
case NEU_REQ_GET_NDRIVER_TAGS:
src/parser/neu_json_rw.c
+90
View File
@@ -181,6 +181,96 @@ void neu_json_decode_write_req_free(neu_json_write_req_t *req)
free(req);
}
int neu_json_decode_write_tags_req(char * buf,
neu_json_write_tags_req_t **result)
{
int ret = 0;
void * json_obj = NULL;
neu_json_write_tags_req_t *req =
calloc(1, sizeof(neu_json_write_tags_req_t));
if (req == NULL) {
return -1;
}
json_obj = neu_json_decode_new(buf);
neu_json_elem_t req_elems[] = {
{
.name = "node",
.t = NEU_JSON_STR,
},
{
.name = "group",
.t = NEU_JSON_STR,
},
{
.name = "tags",
.t = NEU_JSON_OBJECT,
},
};
ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
req_elems);
if (ret != 0) {
free(req);
neu_json_decode_free(json_obj);
return -1;
}
req->node = req_elems[0].v.val_str;
req->group = req_elems[1].v.val_str;
req->n_tag = neu_json_decode_array_size_by_json(json_obj, "tags");
if (req->n_tag <= 0) {
free(req);
neu_json_decode_free(json_obj);
return -1;
}
req->tags = calloc(req->n_tag, sizeof(neu_json_write_tags_elem_t));
for (int i = 0; i < req->n_tag; i++) {
neu_json_elem_t v_elems[] = {
{
.name = "tag",
.t = NEU_JSON_STR,
},
{
.name = "value",
.t = NEU_JSON_VALUE,
},
};
ret = neu_json_decode_array_by_json(
json_obj, "tags", i, NEU_JSON_ELEM_SIZE(v_elems), v_elems);
if (ret != 0) {
free(req->tags);
free(req);
neu_json_decode_free(json_obj);
}
req->tags[i].tag = v_elems[0].v.val_str;
req->tags[i].t = v_elems[1].t;
req->tags[i].value = v_elems[1].v;
}
*result = req;
neu_json_decode_free(json_obj);
return 0;
}
void neu_json_decode_write_tags_req_free(neu_json_write_tags_req_t *req)
{
free(req->group);
free(req->node);
for (int i = 0; i < req->n_tag; i++) {
free(req->tags[i].tag);
if (req->tags[i].t == NEU_JSON_STR) {
free(req->tags[i].value.val_str);
}
}
free(req->tags);
free(req);
}
int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
{
int ret = 0;
src/utils/http.c
+1
View File
@@ -350,6 +350,7 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_COMMAND_EXECUTION_FAILED:
case NEU_ERR_TEMPLATE_NOT_FOUND:
case NEU_ERR_PLUGIN_NOT_FOUND:
case NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS:
status = NNG_HTTP_STATUS_NOT_FOUND;
break;
case NEU_ERR_NODE_EXIST: