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 & modbus tcp): test read tag

Browse Source
This commit is contained in:
hxy7yx
2024-06-06 16:14:27 +08:00
parent 9f291833f1
commit 4a0835cf40
20 changed files with 718 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+4
View File
@@ -68,6 +68,10 @@ typedef struct adapter_callbacks {
void (*scan_tags_response)(neu_adapter_t *adapter, void *r,
int error, void *tags, neu_type_e type,
bool is_array);
void (*test_read_tag_response)(neu_adapter_t *adapter, void *r,
neu_json_type_e t, neu_type_e type,
neu_json_value_u value,
int64_t error);
} driver;
};
} adapter_callbacks_t;
include/neuron/errcodes.h
+21 -20
View File
@@ -118,26 +118,27 @@ typedef enum {
NEU_ERR_TEMPLATE_NAME_TOO_LONG = 2502,
// common error codes for plugins
NEU_ERR_PLUGIN_READ_FAILURE = 3000,
NEU_ERR_PLUGIN_WRITE_FAILURE = 3001,
NEU_ERR_PLUGIN_DISCONNECTED = 3002,
NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ = 3003,
NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE = 3004,
NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH = 3007,
NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED = 3008,
NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE = 3009,
NEU_ERR_PLUGIN_NOT_RUNNING = 3010,
NEU_ERR_PLUGIN_TAG_NOT_READY = 3011,
NEU_ERR_PLUGIN_PACKET_OUT_OF_ORDER = 3012,
NEU_ERR_PLUGIN_NAME_TOO_LONG = 3013,
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_PLUGIN_NOT_SUPPORT_READ_SYNC = 3018,
NEU_ERR_PLUGIN_TYPE_NOT_SUPPORT = 3019,
NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE = 3020,
NEU_ERR_PLUGIN_NOT_SUPPORT_SCAN_TAGS = 3021,
NEU_ERR_PLUGIN_READ_FAILURE = 3000,
NEU_ERR_PLUGIN_WRITE_FAILURE = 3001,
NEU_ERR_PLUGIN_DISCONNECTED = 3002,
NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ = 3003,
NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE = 3004,
NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH = 3007,
NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED = 3008,
NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE = 3009,
NEU_ERR_PLUGIN_NOT_RUNNING = 3010,
NEU_ERR_PLUGIN_TAG_NOT_READY = 3011,
NEU_ERR_PLUGIN_PACKET_OUT_OF_ORDER = 3012,
NEU_ERR_PLUGIN_NAME_TOO_LONG = 3013,
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_PLUGIN_NOT_SUPPORT_READ_SYNC = 3018,
NEU_ERR_PLUGIN_TYPE_NOT_SUPPORT = 3019,
NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE = 3020,
NEU_ERR_PLUGIN_NOT_SUPPORT_SCAN_TAGS = 3021,
NEU_ERR_PLUGIN_NOT_SUPPORT_TEST_READ_TAG = 3022,
NEU_ERR_MQTT_FAILURE = 4000,
NEU_ERR_MQTT_NO_CERTFILESET = 4001,
include/neuron/json/neu_json_rw.h
+20
View File
@@ -181,6 +181,26 @@ void neu_json_metas_to_json_paginate(
neu_tag_meta_t *metas, int n_meta,
neu_json_read_paginate_resp_tag_t *json_tag);
typedef struct {
char * driver;
char * group;
char * tag;
char * address;
neu_attribute_e attribute;
neu_type_e type;
uint8_t precision;
double decimal;
double bias;
neu_datatag_addr_option_u option;
} neu_json_test_read_tag_req_t;
int neu_json_decode_test_read_tag_req(char * buf,
neu_json_test_read_tag_req_t **result);
void neu_json_decode_test_read_tag_req_free(neu_json_test_read_tag_req_t *req);
int neu_json_encode_test_read_tag_resp(void *json_object, void *param);
#ifdef __cplusplus
}
#endif
include/neuron/msg.h
+24
View File
@@ -48,6 +48,8 @@ typedef enum neu_reqresp_type {
NEU_RESP_READ_GROUP,
NEU_REQ_READ_GROUP_PAGINATE,
NEU_RESP_READ_GROUP_PAGINATE,
NEU_REQ_TEST_READ_TAG,
NEU_RESP_TEST_READ_TAG,
NEU_REQ_WRITE_TAG,
NEU_REQ_WRITE_TAGS,
NEU_REQ_WRITE_GTAGS,
@@ -129,6 +131,8 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_RESP_READ_GROUP] = "NEU_RESP_READ_GROUP",
[NEU_REQ_READ_GROUP_PAGINATE] = "NEU_REQ_READ_GROUP_PAGINATE",
[NEU_RESP_READ_GROUP_PAGINATE] = "NEU_RESP_READ_GROUP_PAGINATE",
[NEU_REQ_TEST_READ_TAG] = "NEU_REQ_TEST_READ_TAG",
[NEU_RESP_TEST_READ_TAG] = "NEU_RESP_TEST_READ_TAG",
[NEU_REQ_WRITE_TAG] = "NEU_REQ_WRITE_TAG",
[NEU_REQ_WRITE_TAGS] = "NEU_REQ_WRITE_TAGS",
[NEU_REQ_WRITE_GTAGS] = "NEU_REQ_WRITE_GTAGS",
@@ -1152,6 +1156,26 @@ typedef struct {
neu_type_e type;
} neu_scan_tag_attribute_t;
typedef struct neu_req_test_rea_tag {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
char tag[NEU_TAG_NAME_LEN];
char address[NEU_TAG_ADDRESS_LEN];
neu_attribute_e attribute;
neu_type_e type;
uint8_t precision;
double decimal;
double bias;
neu_datatag_addr_option_u option;
} neu_req_test_read_tag_t;
typedef struct {
neu_type_e type;
neu_json_type_e t;
neu_json_value_u value;
int64_t error;
} neu_resp_test_read_tag_t;
#ifdef __cplusplus
}
#endif
include/neuron/plugin.h
+2
View File
@@ -109,6 +109,8 @@ typedef struct neu_plugin_intf_funs {
int n_tag); // create tags by API
int (*del_tags)(neu_plugin_t *plugin, int n_tag);
int (*scan_tags)(neu_plugin_t *plugin, void *req, char *id);
int (*test_read_tag)(neu_plugin_t *plugin, void *req,
neu_datatag_t tag);
} driver;
};
plugins/modbus/modbus_req.c
+250 -5
View File
@@ -37,6 +37,9 @@ struct modbus_write_tags_data {
static void plugin_group_free(neu_plugin_group_t *pgp);
static int process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
uint16_t response_size);
static int process_protocol_buf_test(neu_plugin_t *plugin, void *req,
modbus_point_t *point,
uint16_t response_size);
void modbus_conn_connected(void *data, int fd)
{
@@ -125,10 +128,10 @@ int modbus_stack_read_retry(neu_plugin_t *plugin, struct modbus_group_data *gd,
struct timespec t4 = { 0 };
nanosleep(&t3, &t4);
plog_notice(plugin, "Resend read req. Times:%hu", j + 1);
int ret = modbus_stack_read(plugin->stack, gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].area,
gd->cmd_sort->cmd[i].start_address,
gd->cmd_sort->cmd[i].n_register, response_size);
int ret = modbus_stack_read(
plugin->stack, gd->cmd_sort->cmd[i].slave_id, gd->cmd_sort->cmd[i].area,
gd->cmd_sort->cmd[i].start_address, gd->cmd_sort->cmd[i].n_register,
response_size, false);
return ret;
}
@@ -175,7 +178,7 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
int ret_r = modbus_stack_read(
plugin->stack, gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].area, gd->cmd_sort->cmd[i].start_address,
gd->cmd_sort->cmd[i].n_register, &response_size);
gd->cmd_sort->cmd[i].n_register, &response_size, false);
if (ret_r > 0) {
ret_buf = process_protocol_buf(
plugin, gd->cmd_sort->cmd[i].slave_id, response_size);
@@ -453,6 +456,174 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
return 0;
}
int modbus_value_handle_test(neu_plugin_t *plugin, void *req,
modbus_point_t *point, uint16_t n_byte,
uint8_t *bytes)
{
(void) req;
neu_json_value_u jvalue = { 0 };
neu_json_type_e jtype;
switch (point->area) {
case MODBUS_AREA_HOLD_REGISTER:
case MODBUS_AREA_INPUT_REGISTER:
if (n_byte >= point->n_register * 2) {
jvalue.val_bytes.bytes =
(uint8_t *) malloc(point->n_register * 2 + 1);
memcpy(jvalue.val_bytes.bytes, bytes, point->n_register * 2);
jvalue.val_bytes.length = point->n_register * 2;
}
break;
case MODBUS_AREA_COIL:
case MODBUS_AREA_INPUT: {
neu_value8_u u8 = { .value = bytes[0] };
jvalue.val_bit = neu_value8_get_bit(u8, 0);
} break;
}
switch (point->type) {
case NEU_TYPE_UINT16:
case NEU_TYPE_INT16: {
jtype = NEU_JSON_INT;
uint16_t tmp_val16;
memcpy(&tmp_val16, jvalue.val_bytes.bytes, sizeof(uint16_t));
free(jvalue.val_bytes.bytes);
jvalue.val_int = ntohs(tmp_val16);
break;
}
case NEU_TYPE_FLOAT: {
uint32_t tmp_valf;
memcpy(&tmp_valf, jvalue.val_bytes.bytes, sizeof(uint32_t));
free(jvalue.val_bytes.bytes);
jtype = NEU_JSON_FLOAT;
jvalue.val_int = ntohl(tmp_valf);
break;
}
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32: {
uint32_t tmp_val32;
memcpy(&tmp_val32, jvalue.val_bytes.bytes, sizeof(uint32_t));
free(jvalue.val_bytes.bytes);
jtype = NEU_JSON_INT;
jvalue.val_int = ntohl(tmp_val32);
break;
}
case NEU_TYPE_DOUBLE: {
uint64_t tmp_vald;
memcpy(&tmp_vald, jvalue.val_bytes.bytes, sizeof(uint64_t));
free(jvalue.val_bytes.bytes);
jtype = NEU_JSON_DOUBLE;
jvalue.val_int = neu_ntohll(tmp_vald);
break;
}
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64: {
uint64_t tmp_val64;
memcpy(&tmp_val64, jvalue.val_bytes.bytes, sizeof(uint64_t));
free(jvalue.val_bytes.bytes);
jtype = NEU_JSON_INT;
jvalue.val_int = neu_ntohll(tmp_val64);
break;
}
case NEU_TYPE_BIT: {
switch (point->area) {
case MODBUS_AREA_HOLD_REGISTER:
case MODBUS_AREA_INPUT_REGISTER: {
jtype = NEU_JSON_BIT;
neu_value16_u v16 = { 0 };
v16.value = htons(*(uint16_t *) jvalue.val_bytes.bytes);
free(jvalue.val_bytes.bytes);
jvalue.val_bit = neu_value16_get_bit(v16, point->option.bit.bit);
break;
}
case MODBUS_AREA_COIL:
case MODBUS_AREA_INPUT:
jtype = NEU_JSON_INT;
break;
}
break;
}
case NEU_TYPE_STRING: {
jtype = NEU_JSON_STR;
size_t str_length = point->n_register * 2;
jvalue.val_str = (char *) jvalue.val_bytes.bytes;
jvalue.val_str[str_length] = '\0';
switch (point->option.string.type) {
case NEU_DATATAG_STRING_TYPE_H:
break;
case NEU_DATATAG_STRING_TYPE_L:
neu_datatag_string_ltoh(jvalue.val_str, str_length);
break;
case NEU_DATATAG_STRING_TYPE_D:
break;
case NEU_DATATAG_STRING_TYPE_E:
break;
}
if (!neu_datatag_string_is_utf8(jvalue.val_str, str_length)) {
jvalue.val_str[0] = '?';
jvalue.val_str[1] = 0;
}
break;
}
default:
break;
}
plugin->common.adapter_callbacks->driver.test_read_tag_response(
plugin->common.adapter, req, jtype, point->type, jvalue,
NEU_ERR_SUCCESS);
return 0;
}
int modbus_test_read_tag(neu_plugin_t *plugin, void *req, neu_datatag_t tag)
{
modbus_point_t point = { 0 };
neu_json_value_u error_value;
error_value.val_int = 0;
int err = modbus_tag_to_point(&tag, &point);
if (err != NEU_ERR_SUCCESS) {
plugin->common.adapter_callbacks->driver.test_read_tag_response(
plugin->common.adapter, req, NEU_JSON_INT, NEU_TYPE_ERROR,
error_value, err);
return 0;
}
uint16_t response_size = 0;
int ret = modbus_stack_read(plugin->stack, point.slave_id, point.area,
point.start_address, point.n_register,
&response_size, true);
if (ret <= 0) {
plugin->common.adapter_callbacks->driver.test_read_tag_response(
plugin->common.adapter, req, NEU_JSON_INT, NEU_TYPE_ERROR,
error_value, NEU_ERR_PLUGIN_READ_FAILURE);
return 0;
}
ret = process_protocol_buf_test(plugin, req, &point, response_size);
if (ret == 0) {
plugin->common.adapter_callbacks->driver.test_read_tag_response(
plugin->common.adapter, req, NEU_JSON_INT, NEU_TYPE_ERROR,
error_value, NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE);
return 0;
} else if (ret == -1) {
plugin->common.adapter_callbacks->driver.test_read_tag_response(
plugin->common.adapter, req, NEU_JSON_INT, NEU_TYPE_ERROR,
error_value, NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE);
return 0;
} else if (ret == -2) {
plugin->common.adapter_callbacks->driver.test_read_tag_response(
plugin->common.adapter, req, NEU_JSON_INT, NEU_TYPE_ERROR,
error_value, NEU_ERR_PLUGIN_READ_FAILURE);
return 0;
}
return 0;
}
int modbus_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
neu_value_u value, bool response)
{
@@ -713,4 +884,78 @@ static int process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
}
free(recv_buf);
return ret;
}
static int process_protocol_buf_test(neu_plugin_t *plugin, void *req,
modbus_point_t *point,
uint16_t response_size)
{
uint8_t * recv_buf = calloc(response_size, 1);
neu_protocol_unpack_buf_t pbuf = { 0 };
ssize_t ret = 0;
if (plugin->protocol == MODBUS_PROTOCOL_TCP) {
if (plugin->is_server) {
ret = neu_conn_tcp_server_recv(plugin->conn, plugin->client_fd,
recv_buf,
sizeof(struct modbus_header));
} else {
ret = neu_conn_recv(plugin->conn, recv_buf,
sizeof(struct modbus_header));
}
if (ret == 0 || ret == -1) {
free(recv_buf);
return 0;
}
if (ret == sizeof(struct modbus_header)) {
struct modbus_header *header = (struct modbus_header *) recv_buf;
int ret1 = 0;
if (htons(header->len) >
response_size - sizeof(struct modbus_header)) {
free(recv_buf);
return -1;
}
plog_recv_protocol(plugin, recv_buf, ret);
if (plugin->is_server) {
ret1 = neu_conn_tcp_server_recv(
plugin->conn, plugin->client_fd,
recv_buf + sizeof(struct modbus_header),
htons(header->len));
} else {
ret1 = neu_conn_recv(plugin->conn,
recv_buf + sizeof(struct modbus_header),
htons(header->len));
}
if (ret1 != htons(header->len)) {
free(recv_buf);
return -1;
}
ret += ret1;
} else {
free(recv_buf);
return -1;
}
if (ret > 0) {
if (ret < 512) {
plog_recv_protocol(plugin, recv_buf, ret);
}
neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
int ret_s = modbus_stack_recv_test(plugin, req, point, &pbuf);
if (ret_s == MODBUS_DEVICE_ERR) {
ret = ret_s;
} else {
if (ret != response_size) {
ret = -1;
} else {
ret = ret_s;
}
}
}
}
free(recv_buf);
return ret;
}
plugins/modbus/modbus_req.h
+4
View File
@@ -63,5 +63,9 @@ int modbus_write_tag(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
neu_value_u value);
int modbus_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags);
int modbus_write_resp(void *ctx, void *req, int error);
int modbus_test_read_tag(neu_plugin_t *plugin, void *req, neu_datatag_t tag);
int modbus_value_handle_test(neu_plugin_t *plugin, void *req,
modbus_point_t *point, uint16_t n_byte,
uint8_t *bytes);
#endif
plugins/modbus/modbus_stack.c
+97 -2
View File
@@ -177,9 +177,104 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
return neu_protocol_unpack_buf_used_size(buf);
}
int modbus_stack_recv_test(neu_plugin_t *plugin, void *req,
modbus_point_t * point,
neu_protocol_unpack_buf_t *buf)
{
struct modbus_header header = { 0 };
struct modbus_code code = { 0 };
int ret = 0;
if (plugin->stack->protocol == MODBUS_PROTOCOL_TCP) {
ret = modbus_header_unwrap(buf, &header);
if (ret < 0) {
plog_warn((neu_plugin_t *) plugin->stack->ctx,
"try modbus rtu driver");
}
if (ret <= 0) {
return -1;
}
}
ret = modbus_code_unwrap(buf, &code);
if (ret <= 0) {
return -1;
}
if (code.slave_id != point->slave_id) {
return -1;
}
switch (code.function) {
case MODBUS_READ_COIL:
case MODBUS_READ_INPUT:
case MODBUS_READ_HOLD_REG:
case MODBUS_READ_INPUT_REG: {
struct modbus_data data = { 0 };
uint8_t * bytes = NULL;
ret = modbus_data_unwrap(buf, &data);
if (ret <= 0) {
return -1;
}
if (data.n_byte == 0xff) {
bytes = neu_protocol_unpack_buf(buf,
header.len -
sizeof(struct modbus_code) -
sizeof(struct modbus_data));
if (bytes == NULL) {
return -1;
}
modbus_value_handle_test(plugin, req, point, data.n_byte, bytes);
} else {
bytes = neu_protocol_unpack_buf(buf, data.n_byte);
if (bytes == NULL) {
return -1;
}
modbus_value_handle_test(plugin, req, point, data.n_byte, bytes);
}
break;
}
case MODBUS_WRITE_S_COIL:
case MODBUS_WRITE_M_HOLD_REG:
case MODBUS_WRITE_M_COIL: {
struct modbus_address address = { 0 };
ret = modbus_address_unwrap(buf, &address);
if (ret <= 0) {
return -1;
}
break;
}
case MODBUS_READ_COIL_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_READ_INPUT_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_READ_HOLD_REG_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_READ_INPUT_REG_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_S_COIL_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_S_HOLD_REG_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_M_HOLD_REG_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_M_COIL_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_S_HOLD_REG:
break;
default:
return -1;
}
return neu_protocol_unpack_buf_used_size(buf);
}
int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
enum modbus_area area, uint16_t start_address,
uint16_t n_reg, uint16_t *response_size)
uint16_t n_reg, uint16_t *response_size, bool is_test)
{
static __thread uint8_t buf[16] = { 0 };
static __thread neu_protocol_pack_buf_t pbuf = { 0 };
@@ -229,7 +324,7 @@ int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
ret = stack->send_fn(stack->ctx, neu_protocol_pack_buf_used_size(&pbuf),
neu_protocol_pack_buf_get(&pbuf));
if (ret <= 0) {
if (ret <= 0 && !is_test) {
stack->value_fn(stack->ctx, 0, 0, NULL, NEU_ERR_PLUGIN_DISCONNECTED);
plog_warn((neu_plugin_t *) stack->ctx, "send read req fail, %hhu!%hu",
slave_id, start_address);
plugins/modbus/modbus_stack.h
+5 -1
View File
@@ -24,6 +24,7 @@
#include <neuron.h>
#include "modbus.h"
#include "modbus_point.h"
typedef struct modbus_stack modbus_stack_t;
@@ -45,10 +46,13 @@ void modbus_stack_destroy(modbus_stack_t *stack);
int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
neu_protocol_unpack_buf_t *buf);
int modbus_stack_recv_test(neu_plugin_t *plugin, void *req,
modbus_point_t * point,
neu_protocol_unpack_buf_t *buf);
int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
enum modbus_area area, uint16_t start_address,
uint16_t n_reg, uint16_t *response_size);
uint16_t n_reg, uint16_t *response_size, bool is_test);
int modbus_stack_write(modbus_stack_t *stack, void *req, uint8_t slave_id,
enum modbus_area area, uint16_t start_address,
uint16_t n_reg, uint8_t *bytes, uint8_t n_byte,
plugins/modbus/modbus_tcp.c
+9
View File
@@ -43,6 +43,8 @@ static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
neu_value_u value);
static int driver_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags);
static int driver_test_read_tag(neu_plugin_t *plugin, void *req,
neu_datatag_t tag);
static const neu_plugin_intf_funs_t plugin_intf_funs = {
.open = driver_open,
@@ -59,6 +61,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
.driver.write_tag = driver_write,
.driver.tag_validator = driver_tag_validator,
.driver.write_tags = driver_write_tags,
.driver.test_read_tag = driver_test_read_tag,
.driver.add_tags = NULL,
.driver.load_tags = NULL,
.driver.del_tags = NULL,
@@ -280,4 +283,10 @@ static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
static int driver_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags)
{
return modbus_write_tags(plugin, req, tags);
}
static int driver_test_read_tag(neu_plugin_t *plugin, void *req,
neu_datatag_t tag)
{
return modbus_test_read_tag(plugin, req, tag);
}
plugins/restful/handle.c
+9
View File
@@ -76,6 +76,9 @@ static struct neu_http_handler cors_handler[] = {
{
.url = "/api/v2/read/paginate",
},
{
.url = "/api/v2/read/test",
},
{
.url = "/api/v2/write",
},
@@ -272,6 +275,12 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/read/paginate",
.value.handler = handle_read_paginate,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/read/test",
.value.handler = handle_test_read_tag,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
plugins/restful/rest.c
+4
View File
@@ -228,6 +228,10 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
handle_read_paginate_resp(header->ctx,
(neu_resp_read_group_paginate_t *) data);
break;
case NEU_RESP_TEST_READ_TAG:
handle_test_read_tag_resp(header->ctx,
(neu_resp_test_read_tag_t *) data);
break;
case NEU_RESP_SCAN_TAGS:
handle_scan_tags_resp(header->ctx, (neu_resp_scan_tags_t *) data);
break;
plugins/restful/rw_handle.c
+64
View File
@@ -135,6 +135,62 @@ void handle_read_paginate(nng_aio *aio)
})
}
void handle_test_read_tag(nng_aio *aio)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_test_read_tag_req_t, neu_json_decode_test_read_tag_req, {
int ret = 0;
neu_reqresp_head_t header = { 0 };
neu_req_test_read_tag_t cmd = { 0 };
int err_type;
header.ctx = aio;
header.type = NEU_REQ_TEST_READ_TAG;
if (req->driver && strlen(req->driver) >= NEU_NODE_NAME_LEN) {
err_type = NEU_ERR_NODE_NAME_TOO_LONG;
goto error;
}
if (req->group && strlen(req->group) >= NEU_GROUP_NAME_LEN) {
err_type = NEU_ERR_GROUP_NAME_TOO_LONG;
goto error;
}
if (req->tag && strlen(req->tag) >= NEU_TAG_NAME_LEN) {
err_type = NEU_ERR_TAG_NAME_TOO_LONG;
goto error;
}
if (req->address && strlen(req->address) >= NEU_TAG_ADDRESS_LEN) {
err_type = NEU_ERR_TAG_ADDRESS_TOO_LONG;
goto error;
}
strcpy(cmd.driver, req->driver);
strcpy(cmd.group, req->group);
strcpy(cmd.tag, req->tag);
strcpy(cmd.address, req->address);
cmd.attribute = req->attribute;
cmd.type = req->type;
cmd.precision = req->precision;
cmd.decimal = req->decimal;
cmd.bias = req->bias;
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);
});
}
goto success;
error:
NEU_JSON_RESPONSE_ERROR(
err_type, { neu_http_response(aio, err_type, result_error); });
success:;
})
}
void handle_write(nng_aio *aio)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
@@ -549,4 +605,12 @@ void handle_read_paginate_resp(nng_aio * aio,
neu_http_ok(aio, result);
free(api_res.tags);
free(result);
}
void handle_test_read_tag_resp(nng_aio *aio, neu_resp_test_read_tag_t *resp)
{
char *result = NULL;
neu_json_encode_by_fn(resp, neu_json_encode_test_read_tag_resp, &result);
neu_http_ok(aio, result);
free(result);
}
plugins/restful/rw_handle.h
+2
View File
@@ -28,11 +28,13 @@ void handle_rw_init();
void handle_rw_uninit();
void handle_read(nng_aio *aio);
void handle_read_paginate(nng_aio *aio);
void handle_test_read_tag(nng_aio *aio);
void handle_write(nng_aio *aio);
void handle_write_tags(nng_aio *aio);
void handle_write_gtags(nng_aio *aio);
void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp);
void handle_read_paginate_resp(nng_aio * aio,
neu_resp_read_group_paginate_t *resp);
void handle_test_read_tag_resp(nng_aio *aio, neu_resp_test_read_tag_t *resp);
#endif
src/adapter/adapter.c
+16
View File
@@ -450,6 +450,11 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
strcpy(pheader->receiver, cmd->driver);
break;
}
case NEU_REQ_TEST_READ_TAG: {
neu_req_test_read_tag_t *cmd = (neu_req_test_read_tag_t *) data;
strcpy(pheader->receiver, cmd->driver);
break;
}
case NEU_REQ_WRITE_TAG: {
neu_req_write_tag_t *cmd = (neu_req_write_tag_t *) data;
strcpy(pheader->receiver, cmd->driver);
@@ -1226,6 +1231,17 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_msg_free(msg);
break;
}
case NEU_REQ_TEST_READ_TAG: {
neu_adapter_driver_test_read_tag((neu_adapter_driver_t *) adapter,
header);
break;
}
case NEU_RESP_TEST_READ_TAG: {
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
neu_msg_free(msg);
break;
}
default:
nlog_warn("adapter: %s recv msg type error, type: %s", adapter->name,
neu_reqresp_type_string(header->type));
src/adapter/driver/driver.c
+58
View File
@@ -343,6 +343,23 @@ static void scan_tags_response(neu_adapter_t *adapter, void *r, neu_error error,
adapter->cb_funs.response(adapter, req, &value);
}
static void test_read_tag_response(neu_adapter_t *adapter, void *r,
neu_json_type_e t, neu_type_e type,
neu_json_value_u value, int64_t error)
{
neu_reqresp_head_t * req = (neu_reqresp_head_t *) r;
neu_resp_test_read_tag_t dvalue = {
.t = t,
.type = type,
.value = value,
.error = error,
};
req->type = NEU_RESP_TEST_READ_TAG;
nlog_notice("test reading tag response <%p>", req->ctx);
adapter->cb_funs.response(adapter, req, &dvalue);
}
neu_adapter_driver_t *neu_adapter_driver_create()
{
neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t));
@@ -354,6 +371,8 @@ neu_adapter_driver_t *neu_adapter_driver_create()
driver->adapter.cb_funs.driver.update_im = update_im;
driver->adapter.cb_funs.driver.update_with_meta = update_with_meta;
driver->adapter.cb_funs.driver.scan_tags_response = scan_tags_response;
driver->adapter.cb_funs.driver.test_read_tag_response =
test_read_tag_response;
return driver;
}
@@ -2652,3 +2671,42 @@ void neu_adapter_driver_scan_tags(neu_adapter_driver_t *driver,
driver->adapter.module->intf_funs->driver.scan_tags(driver->adapter.plugin,
req, cmd->id);
}
void neu_adapter_driver_test_read_tag(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req)
{
neu_json_value_u error_value;
error_value.val_int = 0;
if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) {
driver->adapter.cb_funs.driver.test_read_tag_response(
&driver->adapter, req, NEU_JSON_INT, NEU_TYPE_ERROR, error_value,
NEU_ERR_PLUGIN_NOT_RUNNING);
nlog_warn("%s not running", driver->adapter.name);
return;
}
if (driver->adapter.module->intf_funs->driver.test_read_tag == NULL) {
driver->adapter.cb_funs.driver.test_read_tag_response(
&driver->adapter, req, NEU_JSON_INT, NEU_TYPE_ERROR, error_value,
NEU_ERR_PLUGIN_NOT_SUPPORT_TEST_READ_TAG);
nlog_warn("%s not support testing reading", driver->adapter.name);
return;
}
neu_req_test_read_tag_t *cmd = (neu_req_test_read_tag_t *) &req[1];
neu_datatag_t datatag;
datatag.name = cmd->tag;
datatag.address = cmd->address;
datatag.attribute = cmd->attribute;
datatag.type = cmd->type;
datatag.precision = cmd->precision;
datatag.decimal = cmd->decimal;
datatag.bias = cmd->bias;
datatag.option = cmd->option;
driver->adapter.module->intf_funs->driver.test_read_tag(
driver->adapter.plugin, req, datatag);
}
src/adapter/driver/driver_internal.h
+2
View File
@@ -35,6 +35,8 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req);
void neu_adapter_driver_read_group_paginate(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req);
void neu_adapter_driver_test_read_tag(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req);
void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req);
src/base/msg_internal.h
+2
View File
@@ -36,6 +36,8 @@ extern "C" {
XX(NEU_RESP_READ_GROUP, neu_resp_read_group_t) \
XX(NEU_REQ_READ_GROUP_PAGINATE, neu_req_read_group_paginate_t) \
XX(NEU_RESP_READ_GROUP_PAGINATE, neu_resp_read_group_paginate_t) \
XX(NEU_REQ_TEST_READ_TAG, neu_req_test_read_tag_t) \
XX(NEU_RESP_TEST_READ_TAG, neu_resp_test_read_tag_t) \
XX(NEU_REQ_WRITE_TAG, neu_req_write_tag_t) \
XX(NEU_REQ_WRITE_TAGS, neu_req_write_tags_t) \
XX(NEU_REQ_WRITE_GTAGS, neu_req_write_gtags_t) \
src/core/manager.c
+2
View File
@@ -1041,6 +1041,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_PRGFILE_PROCESS:
case NEU_REQ_PRGFILE_UPLOAD:
case NEU_REQ_SCAN_TAGS:
case NEU_REQ_TEST_READ_TAG:
case NEU_REQ_GET_NODE_STATE: {
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
NULL) {
@@ -1250,6 +1251,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_RESP_ERROR:
case NEU_RESP_READ_GROUP:
case NEU_RESP_READ_GROUP_PAGINATE:
case NEU_RESP_TEST_READ_TAG:
case NEU_RESP_PRGFILE_PROCESS:
case NEU_RESP_SCAN_TAGS:
forward_msg(manager, header, header->receiver);
src/parser/neu_json_rw.c
+123
View File
@@ -30,6 +30,7 @@
#include "json/json.h"
#include "define.h"
#include "msg.h"
#include "tag.h"
#include "json/neu_json_rw.h"
@@ -182,6 +183,31 @@ int neu_json_encode_read_paginate_resp(void *json_object, void *param)
return ret;
}
int neu_json_encode_test_read_tag_resp(void *json_object, void *param)
{
int ret = 0;
neu_resp_test_read_tag_t *resp = (neu_resp_test_read_tag_t *) param;
neu_json_elem_t resp_elem = { 0 };
if (resp->type == NEU_TYPE_ERROR) {
resp_elem.name = "error";
resp_elem.t = NEU_JSON_INT;
resp_elem.v.val_int = resp->error;
} else {
resp_elem.name = "value";
resp_elem.t = resp->t;
resp_elem.v = resp->value;
}
ret = neu_json_encode_field(json_object, &resp_elem, 1);
if (resp->t == NEU_JSON_STR) {
free(resp->value.val_str);
}
return ret;
}
int neu_json_encode_read_resp1(void *json_object, void *param)
{
int ret = 0;
@@ -718,6 +744,103 @@ void neu_json_decode_read_paginate_req_free(neu_json_read_paginate_req_t *req)
free(req);
}
int neu_json_decode_test_read_tag_req(char * buf,
neu_json_test_read_tag_req_t **result)
{
int ret = 0;
void * json_obj = NULL;
neu_json_test_read_tag_req_t *req =
calloc(1, sizeof(neu_json_test_read_tag_req_t));
if (req == NULL) {
return -1;
}
json_obj = neu_json_decode_new(buf);
if (NULL == json_obj) {
free(req);
return -1;
}
neu_json_elem_t req_elems[] = {
{
.name = "driver",
.t = NEU_JSON_STR,
},
{
.name = "group",
.t = NEU_JSON_STR,
},
{
.name = "tag",
.t = NEU_JSON_STR,
},
{
.name = "address",
.t = NEU_JSON_STR,
},
{
.name = "attribute",
.t = NEU_JSON_INT,
},
{
.name = "type",
.t = NEU_JSON_INT,
},
{
.name = "precision",
.t = NEU_JSON_INT,
},
{
.name = "decimal",
.t = NEU_JSON_DOUBLE,
},
{
.name = "bias",
.t = NEU_JSON_DOUBLE,
},
};
ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
req_elems);
if (ret != 0) {
goto decode_fail;
}
req->driver = req_elems[0].v.val_str;
req->group = req_elems[1].v.val_str;
req->tag = req_elems[2].v.val_str;
req->address = req_elems[3].v.val_str;
req->attribute = req_elems[4].v.val_int;
req->type = req_elems[5].v.val_int;
req->precision = req_elems[6].v.val_int;
req->decimal = req_elems[7].v.val_double;
req->bias = req_elems[8].v.val_double;
*result = req;
goto decode_exit;
decode_fail:
free(req);
free(req_elems[0].v.val_str);
free(req_elems[1].v.val_str);
free(req_elems[2].v.val_str);
free(req_elems[3].v.val_str);
ret = -1;
decode_exit:
if (json_obj != NULL) {
neu_json_decode_free(json_obj);
}
return ret;
}
void neu_json_decode_test_read_tag_req_free(neu_json_test_read_tag_req_t *req)
{
free(req->driver);
free(req->group);
free(req->tag);
free(req->address);
free(req);
}
int neu_json_encode_read_periodic_resp(void *json_object, void *param)
{
int ret = 0;