GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-18 13:41:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

Improve OPCUA driver error code processing for reading and writing data

Browse Source
This commit is contained in:
gc87
2021-09-29 14:11:18 +08:00
committed by Parker Liu
parent 75e863ab9e
commit 02ce6dc213
7 changed files with 585 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/command_rw.c
+368 -11
View File
@@ -23,8 +23,8 @@
#include "command_rw.h"
void command_read_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req)
void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req)
{
log_info("READ uuid:%s, node id:%u", mqtt->uuid, req->node_id);
@@ -46,24 +46,381 @@ void command_read_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
UNUSED(req_id);
}
void command_read_response(neu_plugin_t *plugin, neu_taggrp_config_t *config,
neu_data_val_t *array)
static int wrap_read_response_json_object(neu_fixed_array_t * array,
neu_parse_read_res_t *json)
{
neu_int_val_t * int_val;
neu_data_val_t *val;
int length = array->length;
json->n_tag = length - 1;
if (0 == json->n_tag) {
return -1;
}
json->tags = (neu_parse_read_res_tag_t *) calloc(
length - 1, sizeof(neu_parse_read_res_tag_t));
for (int i = 1; i < length; i++) {
int_val = neu_fixed_array_get(array, i);
val = int_val->val;
neu_dtype_e type = neu_dvalue_get_type(val);
json->tags[i - 1].name = strdup((char *) "");
json->tags[i - 1].tag_id = int_val->key;
switch (type) {
case NEU_DTYPE_BYTE: {
uint8_t value;
neu_dvalue_get_uint8(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
break;
}
case NEU_DTYPE_INT8: {
int8_t value;
neu_dvalue_get_int8(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
break;
}
case NEU_DTYPE_INT16: {
int16_t value;
neu_dvalue_get_int16(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
break;
}
case NEU_DTYPE_INT32: {
int32_t value;
neu_dvalue_get_int32(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
break;
}
case NEU_DTYPE_INT64: {
int64_t value;
neu_dvalue_get_int64(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
break;
}
case NEU_DTYPE_UINT8: {
uint8_t value;
neu_dvalue_get_uint8(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
break;
}
case NEU_DTYPE_UINT16: {
uint16_t value;
neu_dvalue_get_uint16(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
break;
}
case NEU_DTYPE_UINT32: {
uint32_t value;
neu_dvalue_get_uint32(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
break;
}
case NEU_DTYPE_UINT64: {
uint64_t value;
neu_dvalue_get_uint64(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
break;
}
case NEU_DTYPE_FLOAT: {
float value;
neu_dvalue_get_float(val, &value);
json->tags[i - 1].t = NEU_JSON_FLOAT;
json->tags[i - 1].value.val_float = value;
break;
}
case NEU_DTYPE_DOUBLE: {
double value;
neu_dvalue_get_double(val, &value);
json->tags[i - 1].t = NEU_JSON_DOUBLE;
json->tags[i - 1].value.val_double = value;
break;
}
case NEU_DTYPE_CSTR: {
char *value;
neu_dvalue_get_ref_cstr(val, &value);
json->tags[i - 1].t = NEU_JSON_STR;
json->tags[i - 1].value.val_str = value;
break;
}
default:
break;
}
}
return 0;
}
static void clean_read_response_json_object(neu_parse_read_res_t *json)
{
if (NULL == json) {
return;
}
if (NULL == json->tags) {
return;
}
int length = json->n_tag;
for (int i = 0; i < length; i++) {
if (NULL != json->tags[i].name) {
free(json->tags[i].name);
}
}
free(json->tags);
}
char *command_read_once_response(neu_plugin_t *plugin, const char *uuid,
neu_data_val_t *resp_val)
{
UNUSED(plugin);
UNUSED(config);
UNUSED(array);
UNUSED(uuid);
UNUSED(resp_val);
neu_fixed_array_t *array;
int rc = neu_dvalue_get_ref_array(resp_val, &array);
if (0 != rc) {
log_info("Get array ref error");
return NULL;
}
if (0 >= array->length) {
return NULL;
}
neu_int_val_t * int_val;
neu_data_val_t *val_err;
int32_t error;
int_val = neu_fixed_array_get(array, 0);
val_err = int_val->val;
neu_dvalue_get_errorcode(val_err, &error);
neu_parse_read_res_t json;
memset(&json, 0, sizeof(neu_parse_read_res_t));
rc = wrap_read_response_json_object(array, &json);
if (0 != rc) {
return NULL;
}
char *json_str = NULL;
rc = neu_json_encode_by_fn(&json, neu_parse_encode_read, &json_str);
clean_read_response_json_object(&json);
if (0 != rc) {
log_info("Json string parse error:%d", rc);
return NULL;
}
return json_str;
}
char *command_read_cycle_response(neu_plugin_t * plugin,
neu_data_val_t *resp_val)
{
// TODO: Generate UUID
return command_read_once_response(plugin, "", resp_val);
}
static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id,
neu_data_val_t *write_val)
{
vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER);
int rc = common_node_id_exist(&nodes, dest_node_id);
vector_uninit(&nodes);
if (0 != rc) {
return -1;
}
vector_t configs = neu_system_get_group_configs(plugin, dest_node_id);
neu_taggrp_config_t *config;
neu_taggrp_config_t *c;
config = *(neu_taggrp_config_t **) vector_get(&configs, 0);
c = (neu_taggrp_config_t *) neu_taggrp_cfg_ref(config);
uint32_t req_id =
neu_plugin_send_write_cmd(plugin, dest_node_id, c, write_val);
GROUP_CONFIGS_UINIT(configs);
return req_id;
}
static void set_int_type_value(neu_data_val_t *val, const int64_t data)
{
if (INT8_MIN <= data && INT8_MAX >= data) {
val = neu_dvalue_new(NEU_DTYPE_INT8);
neu_dvalue_set_int8(val, (int8_t) data);
return;
}
if (INT16_MIN <= data && INT16_MAX >= data) {
val = neu_dvalue_new(NEU_DTYPE_INT16);
neu_dvalue_set_int16(val, (int16_t) data);
return;
}
if (INT32_MIN <= data && INT32_MAX >= data) {
val = neu_dvalue_new(NEU_DTYPE_INT32);
neu_dvalue_set_int32(val, (int32_t) data);
return;
}
if (INT64_MIN <= data && INT64_MAX >= data) {
val = neu_dvalue_new(NEU_DTYPE_INT64);
neu_dvalue_set_int64(val, (int64_t) data);
return;
}
}
void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_write_req_t *write_req)
{
UNUSED(plugin);
UNUSED(mqtt);
UNUSED(write_req);
log_info("WRITE uuid:%s", mqtt->uuid);
neu_data_val_t * write_val;
neu_fixed_array_t *array;
write_val = neu_dvalue_unit_new();
if (write_val == NULL) {
log_error("Failed to allocate data value for write data");
return;
}
array = neu_fixed_array_new(write_req->n_tag, sizeof(neu_int_val_t));
if (array == NULL) {
log_error("Failed to allocate array for write data");
neu_dvalue_free(write_val);
return;
}
neu_int_val_t int_val;
neu_data_val_t *val;
for (int i = 0; i < write_req->n_tag; i++) {
enum neu_json_type type = write_req->tags[i].t;
union neu_json_value value = write_req->tags[i].value;
neu_datatag_id_t id = write_req->tags[i].tag_id;
switch (type) {
case NEU_JSON_INT: {
set_int_type_value(val, value.val_int);
break;
}
case NEU_JSON_BIT: {
val = neu_dvalue_new(NEU_DTYPE_BIT);
neu_dvalue_set_bit(val, value.val_bit);
break;
}
case NEU_JSON_STR: {
val = neu_dvalue_new(NEU_DTYPE_CSTR);
neu_dvalue_set_cstr(val, strdup(value.val_str));
break;
}
case NEU_JSON_DOUBLE: {
val = neu_dvalue_new(NEU_DTYPE_DOUBLE);
neu_dvalue_set_double(val, value.val_double);
break;
}
case NEU_JSON_FLOAT: {
val = neu_dvalue_new(NEU_DTYPE_FLOAT);
neu_dvalue_set_float(val, value.val_float);
break;
}
case NEU_JSON_BOOL: {
val = neu_dvalue_new(NEU_DTYPE_BOOL);
neu_dvalue_set_bool(val, value.val_bool);
break;
}
default:
break;
}
neu_int_val_init(&int_val, id, val);
neu_fixed_array_set(array, i, (void *) &int_val);
}
neu_dvalue_init_move_array(write_val, NEU_DTYPE_INT_VAL, array);
write_command(plugin, write_req->node_id, write_val);
neu_dvalue_free(write_val);
}
void command_write_response(neu_plugin_t *plugin, neu_data_val_t *array)
static int wrap_write_response_json_object(neu_fixed_array_t * array,
neu_parse_read_res_t *json)
{
neu_int_val_t * int_val;
neu_data_val_t *val;
int length = array->length;
json->n_tag = length - 1;
json->tags = (neu_parse_read_res_tag_t *) calloc(
length - 1, sizeof(neu_parse_read_res_tag_t));
for (int i = 1; i < length; i++) {
int_val = neu_fixed_array_get(array, i);
val = int_val->val;
json->tags[i - 1].name = strdup((char *) "");
json->tags[i - 1].tag_id = int_val->key;
int32_t value;
neu_dvalue_get_int32(val, &value);
json->tags[i - 1].t = NEU_JSON_INT;
json->tags[i - 1].value.val_int = value;
}
return 0;
}
static void clean_write_response_json_object(neu_parse_read_res_t *json)
{
if (NULL == json) {
return;
}
if (NULL == json->tags) {
return;
}
int length = json->n_tag;
for (int i = 0; i < length; i++) {
if (NULL != json->tags[i].name) {
free(json->tags[i].name);
}
}
free(json->tags);
}
char *command_write_response(neu_plugin_t *plugin, const char *uuid,
neu_data_val_t *resp_val)
{
UNUSED(plugin);
UNUSED(array);
UNUSED(uuid);
neu_fixed_array_t *array;
neu_dvalue_get_ref_array(resp_val, &array);
neu_int_val_t * int_val;
neu_data_val_t *val_err;
int32_t error;
int_val = neu_fixed_array_get(array, 0);
val_err = int_val->val;
neu_dvalue_get_errorcode(val_err, &error);
neu_parse_read_res_t json;
wrap_write_response_json_object(array, &json);
char *json_str = NULL;
neu_json_encode_by_fn(&json, neu_parse_encode_read, &json_str);
clean_write_response_json_object(&json);
return json_str;
}
plugins/mqtt/command_rw.h
+10 -7
View File
@@ -28,13 +28,16 @@ extern "C" {
#include "common.h"
void command_read_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req);
void command_read_response(neu_plugin_t *plugin, neu_taggrp_config_t *config,
neu_data_val_t *array);
void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_write_req_t *write_req);
void command_write_response(neu_plugin_t *plugin, neu_data_val_t *array);
void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req);
char *command_read_once_response(neu_plugin_t *plugin, const char *uuid,
neu_data_val_t *resp_val);
char *command_read_cycle_response(neu_plugin_t * plugin,
neu_data_val_t *resp_val);
void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_write_req_t *write_req);
char *command_write_response(neu_plugin_t *plugin, const char *uuid,
neu_data_val_t *resp_val);
#ifdef __cplusplus
}
plugins/mqtt/message_handle.c
-223
View File
@@ -153,193 +153,6 @@ void message_handle_read(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
UNUSED(req_id);
}
static neu_datatag_t *get_datatag_by_id(neu_plugin_t *plugin, datatag_id_t id)
{
neu_datatag_table_t *table =
neu_system_get_datatags_table(plugin, req_node_id);
if (NULL == table) {
return NULL;
}
return neu_datatag_tbl_get(table, id);
}
void message_handle_read_result(neu_plugin_t * plugin,
neu_taggrp_config_t *grp_config,
neu_variable_t * variable)
{
log_info("Group config name:%s", neu_taggrp_cfg_get_name(grp_config));
if (NULL == variable) {
return;
}
char * result = NULL;
neu_parse_read_res_t res = { 0 };
vector_t *ids = neu_taggrp_cfg_get_datatag_ids(grp_config);
if (NULL == ids) {
return;
}
// res.error = neu_variable_get_error(variable);
res.n_tag = ids->size;
neu_variable_t *v = neu_variable_next(variable);
res.tags = (struct neu_parse_read_res_tag *) calloc(
res.n_tag, sizeof(struct neu_parse_read_res_tag));
datatag_id_t tag_id;
int index = 0;
VECTOR_FOR_EACH(ids, iter)
{
tag_id = *(datatag_id_t *) iterator_get(&iter);
neu_datatag_t *datatag = get_datatag_by_id(plugin, tag_id);
if (NULL == datatag) {
continue;
}
if (NULL == v) {
continue;
}
int type = neu_variable_get_type(v);
res.tags[index].name = strdup(datatag->name);
switch (type) {
case NEU_DATATYPE_BYTE: {
int8_t value;
neu_variable_get_byte(v, &value);
res.tags[index].value.val_bit = value;
break;
}
case NEU_DATATYPE_BOOLEAN: {
bool value;
neu_variable_get_boolean(v, &value);
res.tags[index].value.val_bool = value;
break;
}
case NEU_DATATYPE_WORD: {
int16_t value;
neu_variable_get_word(v, &value);
res.tags[index].value.val_int = value;
break;
}
case NEU_DATATYPE_DWORD: {
int32_t value;
neu_variable_get_dword(v, &value);
res.tags[index].value.val_int = value;
break;
}
case NEU_DATATYPE_STRING:
break;
case NEU_DATATYPE_FLOAT: {
float value;
neu_variable_get_float(v, &value);
res.tags[index].value.val_float = value;
break;
}
default:
break;
}
v = neu_variable_next(v);
index++;
}
neu_parse_mqtt_t mqtt = { .function = NEU_MQTT_OP_READ, .uuid = "123-456" };
int rc = neu_json_encode_with_mqtt(&res, neu_parse_encode_read, &mqtt,
neu_parse_encode_mqtt_param, &result);
if (0 == rc) {
send_mqtt_response(result, strlen(result));
}
int i;
for (i = 0; i < res.n_tag; i++) {
free(res.tags[i].name);
}
free(res.tags);
free(result);
}
static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id,
neu_variable_t *data_var)
{
vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER);
int rc = node_id_exist(&nodes, dest_node_id);
vector_uninit(&nodes);
if (0 != rc) {
return -1;
}
vector_t configs = neu_system_get_group_configs(plugin, dest_node_id);
neu_taggrp_config_t *config;
neu_taggrp_config_t *c;
config = *(neu_taggrp_config_t **) vector_get(&configs, 0);
c = (neu_taggrp_config_t *) neu_taggrp_cfg_ref(config);
uint32_t req_id =
neu_plugin_send_write_cmd(plugin, dest_node_id, c, data_var);
GROUP_CONFIGS_UINIT(configs);
return req_id;
}
void message_handle_write(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
struct neu_parse_write_req *write_req)
{
log_info("WRITE uuid:%s", mqtt->uuid);
req_node_id = write_req->node_id;
neu_variable_t *head = NULL;
head = neu_variable_create();
neu_variable_set_error(head, 0);
for (int i = 0; i < write_req->n_tag; i++) {
enum neu_json_type type = write_req->tags[i].t;
union neu_json_value value = write_req->tags[i].value;
neu_variable_t * v = neu_variable_create();
char key[16] = { 0 };
sprintf(key, "%d", write_req->tags[i].tag_id);
neu_variable_set_key(v, key);
switch (type) {
case NEU_JSON_INT:
neu_variable_set_dword(v, value.val_int);
break;
case NEU_JSON_STR:
neu_variable_set_string(v, value.val_str);
break;
case NEU_JSON_DOUBLE:
neu_variable_set_double(v, value.val_double);
break;
case NEU_JSON_BOOL:
neu_variable_set_boolean(v, value.val_bool);
break;
default:
break;
}
neu_variable_add_item(head, v);
}
write_command(plugin, write_req->node_id, head->next);
neu_variable_destroy(head);
char * json_str = NULL;
neu_parse_error_t error = { 0 };
int rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt,
neu_parse_encode_mqtt_param, &json_str);
if (0 == rc) {
send_mqtt_response(json_str, strlen(json_str));
}
free(json_str);
}
void message_handle_get_tag_list(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_get_tags_req_t *req)
{
@@ -800,42 +613,6 @@ void message_handle_delete_group_config(neu_plugin_t * plugin,
free(json_str);
}
static neu_taggrp_config_t *get_group_config(vector_t * v,
const char *config_name)
{
neu_taggrp_config_t *config;
VECTOR_FOR_EACH(v, iter)
{
config = *(neu_taggrp_config_t **) iterator_get(&iter);
if (NULL == config) {
continue;
}
if (0 == strcmp(config_name, neu_taggrp_cfg_get_name(config))) {
return config;
}
}
return NULL;
}
static int datatag_id_exist(vector_t *v, const uint32_t id, int *index)
{
*index = 0;
datatag_id_t *datatag_id;
VECTOR_FOR_EACH(v, iter)
{
datatag_id = (datatag_id_t *) iterator_get(&iter);
if (id == *datatag_id) {
return 0;
}
(*index)++;
}
*index = -1;
return -1;
}
// static int add_datatag_ids(neu_plugin_t * plugin,
// struct neu_parse_add_tag_ids_req *req)
//{
plugins/mqtt/message_handle.h
-7
View File
@@ -38,13 +38,6 @@ void message_handle_set_paho_client(paho_client_t *paho);
void message_handle_init_tags(neu_plugin_t *plugin);
void message_handle_read(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req);
void message_handle_read_result(neu_plugin_t * plugin,
neu_taggrp_config_t *grp_config,
neu_variable_t * variable);
void message_handle_write(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_write_req_t *write_req);
void message_handle_get_tag_list(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_get_tags_req_t *req);
void message_handle_add_tag(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
plugins/mqtt/plugin_mqtt.c
+51 -67
View File
@@ -80,56 +80,6 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len,
return;
}
// switch (mqtt->function) {
// case NEU_MQTT_OP_GET_GROUP_CONFIG:
// rc = neu_parse_decode_get_group_config(
// json_str, (neu_parse_get_group_config_req_t **) &result);
// break;
// case NEU_MQTT_OP_ADD_GROUP_CONFIG:
// rc = neu_parse_decode_add_group_config(
// json_str, (neu_parse_add_group_config_req_t **) &result);
// break;
// case NEU_MQTT_OP_UPDATE_GROUP_CONFIG:
// rc = neu_parse_decode_update_group_config(
// json_str, (neu_parse_update_group_config_req_t **) &result);
// break;
// case NEU_MQTT_OP_DELETE_GROUP_CONFIG:
// rc = neu_parse_decode_del_group_config(
// json_str, (neu_parse_del_group_config_req_t **) &result);
// break;
// case NEU_MQTT_OP_ADD_DATATAG_IDS_CONFIG:
// rc = -1;
// break;
// case NEU_MQTT_OP_DELETE_DATATAG_IDS_CONFIG:
// rc = -1;
// break;
// case NEU_MQTT_OP_READ:
// rc = neu_parse_decode_read(json_str, (neu_parse_read_req_t **)
// &result); break;
// case NEU_MQTT_OP_WRITE:
// rc = neu_parse_decode_write(json_str,
// (neu_parse_write_req_t **) &result);
// break;
// case NEU_MQTT_OP_GET_TAGS:
// rc = neu_parse_decode_get_tags(json_str,
// (neu_parse_get_tags_req_t **)
// &result);
// break;
// case NEU_MQTT_OP_ADD_TAGS:
// rc = neu_parse_decode_add_tags(json_str,
// (neu_parse_add_tags_req_t **)
// &result);
// break;
// default:
// rc = -1;
// break;
// }
// if (0 != rc) {
// log_error("JSON parsing matt failed");
// return;
// }
switch (mqtt->function) {
case NEU_MQTT_OP_GET_GROUP_CONFIG:
message_handle_get_group_config(
@@ -162,15 +112,22 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len,
case NEU_MQTT_OP_READ: {
rc = neu_parse_decode_read(json_str, (neu_parse_read_req_t **) &result);
if (0 == rc) {
command_read_request(plugin, mqtt, (neu_parse_read_req_t *) result);
command_read_once_request(plugin, mqtt,
(neu_parse_read_req_t *) result);
neu_parse_decode_read_free((neu_parse_read_req_t *) result);
}
break;
}
case NEU_MQTT_OP_WRITE:
message_handle_write(plugin, mqtt, (neu_parse_write_req_t *) result);
neu_parse_decode_write_free(result);
case NEU_MQTT_OP_WRITE: {
rc = neu_parse_decode_write(json_str,
(neu_parse_write_req_t **) &result);
if (0 == rc) {
command_write_request(plugin, mqtt,
(neu_parse_write_req_t *) result);
neu_parse_decode_write_free(result);
}
break;
}
case NEU_MQTT_OP_GET_TAGS:
message_handle_get_tag_list(plugin, mqtt,
(neu_parse_get_tags_req_t *) result);
@@ -296,26 +253,53 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
log_info("send request to plugin: %s", neu_plugin_module.module_name);
switch (req->req_type) {
case NEU_REQRESP_READ_RESP: {
neu_reqresp_read_resp_t *read_resp;
read_resp = (neu_reqresp_read_resp_t *) req->buf;
char *json_str =
command_read_once_response(plugin, "", read_resp->data_val);
if (NULL != json_str) {
client_error error = paho_client_publish(
plugin->paho, "neuronlite/response", 0,
(unsigned char *) json_str, strlen(json_str));
log_debug("Publish error code:%d, json:%s", error, json_str);
free(json_str);
}
break;
}
case NEU_REQRESP_WRITE_RESP: {
neu_reqresp_write_resp_t *write_resp;
write_resp = (neu_reqresp_write_resp_t *) req->buf;
char *json_str =
command_write_response(plugin, "", write_resp->data_val);
if (NULL != json_str) {
client_error error = paho_client_publish(
plugin->paho, "neuronlite/response", 0,
(unsigned char *) json_str, strlen(json_str));
log_debug("Publish error code:%d, json:%s", error, json_str);
free(json_str);
}
break;
}
case NEU_REQRESP_TRANS_DATA: {
neu_reqresp_data_t *neu_data;
assert(req->buf_len == sizeof(neu_reqresp_data_t));
neu_data = (neu_reqresp_data_t *) req->buf;
neu_variable_t *variable = NULL; // neu_data->data_var;
size_t count = neu_variable_count(variable);
log_debug("variable count: %ld", count);
if (0 < count) {
message_handle_read_result(plugin, neu_data->grp_config, variable);
neu_data = (neu_reqresp_data_t *) req->buf;
char *json_str =
command_read_cycle_response(plugin, neu_data->data_val);
if (NULL != json_str) {
client_error error = paho_client_publish(
plugin->paho, "neuronlite/response", 0,
(unsigned char *) json_str, strlen(json_str));
log_debug("Publish error code:%d, json:%s", error, json_str);
free(json_str);
}
log_debug("------------------------------------");
break;
}
case NEU_REQRESP_WRITE_DATA: {
neu_reqresp_data_t *neu_data;
assert(req->buf_len == sizeof(neu_reqresp_data_t));
neu_data = (neu_reqresp_data_t *) req->buf;
neu_variable_t *variable = NULL; // neu_data->data_var;
size_t count = neu_variable_count(variable);
if (0 < count) { }
break;
}
plugins/opcua/plugin_handle.c
+120 -38
View File
@@ -26,6 +26,24 @@
#define UNUSED(x) (void) (x)
#define SET_RESULT_ERRORCODE(resp_val, code) \
{ \
neu_fixed_array_t *array; \
array = neu_fixed_array_new(1, sizeof(neu_int_val_t)); \
if (array == NULL) { \
log_error("Failed to allocate array for response tags"); \
neu_dvalue_free(resp_val); \
return -10; \
} \
neu_int_val_t int_val; \
neu_data_val_t *val_err; \
val_err = neu_dvalue_new(NEU_DTYPE_ERRORCODE); \
neu_dvalue_set_errorcode(val_err, code); \
neu_int_val_init(&int_val, 0, val_err); \
neu_fixed_array_set(array, 0, (void *) &int_val); \
neu_dvalue_init_move_array(resp_val, NEU_DTYPE_INT_VAL, array); \
}
#define NAMESPACE_LEN 16
#define IDENTIFIER_LEN 128
@@ -73,9 +91,9 @@ static int fixed_address(const char *address, int *identifier_type,
return 0;
}
static void generate_read_result_array(vector_t * datas,
neu_fixed_array_t *array)
static int generate_read_result_array(vector_t *datas, neu_fixed_array_t *array)
{
int rc = 0;
opcua_data_t *data;
int total = datas->size;
@@ -88,7 +106,8 @@ static void generate_read_result_array(vector_t * datas,
val = neu_dvalue_new(NEU_DTYPE_ERRORCODE);
neu_dvalue_set_errorcode(val, data->error);
neu_int_val_init(&int_val, data->opcua_node.id, val);
neu_fixed_array_set(array, i, (void *) &int_val);
neu_fixed_array_set(array, i + 1, (void *) &int_val);
rc++;
continue;
}
@@ -158,8 +177,10 @@ static void generate_read_result_array(vector_t * datas,
}
neu_int_val_init(&int_val, data->opcua_node.id, val);
neu_fixed_array_set(array, i, (void *) &int_val);
neu_fixed_array_set(array, i + 1, (void *) &int_val);
}
return rc;
}
static void generate_read_node_vector(opc_handle_context_t *context,
@@ -193,6 +214,7 @@ static void generate_read_node_vector(opc_handle_context_t *context,
}
opcua_data_t read_data;
memset(&read_data, 0, sizeof(opcua_data_t));
read_data.opcua_node.name = datatag->name;
read_data.opcua_node.id = tag_id;
read_data.opcua_node.namespace_index = atoi(namespace_str);
@@ -206,36 +228,35 @@ static void generate_read_node_vector(opc_handle_context_t *context,
}
}
static void generate_write_result_array(vector_t * datas,
neu_data_val_t *resp_val)
static int generate_write_result_array(vector_t * datas,
neu_fixed_array_t *array)
{
int size = datas->size;
neu_fixed_array_t *array;
array = neu_fixed_array_new(size, sizeof(neu_int_val_t));
if (array == NULL) {
log_error("Failed to allocate array for response write data");
neu_dvalue_free(resp_val);
}
neu_int_val_t resp_int_val;
int index = 1;
int error = 0;
VECTOR_FOR_EACH(datas, iter)
{
opcua_data_t * data = (opcua_data_t *) iterator_get(&iter);
opcua_data_t *data = (opcua_data_t *) iterator_get(&iter);
if (0 != data->error) {
error++;
}
neu_data_val_t *val_i32;
val_i32 = neu_dvalue_unit_new();
neu_dvalue_init_int32(val_i32, data->error);
neu_int_val_t resp_int_val;
neu_int_val_init(&resp_int_val, data->opcua_node.id, val_i32);
neu_fixed_array_set(array, index, (void *) &resp_int_val);
index++;
}
neu_dvalue_init_move_array(resp_val, NEU_DTYPE_INT_VAL, array);
return error;
}
static void generate_write_node_vector(opc_handle_context_t *context,
neu_fixed_array_t * array,
vector_t * datas)
{
int size = neu_fixed_array_size(array);
int size = array->length;
neu_int_val_t * int_val;
neu_data_val_t *val;
uint32_t id;
@@ -248,7 +269,10 @@ static void generate_write_node_vector(opc_handle_context_t *context,
neu_datatag_t *datatag = get_datatag_by_id(context->table, id);
if (NULL == datatag) {
continue; // error return ?
opcua_data_t write_data;
write_data.error = -10;
vector_push_back(datas, &write_data);
continue;
}
char namespace_str[NAMESPACE_LEN] = { 0 };
@@ -355,69 +379,127 @@ int plugin_handle_read_once(opc_handle_context_t *context,
neu_taggrp_config_t * config,
neu_data_val_t * resp_val)
{
int code = 0;
if (NULL == config) {
code = -1;
log_error("Group config is NULL");
return -1;
SET_RESULT_ERRORCODE(resp_val, code);
return code;
}
const vector_t *ids = neu_taggrp_cfg_ref_datatag_ids(config);
if (NULL == ids) {
code = -2;
log_error("Datatag vector is NULL");
return -2;
SET_RESULT_ERRORCODE(resp_val, code);
return code;
}
vector_t data;
vector_init(&data, ids->size, sizeof(opcua_data_t));
generate_read_node_vector(context, config, &data);
vector_t datas;
vector_init(&datas, ids->size, sizeof(opcua_data_t));
generate_read_node_vector(context, config, &datas);
int rc = open62541_client_is_connected(context->client);
if (0 != rc) {
code = -3;
log_error("Open62541 client offline");
vector_uninit(&data);
return -3;
SET_RESULT_ERRORCODE(resp_val, code);
vector_uninit(&datas);
return code;
}
rc = open62541_client_read(context->client, &data);
rc = open62541_client_read(context->client, &datas);
if (0 != rc) {
code = -4;
log_error("Open62541 read error:%d", rc);
return -4;
SET_RESULT_ERRORCODE(resp_val, code);
vector_uninit(&datas);
return code;
}
neu_fixed_array_t *array;
array = neu_fixed_array_new(ids->size, sizeof(neu_int_val_t));
if (array == NULL) {
array = neu_fixed_array_new(ids->size + 1, sizeof(neu_int_val_t));
if (NULL == array) {
code = -5;
log_error("Failed to allocate array for response tags");
neu_dvalue_free(resp_val);
return code;
}
generate_read_result_array(&data, array);
vector_uninit(&data);
rc = generate_read_result_array(&datas, array);
neu_int_val_t int_val;
neu_data_val_t *val_err;
val_err = neu_dvalue_new(NEU_DTYPE_ERRORCODE);
neu_dvalue_set_errorcode(val_err, rc);
neu_int_val_init(&int_val, 0, val_err);
neu_fixed_array_set(array, 0, (void *) &int_val);
neu_dvalue_init_move_array(resp_val, NEU_DTYPE_INT_VAL, array);
return 0;
vector_uninit(&datas);
return code;
}
int plugin_handle_write_value(opc_handle_context_t *context,
neu_data_val_t * write_val,
neu_data_val_t * resp_val)
{
int code = 0;
neu_fixed_array_t *array;
neu_dvalue_get_ref_array(write_val, &array);
int size = array->length;
vector_t datas;
vector_init(&datas, 10, sizeof(opcua_data_t));
vector_init(&datas, size, sizeof(opcua_data_t));
generate_write_node_vector(context, array, &datas);
// if (size != datas.size) {
// code = -1;
// log_error("open62541 client offline");
// vector_uninit(&datas);
// SET_RESULT_ERRORCODE(resp_val, code);
// return code;
// }
int rc = open62541_client_is_connected(context->client);
if (0 != rc) {
code = -2;
log_error("open62541 client offline");
return -1;
vector_uninit(&datas);
SET_RESULT_ERRORCODE(resp_val, code);
return code;
}
rc = open62541_client_write(context->client, &datas);
generate_write_result_array(&datas, resp_val);
vector_uninit(&datas);
if (0 != rc) {
code = -3;
log_error("open62541 write error");
vector_uninit(&datas);
SET_RESULT_ERRORCODE(resp_val, code);
return code;
}
neu_fixed_array_t *resp_array;
resp_array = neu_fixed_array_new(size + 1, sizeof(neu_int_val_t));
if (NULL == resp_array) {
code = -4;
log_error("Failed to allocate array for response tags");
vector_uninit(&datas);
neu_dvalue_free(resp_val);
return code;
}
rc = generate_write_result_array(&datas, resp_array);
neu_int_val_t int_val;
neu_data_val_t *val_err;
val_err = neu_dvalue_new(NEU_DTYPE_ERRORCODE);
neu_dvalue_set_errorcode(val_err, rc);
neu_int_val_init(&int_val, 0, val_err);
neu_fixed_array_set(resp_array, 0, (void *) &int_val);
neu_dvalue_init_move_array(resp_val, NEU_DTYPE_INT_VAL, resp_array);
vector_uninit(&datas);
return 0;
}
plugins/opcua/plugin_opcua.c
+36 -9
View File
@@ -149,12 +149,7 @@ static int opcua_plugin_config(neu_plugin_t *plugin, neu_config_t *configs)
static void periodic_response(neu_plugin_t *plugin, neu_taggrp_config_t *config,
neu_data_val_t *resp_val)
{
UNUSED(plugin);
UNUSED(config);
UNUSED(resp_val);
// neu_variable_t *head = neu_variable_create();
// neu_plugin_response_trans_data(plugin, config, head, 0);
// neu_variable_destroy(head);
neu_plugin_response_trans_data(plugin, config, resp_val, 0);
}
static void subscribe_response(neu_plugin_t * plugin,
@@ -187,13 +182,18 @@ static int opcua_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
int rc = 0;
UNUSED(rc);
log_info("send request to plugin: %s", neu_plugin_module.module_name);
const adapter_callbacks_t *adapter_callbacks;
adapter_callbacks = plugin->common.adapter_callbacks;
switch (req->req_type) {
case NEU_REQRESP_READ_DATA: {
neu_reqresp_read_t *read_req = (neu_reqresp_read_t *) req->buf;
neu_data_val_t * resp_val;
neu_response_t resp;
neu_reqresp_data_t data_resp;
resp_val = neu_dvalue_unit_new();
if (resp_val == NULL) {
if (NULL == resp_val) {
log_error("Failed to allocate data value for response tags");
rv = -1;
break;
@@ -201,8 +201,21 @@ static int opcua_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
plugin_handle_read_once(plugin->handle_context, read_req->grp_config,
resp_val);
neu_plugin_response_trans_data(plugin, read_req->grp_config, resp_val,
req->req_id);
data_resp.grp_config = read_req->grp_config;
data_resp.data_val = resp_val;
if (NULL == data_resp.data_val) {
rv = -1;
break;
}
memset(&resp, 0, sizeof(resp));
resp.req_id = req->req_id;
resp.resp_type = NEU_REQRESP_READ_RESP;
resp.recver_id = req->sender_id;
resp.buf_len = sizeof(neu_reqresp_data_t);
resp.buf = &data_resp;
rv = adapter_callbacks->response(plugin->common.adapter, &resp);
break;
}
case NEU_REQRESP_WRITE_DATA: {
@@ -216,7 +229,21 @@ static int opcua_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
break;
}
neu_response_t resp;
neu_reqresp_data_t data_resp;
plugin_handle_write_value(plugin->handle_context, write_val, resp_val);
data_resp.grp_config = write_req->grp_config;
data_resp.data_val = resp_val;
memset(&resp, 0, sizeof(resp));
resp.req_id = req->req_id;
resp.resp_type = NEU_REQRESP_WRITE_RESP;
resp.recver_id = req->sender_id;
resp.buf_len = sizeof(neu_reqresp_data_t);
resp.buf = &data_resp;
rv = adapter_callbacks->response(plugin->common.adapter, &resp);
break;
}
case NEU_REQRESP_SUBSCRIBE_NODE: {