diff --git a/plugins/mqtt/command_rw.c b/plugins/mqtt/command_rw.c index 1a88e2ca..ac6a9e70 100644 --- a/plugins/mqtt/command_rw.c +++ b/plugins/mqtt/command_rw.c @@ -23,26 +23,26 @@ #include "command_rw.h" -void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, - neu_parse_read_req_t *req) +int 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); int rc = common_has_node(plugin, req->node_id); if (0 != rc) { log_debug("The requested node does not exist"); - return; + return -1; } neu_taggrp_config_t *config = neu_system_find_group_config( plugin, req->node_id, req->group_config_name); if (NULL == config) { log_debug("The requested config does not exist"); - return; + return -2; } uint32_t req_id = neu_plugin_get_event_id(plugin); neu_plugin_send_read_cmd(plugin, req_id, req->node_id, config); - UNUSED(req_id); + return req_id; } static int wrap_read_response_json_object(neu_fixed_array_t * array, @@ -208,7 +208,8 @@ char *command_read_once_response(neu_plugin_t * plugin, } char *json_str = NULL; - rc = neu_json_encode_by_fn(&json, neu_parse_encode_read, &json_str); + rc = neu_json_encode_with_mqtt(&json, neu_parse_encode_read, parse_header, + neu_parse_encode_mqtt_param, &json_str); clean_read_response_json_object(&json); if (0 != rc) { log_info("Json string parse error:%d", rc); @@ -229,54 +230,18 @@ char *command_read_cycle_response(neu_plugin_t * plugin, return command_read_once_response(plugin, &parse_header, resp_val); } -static bool match_name_grp_config(const void *key, const void *item) -{ - if (NULL == key || NULL == item) { - return false; - } - const char * name = (const char *) key; - neu_taggrp_config_t *config = *(neu_taggrp_config_t **) item; - if (0 == strcmp(name, neu_taggrp_cfg_get_name(config))) { - return true; - } - - return false; -} - -static neu_taggrp_config_t * -get_group_config_by_name(neu_plugin_t *plugin, uint32_t dest_node_id, - const char *group_config_name) -{ - vector_t configs = neu_system_get_group_configs(plugin, dest_node_id); - neu_taggrp_config_t *config; - neu_taggrp_config_t *c; - // config1 = *(neu_taggrp_config_t **) vector_get(&configs, 0); - config = *(neu_taggrp_config_t **) vector_find_item( - &configs, group_config_name, match_name_grp_config); - if (NULL == config) { - GROUP_CONFIGS_UNINIT(configs); - return NULL; - } - c = (neu_taggrp_config_t *) neu_taggrp_cfg_ref(config); - GROUP_CONFIGS_UNINIT(configs); - - return c; -} - static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id, const char * group_config_name, 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); - + int rc = common_has_node(plugin, dest_node_id); if (0 != rc) { return -1; } neu_taggrp_config_t *config; - config = get_group_config_by_name(plugin, dest_node_id, group_config_name); + config = + neu_system_find_group_config(plugin, dest_node_id, group_config_name); if (NULL == config) { return -2; } @@ -287,8 +252,8 @@ static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id, return req_id; } -void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, - neu_parse_write_req_t *write_req) +int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_write_req_t *write_req) { log_info("WRITE uuid:%s, group config name:%s", mqtt->uuid, write_req->group_config_name); @@ -298,14 +263,14 @@ void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, write_val = neu_dvalue_unit_new(); if (write_val == NULL) { log_error("Failed to allocate data value for write data"); - return; + return -1; } 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; + return -2; } neu_int_val_t int_val; @@ -350,9 +315,10 @@ void command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, } neu_dvalue_init_move_array(write_val, NEU_DTYPE_INT_VAL, array); - write_command(plugin, write_req->node_id, write_req->group_config_name, - write_val); + int req_id = write_command(plugin, write_req->node_id, + write_req->group_config_name, write_val); neu_dvalue_free(write_val); + return req_id; } static int wrap_write_response_json_object(neu_fixed_array_t * array, diff --git a/plugins/mqtt/command_rw.h b/plugins/mqtt/command_rw.h index f94b9bd3..b2d9bd09 100644 --- a/plugins/mqtt/command_rw.h +++ b/plugins/mqtt/command_rw.h @@ -28,14 +28,14 @@ extern "C" { #include "common.h" -void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, +int 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, neu_parse_mqtt_t *parse_header, 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, +int 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); diff --git a/plugins/mqtt/plugin_mqtt.c b/plugins/mqtt/plugin_mqtt.c index b3f4e865..1525f624 100644 --- a/plugins/mqtt/plugin_mqtt.c +++ b/plugins/mqtt/plugin_mqtt.c @@ -47,6 +47,12 @@ const neu_plugin_module_t neu_plugin_module; +struct context { + neu_list_node node; + int req_id; + neu_parse_mqtt_t parse_header; +}; + struct neu_plugin { neu_plugin_common_t common; option_t option; @@ -55,6 +61,64 @@ struct neu_plugin { neu_parse_mqtt_t * parse_header; }; +struct context *context_create() +{ + struct context *ctx = NULL; + ctx = malloc(sizeof(struct context)); + if (NULL != ctx) { + memset(ctx, 0, sizeof(struct context)); + } + return ctx; +} + +void context_list_add(neu_list *list, struct context *ctx, int req_id, + neu_parse_mqtt_t *parse_header) +{ + ctx->req_id = req_id; + ctx->parse_header.function = parse_header->function; + ctx->parse_header.uuid = strdup(parse_header->uuid); + + NEU_LIST_NODE_INIT(&ctx->node); + neu_list_append(list, ctx); +} + +struct context *context_list_find(neu_list *list, const int id) +{ + struct context *item; + NEU_LIST_FOREACH(list, item) + { + if (id == item->req_id) { + return item; + } + } + return NULL; +} + +void context_destroy(struct context *ctx) +{ + if (NULL != ctx) { + if (NULL != ctx->parse_header.uuid) { + free(ctx->parse_header.uuid); + } + free(ctx); + } +} + +void context_list_remove(neu_list *list, struct context *ctx) +{ + UNUSED(list); + UNUSED(ctx); +} + +void context_list_destroy(neu_list *list) +{ + while (!neu_list_empty(list)) { + struct context *ctx = neu_list_first(list); + neu_list_remove(list, ctx); + context_destroy(ctx); + } +} + static int plugin_subscribe(neu_plugin_t *plugin, const char *topic, const int qos, subscribe_handle handle) { @@ -137,7 +201,8 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, neu_parse_read_req_t *req = NULL; rc = neu_parse_decode_read(json_str, &req); if (0 == rc) { - command_read_once_request(plugin, mqtt, req); + int req_id = command_read_once_request(plugin, mqtt, req); + UNUSED(req_id); neu_parse_decode_read_free(req); } break; @@ -146,7 +211,8 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, neu_parse_write_req_t *req = NULL; rc = neu_parse_decode_write(json_str, &req); if (0 == rc) { - command_write_request(plugin, mqtt, req); + int req_id = command_write_request(plugin, mqtt, req); + UNUSED(req_id); neu_parse_decode_write_free(req); } break; @@ -228,7 +294,17 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, } MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); - free(json_str); + + // if (NULL != mqtt) { + // if (NULL != mqtt->uuid) { + // free(mqtt->uuid); + // } + // free(mqtt); + // } + + if (NULL != json_str) { + free(json_str); + } UNUSED(topic_len); } @@ -295,6 +371,8 @@ static int mqtt_plugin_init(neu_plugin_t *plugin) plugin->parse_header = NULL; + NEU_LIST_INIT(&plugin->context_list, struct context, node); + // Paho mqtt client setup client_error error = paho_client_open(&plugin->option, plugin, &plugin->paho); diff --git a/plugins/opcua/plugin_handle.c b/plugins/opcua/plugin_handle.c index 5156d5b9..181ca4c6 100644 --- a/plugins/opcua/plugin_handle.c +++ b/plugins/opcua/plugin_handle.c @@ -548,6 +548,7 @@ static void periodic_read(nng_aio *aio, void *arg, int code) stop_periodic_read(tuple); + neu_taggrp_cfg_anchor(new_config); if (new_config != NULL) { neu_taggrp_cfg_free(tuple->config); tuple->config = new_config; @@ -636,7 +637,6 @@ static opc_subscribe_tuple_t *find_subscribe(opc_handle_context_t *context, static void stop_periodic_read(opc_subscribe_tuple_t *tuple) { nng_aio_cancel(tuple->aio); - nng_aio_free(tuple->aio); } static void remove_subscribe(opc_subscribe_tuple_t *tuple)