diff --git a/plugins/mqtt/command_rw.c b/plugins/mqtt/command_rw.c index c6cc993c..54bc1fe1 100644 --- a/plugins/mqtt/command_rw.c +++ b/plugins/mqtt/command_rw.c @@ -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; } \ No newline at end of file diff --git a/plugins/mqtt/command_rw.h b/plugins/mqtt/command_rw.h index 3491459e..73b3e7e4 100644 --- a/plugins/mqtt/command_rw.h +++ b/plugins/mqtt/command_rw.h @@ -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 } diff --git a/plugins/mqtt/message_handle.c b/plugins/mqtt/message_handle.c index 84d6b95a..1be85239 100644 --- a/plugins/mqtt/message_handle.c +++ b/plugins/mqtt/message_handle.c @@ -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) //{ diff --git a/plugins/mqtt/message_handle.h b/plugins/mqtt/message_handle.h index ed949609..2c175430 100644 --- a/plugins/mqtt/message_handle.h +++ b/plugins/mqtt/message_handle.h @@ -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, diff --git a/plugins/mqtt/plugin_mqtt.c b/plugins/mqtt/plugin_mqtt.c index bf58ec9f..d850be93 100644 --- a/plugins/mqtt/plugin_mqtt.c +++ b/plugins/mqtt/plugin_mqtt.c @@ -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; } diff --git a/plugins/opcua/plugin_handle.c b/plugins/opcua/plugin_handle.c index 9b364fac..005b8713 100644 --- a/plugins/opcua/plugin_handle.c +++ b/plugins/opcua/plugin_handle.c @@ -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; } diff --git a/plugins/opcua/plugin_opcua.c b/plugins/opcua/plugin_opcua.c index c603a294..b086ba03 100644 --- a/plugins/opcua/plugin_opcua.c +++ b/plugins/opcua/plugin_opcua.c @@ -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: {