From 83884fd7c64fce31304a7bfd135eba313e4f005e Mon Sep 17 00:00:00 2001 From: gc87 Date: Tue, 12 Oct 2021 15:56:20 +0800 Subject: [PATCH] Refactor the mqtt driver management interface 1. Refactor the group config operation interface 2. Refactor the datatag table operation interface --- plugins/mqtt/CMakeLists.txt | 11 +- plugins/mqtt/command_datatag.c | 402 ++++++++++++++++++++++++++++ plugins/mqtt/command_datatag.h | 43 +++ plugins/mqtt/command_group_config.c | 258 ++++++++++++++++++ plugins/mqtt/command_group_config.h | 43 +++ plugins/mqtt/command_rw.c | 6 +- plugins/mqtt/common.c | 61 +++++ plugins/mqtt/common.h | 17 +- plugins/mqtt/plugin_mqtt.c | 163 +++++++---- 9 files changed, 942 insertions(+), 62 deletions(-) create mode 100644 plugins/mqtt/command_datatag.c create mode 100644 plugins/mqtt/command_datatag.h create mode 100644 plugins/mqtt/command_group_config.c create mode 100644 plugins/mqtt/command_group_config.h diff --git a/plugins/mqtt/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index 4ff5fb49..48404552 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -3,7 +3,16 @@ find_package(eclipse-paho-mqtt-c 1.3.9 REQUIRED) file(GLOB PARSER_SRC "${CMAKE_SOURCE_DIR}/src/parser/*.c") file(GLOB CONFIG_SRC "${CMAKE_SOURCE_DIR}/src/config.c") set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}") -set(src ${PARSER_SRC} ${CONFIG_SRC} plugin_mqtt.c paho_client.c message_handle.c common.c command_rw.c) +set(src + ${PARSER_SRC} + ${CONFIG_SRC} + plugin_mqtt.c + paho_client.c + message_handle.c + common.c + command_rw.c + command_group_config.c + command_datatag.c) add_library(plugin-mqtt SHARED ${src}) target_include_directories(plugin-mqtt PRIVATE diff --git a/plugins/mqtt/command_datatag.c b/plugins/mqtt/command_datatag.c new file mode 100644 index 00000000..220cf987 --- /dev/null +++ b/plugins/mqtt/command_datatag.c @@ -0,0 +1,402 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#include + +#include "command_datatag.h" + +typedef struct { + char * name; + int type; + char * address; + int attribute; + char * group_config_name; + neu_datatag_id_t id; +} tag_t; + +struct visit_context { + neu_datatag_table_t *table; + vector_t * tags; +}; + +void visit_func(neu_taggrp_config_t *config, void *context) +{ + struct visit_context *ctx = (struct visit_context *) context; + if (NULL == config) { + return; + } + + char *group_config_name = strdup((char *) neu_taggrp_cfg_get_name(config)); + vector_t *tag_ids = neu_taggrp_cfg_get_datatag_ids(config); + if (NULL == tag_ids) { + return; + } + + neu_datatag_id_t id; + VECTOR_FOR_EACH(tag_ids, id_iter) + { + id = *(neu_datatag_id_t *) iterator_get(&id_iter); + + tag_t *tag = malloc(sizeof(tag_t)); + if (NULL == tag) { + continue; + } + memset(tag, 0, sizeof(tag_t)); + + neu_datatag_t *datatag = NULL; + datatag = neu_datatag_tbl_get(ctx->table, id); + if (NULL == datatag) { + continue; + } + + tag->name = strdup(datatag->name); + tag->type = datatag->type; + tag->address = strdup(datatag->addr_str); + tag->attribute = datatag->attribute; + tag->group_config_name = strdup(group_config_name); + tag->id = id; + + log_debug("id:%u, config name:%s, " + "attr:%d, address:%s, type:%d, name:%s", + tag->id, tag->group_config_name, tag->attribute, tag->address, + tag->type, tag->name); + + vector_push_back(ctx->tags, tag); + } +} + +static int get_all_tag(neu_plugin_t *plugin, const uint32_t node_id, + vector_t *tags) +{ + int rc = common_has_node(plugin, node_id); + if (0 != rc) { + return -1; + } + + vector_t configs = neu_system_get_group_configs(plugin, node_id); + if (0 == configs.size) { + GROUP_CONFIGS_UNINIT(configs); + return -2; + } + + neu_datatag_table_t *table = common_get_datatags_table(plugin, node_id); + + if (NULL == table) { + GROUP_CONFIGS_UNINIT(configs); + return -3; + } + + struct visit_context ctx = { .table = table, .tags = tags }; + common_group_configs_freach(plugin, node_id, &ctx, visit_func); + + return 0; +} + +static int wrap_get_tags_response_json_object(vector_t * tags, + neu_parse_get_tags_res_t *json) +{ + json->n_tag = tags->size; + json->tags = malloc(tags->size * sizeof(neu_parse_get_tags_res_tag_t)); + memset(json->tags, 0, tags->size * sizeof(neu_parse_get_tags_res_tag_t)); + + tag_t *tag = NULL; + int size = tags->size; + for (int i = 0; i < size; i++) { + tag = vector_get(tags, i); + if (NULL == tag) { + continue; + } + + json->tags[i].name = tag->name; + json->tags[i].type = tag->type; + json->tags[i].address = tag->address; + json->tags[i].attribute = tag->attribute; + json->tags[i].group_config_name = tag->group_config_name; + json->tags[i].id = tag->id; + } + + return 0; +} + +static void free_get_tags_response_object(vector_t * tags, + neu_parse_get_tags_res_t *json) +{ + tag_t *tag = NULL; + int size = tags->size; + for (int i = 0; i < size; i++) { + tag = vector_get(tags, i); + if (NULL == tag) { + continue; + } + + if (NULL != tag->name) { + free(tag->name); + } + if (NULL != tag->address) { + free(tag->address); + } + if (NULL != tag->group_config_name) { + free(tag->group_config_name); + } + } + + if (NULL != json->tags) { + free(json->tags); + } +} + +char *command_get_tags(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_get_tags_req_t *req) +{ + log_info("Get tag list uuid:%s, node id:%d", mqtt->uuid, req->node_id); + char * json_str = NULL; + neu_parse_get_tags_res_t res = { 0 }; + neu_parse_error_t error = { 0 }; + + uint32_t node_id = req->node_id; + vector_t tags; + vector_init(&tags, 0, sizeof(tag_t)); + int rc = get_all_tag(plugin, node_id, &tags); + + if (0 != rc) { + error.error = rc; + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + return NULL; + } + + rc = wrap_get_tags_response_json_object(&tags, &res); + rc = neu_json_encode_with_mqtt(&res, neu_parse_encode_get_tags, mqtt, + neu_parse_encode_mqtt_param, &json_str); + + free_get_tags_response_object(&tags, &res); + vector_uninit(&tags); + + if (0 == rc) { + return json_str; + } + + return NULL; +} + +char *command_add_tags(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_add_tags_req_t *req) +{ + log_info("Add tags uuid:%s, node id:%d, group config name:%s", mqtt->uuid, + req->node_id, req->group_config_name); + + char * json_str = NULL; + uint32_t node_id = req->node_id; + neu_parse_error_t error = { 0 }; + + int rc = common_has_node(plugin, node_id); + if (0 != rc) { + error.error = -1; + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + return NULL; + } + + neu_taggrp_config_t *config; + config = common_get_group_config(plugin, node_id, req->group_config_name); + if (NULL == config) { + error.error = -2; + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + return NULL; + } + + neu_datatag_table_t *table = common_get_datatags_table(plugin, node_id); + if (NULL == table) { + error.error = -3; + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + return NULL; + } + + for (int i = 0; i < req->n_tag; i++) { + neu_datatag_t *datatag = malloc(sizeof(neu_datatag_t)); + datatag->name = strdup(req->tags[i].name); + datatag->attribute = req->tags[i].attribute; + datatag->type = req->tags[i].type; + datatag->addr_str = strdup(req->tags[i].address); + datatag_id_t id = neu_datatag_tbl_add(table, datatag); + + log_info("Add datatag to the datatag table, id:%ld", id); + if (0 != id) { + vector_t *ids = neu_taggrp_cfg_get_datatag_ids(config); + int * tag_id = malloc(sizeof(uint32_t)); + *tag_id = id; + rc = vector_push_back(ids, tag_id); + log_info("Add datatag id to the config:%s, id:%ld, error code:%d", + req->group_config_name, id, rc); + } + } + + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + + return NULL; +} + +char *command_update_tags(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_update_tags_req_t *req) +{ + log_info("Update tags uuid:%s, node id:%d", mqtt->uuid, req->node_id); + + char * json_str = NULL; + uint32_t node_id = req->node_id; + neu_parse_error_t error = { 0 }; + + int rc = common_has_node(plugin, node_id); + if (0 != rc) { + error.error = -1; + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + return NULL; + } + + neu_datatag_table_t *table = common_get_datatags_table(plugin, node_id); + if (NULL == table) { + error.error = -3; + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + return NULL; + } + + for (int i = 0; i < req->n_tag; i++) { + neu_datatag_t *datatag = + neu_datatag_tbl_get(table, req->tags[i].tag_id); + + if (NULL == datatag) { + continue; + } + + free(datatag->name); + free(datatag->addr_str); + + datatag->id = req->tags[i].tag_id; + datatag->name = strdup(req->tags[i].name); + datatag->attribute = req->tags[i].attribute; + datatag->type = req->tags[i].type; + datatag->addr_str = strdup(req->tags[i].address); + rc = neu_datatag_tbl_update(table, req->tags[i].tag_id, datatag); + + log_info("Update datatag to the datatag table, id:%ld", datatag->id); + } + + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + + if (0 == rc) { + return json_str; + } + + return NULL; +} + +char *command_delete_tags(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_del_tags_req_t *req) +{ + log_info("Delete tags uuid:%s, node id:%d, group config:%s", mqtt->uuid, + req->node_id, req->group_config_name); + + char * json_str = NULL; + uint32_t node_id = req->node_id; + neu_parse_error_t error = { 0 }; + + int rc = common_has_node(plugin, node_id); + if (0 != rc) { + error.error = -1; + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + return NULL; + } + + neu_taggrp_config_t *config = + neu_system_find_group_config(plugin, node_id, req->group_config_name); + if (NULL == config) { + error.error = -2; + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + return NULL; + } + + neu_datatag_table_t *table = + neu_system_get_datatags_table(plugin, req->node_id); + if (NULL == table) { + error.error = -3; + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + return NULL; + } + + for (int i = 0; i < req->n_tag_id; i++) { + rc = neu_datatag_tbl_remove(table, req->tag_ids[i]); + if (0 == rc) { + vector_t *ids = neu_taggrp_cfg_get_datatag_ids(config); + VECTOR_FOR_EACH(ids, iter) + { + neu_datatag_id_t *id = (neu_datatag_id_t *) iterator_get(&iter); + if (*id == req->tag_ids[i]) { + iterator_erase(ids, &iter); + break; + } + } + } + } + + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + if (0 == rc) { + return json_str; + } + + return NULL; +} \ No newline at end of file diff --git a/plugins/mqtt/command_datatag.h b/plugins/mqtt/command_datatag.h new file mode 100644 index 00000000..98c88df9 --- /dev/null +++ b/plugins/mqtt/command_datatag.h @@ -0,0 +1,43 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#ifndef NEURON_PLUGIN_MQTT_COMMAND_DATATAG +#define NEURON_PLUGIN_MQTT_COMMAND_DATATAG + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#include "common.h" + +char *command_get_tags(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_get_tags_req_t *req); +char *command_add_tags(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_add_tags_req_t *req); +char *command_update_tags(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_update_tags_req_t *req); +char *command_delete_tags(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_del_tags_req_t *req); + +#ifdef __cplusplus +} +#endif +#endif \ No newline at end of file diff --git a/plugins/mqtt/command_group_config.c b/plugins/mqtt/command_group_config.c new file mode 100644 index 00000000..5d7172f9 --- /dev/null +++ b/plugins/mqtt/command_group_config.c @@ -0,0 +1,258 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#include + +#include "command_group_config.h" + +static void get_all_group_configs(neu_plugin_t *plugin, const uint32_t node_id, + neu_parse_get_group_config_res_t *res) +{ + vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER); + int rc = common_node_id_exist(&nodes, node_id); + vector_uninit(&nodes); + if (0 != rc) { + return; + } + + vector_t configs = neu_system_get_group_configs(plugin, node_id); + if (0 == configs.size) { + GROUP_CONFIGS_UNINIT(configs); + return; + } + + res->n_config = configs.size; + res->rows = (neu_parse_get_group_config_res_row_t *) malloc( + sizeof(neu_parse_get_group_config_res_row_t) * configs.size); + memset(res->rows, 0, + sizeof(neu_parse_get_group_config_res_row_t) * configs.size); + + int count = 0; + neu_taggrp_config_t *config; + VECTOR_FOR_EACH(&configs, iter) + { + config = *(neu_taggrp_config_t **) iterator_get(&iter); + if (NULL == config) { + continue; + } + + res->rows[count].name = + strdup((char *) neu_taggrp_cfg_get_name(config)); + res->rows[count].interval = neu_taggrp_cfg_get_interval(config); + + vector_t *pipes = neu_taggrp_cfg_get_subpipes(config); + if (NULL != pipes) { + res->rows[count].pipe_count = pipes->size; + } + + vector_t *ids = neu_taggrp_cfg_get_datatag_ids(config); + if (NULL != ids) { + res->rows[count].tag_count = ids->size; + } + + count++; + } + GROUP_CONFIGS_UNINIT(configs); +} + +static void +free_parse_get_group_config_res(neu_parse_get_group_config_res_t *res) +{ + if (NULL == res) { + return; + } + + if (NULL == res->rows) { + return; + } + + int i; + for (i = 0; i < res->n_config; i++) { + if (NULL == res->rows[i].name) { + free(res->rows[i].name); + } + } + free(res->rows); +} + +char *command_get_group_configs(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_get_group_config_req_t *req) +{ + log_debug("uuid:%s, node id:%d", mqtt->uuid, req->node_id); + + char * json_str = NULL; + neu_parse_get_group_config_res_t res = { 0 }; + get_all_group_configs(plugin, req->node_id, &res); + neu_json_encode_with_mqtt(&res, neu_parse_encode_get_group_config, mqtt, + neu_parse_encode_mqtt_param, &json_str); + free_parse_get_group_config_res(&res); + + return json_str; +} + +static int add_group_config(neu_plugin_t *plugin, const char *config_name, + const uint32_t node_id, const int read_interval) +{ + if (0 == node_id) { + return -1; + } + + vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER); + int rc = common_node_id_exist(&nodes, node_id); + vector_uninit(&nodes); + if (0 != rc) { + return -2; + } + + vector_t configs = neu_system_get_group_configs(plugin, node_id); + rc = common_config_exist(&configs, config_name); + GROUP_CONFIGS_UNINIT(configs); + + if (0 == rc) { + return -4; + } + + neu_taggrp_config_t *grp_config = neu_taggrp_cfg_new((char *) config_name); + if (NULL == grp_config) { + return -5; + } + neu_taggrp_cfg_set_interval(grp_config, read_interval); + intptr_t error = neu_system_add_group_config(plugin, node_id, grp_config); + if (0 != error) { + return -6; + } + return 0; +} + +char *command_add_group_config(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_add_group_config_req_t *req) +{ + log_debug("uuid:%s, config:%s, node id:%d, " + "read interval:%d", + mqtt->uuid, req->name, req->node_id, req->interval); + + int rc = add_group_config(plugin, req->name, req->node_id, req->interval); + + char * json_str = NULL; + neu_parse_error_t error = { .error = rc }; + + neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + return json_str; +} + +static int update_group_config(neu_plugin_t *plugin, const char *config_name, + const uint32_t node_id, const int read_interval) +{ + if (0 == node_id) { + return -1; + } + + vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER); + int rc = common_node_id_exist(&nodes, node_id); + vector_uninit(&nodes); + if (0 != rc) { + return -2; + } + + vector_t configs = neu_system_get_group_configs(plugin, node_id); + rc = common_config_exist(&configs, config_name); + GROUP_CONFIGS_UNINIT(configs); + + if (0 != rc) { + return -4; + } + + neu_taggrp_config_t *grp_config = neu_taggrp_cfg_new((char *) config_name); + if (NULL == grp_config) { + return -5; + } + neu_taggrp_cfg_set_interval(grp_config, read_interval); + + intptr_t error = + neu_system_update_group_config(plugin, node_id, grp_config); + if (0 != error) { + return -6; + } + return 0; +} + +char *command_update_group_config(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_update_group_config_req_t *req) +{ + log_debug("uuid:%s, config:%s, src node id:%d," + "read interval:%d", + mqtt->uuid, req->name, req->node_id, req->interval); + + int rc = + update_group_config(plugin, req->name, req->node_id, req->interval); + + char * json_str = NULL; + neu_parse_error_t error = { .error = rc }; + + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + return json_str; +} + +static int delete_group_config(neu_plugin_t *plugin, const char *config_name, + const uint32_t node_id) +{ + if (0 == node_id) { + return -1; + } + + vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER); + int rc = common_node_id_exist(&nodes, node_id); + vector_uninit(&nodes); + if (0 != rc) { + return -2; + } + + vector_t configs = neu_system_get_group_configs(plugin, node_id); + rc = common_config_exist(&configs, config_name); + GROUP_CONFIGS_UNINIT(configs); + + if (0 != rc) { + return -4; + } + + intptr_t error = + neu_system_del_group_config(plugin, node_id, (char *) config_name); + if (0 != error) { + return -5; + } + return 0; +} + +char *command_delete_group_config(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_del_group_config_req_t *req) +{ + log_debug("uuid:%s, config:%s, src node id:%d", mqtt->uuid, req->name, + req->node_id); + + int rc = delete_group_config(plugin, req->name, req->node_id); + + char * json_str = NULL; + neu_parse_error_t error = { .error = rc }; + + rc = neu_json_encode_with_mqtt(&error, neu_parse_encode_error, mqtt, + neu_parse_encode_mqtt_param, &json_str); + return json_str; +} diff --git a/plugins/mqtt/command_group_config.h b/plugins/mqtt/command_group_config.h new file mode 100644 index 00000000..41aac602 --- /dev/null +++ b/plugins/mqtt/command_group_config.h @@ -0,0 +1,43 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#ifndef NEURON_PLUGIN_MQTT_COMMAND_GROUP_CONFIG +#define NEURON_PLUGIN_MQTT_COMMAND_GROUP_CONFIG + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#include "common.h" + +char *command_get_group_configs(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_get_group_config_req_t *req); +char *command_add_group_config(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_add_group_config_req_t *req); +char *command_update_group_config(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_update_group_config_req_t *req); +char *command_delete_group_config(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, + neu_parse_del_group_config_req_t *req); + +#ifdef __cplusplus +} +#endif +#endif \ No newline at end of file diff --git a/plugins/mqtt/command_rw.c b/plugins/mqtt/command_rw.c index 448afc1a..d4576678 100644 --- a/plugins/mqtt/command_rw.c +++ b/plugins/mqtt/command_rw.c @@ -42,7 +42,7 @@ void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, c = (neu_taggrp_config_t *) neu_taggrp_cfg_ref(config); uint32_t req_id = neu_plugin_send_read_cmd(plugin, req->node_id, c); - GROUP_CONFIGS_UINIT(configs); + GROUP_CONFIGS_UNINIT(configs); UNUSED(req_id); } @@ -258,11 +258,11 @@ get_group_config_by_name(neu_plugin_t *plugin, uint32_t dest_node_id, config = *(neu_taggrp_config_t **) vector_find_item( &configs, group_config_name, match_name_grp_config); if (NULL == config) { - GROUP_CONFIGS_UINIT(configs); + GROUP_CONFIGS_UNINIT(configs); return NULL; } c = (neu_taggrp_config_t *) neu_taggrp_cfg_ref(config); - GROUP_CONFIGS_UINIT(configs); + GROUP_CONFIGS_UNINIT(configs); return c; } diff --git a/plugins/mqtt/common.c b/plugins/mqtt/common.c index ab29b4ef..d172611e 100644 --- a/plugins/mqtt/common.c +++ b/plugins/mqtt/common.c @@ -54,3 +54,64 @@ int common_config_exist(vector_t *v, const char *config_name) } return -1; } + +int common_has_node(neu_plugin_t *plugin, const uint32_t node_id) +{ + vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER); + int rc = common_node_id_exist(&nodes, node_id); + vector_uninit(&nodes); + return rc; +} + +int common_has_group_config(neu_plugin_t *plugin, const uint32_t node_id, + const char *group_config_name) +{ + vector_t configs = neu_system_get_group_configs(plugin, node_id); + int rc = common_config_exist(&configs, group_config_name); + GROUP_CONFIGS_UNINIT(configs); + return rc; +} + +neu_taggrp_config_t *common_get_group_config(neu_plugin_t * plugin, + const uint32_t node_id, + const char * group_config_name) +{ + vector_t configs = neu_system_get_group_configs(plugin, node_id); + neu_taggrp_config_t *config = NULL; + neu_taggrp_config_t *ret_config = NULL; + VECTOR_FOR_EACH(&configs, iter) + { + config = *(neu_taggrp_config_t **) iterator_get(&iter); + if (NULL == config) { + continue; + } + + if (0 == strcmp(group_config_name, neu_taggrp_cfg_get_name(config))) { + ret_config = config; + } + } + GROUP_CONFIGS_UNINIT(configs); + + return ret_config; +} + +void common_group_configs_freach(neu_plugin_t *plugin, const uint32_t node_id, + void *context, + void visit_func(neu_taggrp_config_t *, void *)) +{ + vector_t configs = neu_system_get_group_configs(plugin, node_id); + neu_taggrp_config_t *config; + VECTOR_FOR_EACH(&configs, iter) + { + config = *(neu_taggrp_config_t **) iterator_get(&iter); + visit_func(config, context); + } + GROUP_CONFIGS_UNINIT(configs); +} + +neu_datatag_table_t *common_get_datatags_table(neu_plugin_t * plugin, + const uint32_t node_id) +{ + neu_datatag_table_t *table = neu_system_get_datatags_table(plugin, node_id); + return table; +} \ No newline at end of file diff --git a/plugins/mqtt/common.h b/plugins/mqtt/common.h index 87f25762..f18c39a9 100644 --- a/plugins/mqtt/common.h +++ b/plugins/mqtt/common.h @@ -33,7 +33,7 @@ extern "C" { #define UNUSED(x) (void) (x) -#define GROUP_CONFIGS_UINIT(configs) \ +#define GROUP_CONFIGS_UNINIT(configs) \ { \ VECTOR_FOR_EACH(&configs, iter) \ { \ @@ -47,6 +47,21 @@ extern "C" { int common_node_id_exist(vector_t *v, const uint32_t node_id); int common_config_exist(vector_t *v, const char *config_name); +int common_has_node(neu_plugin_t *plugin, const uint32_t node_id); + +int common_has_group_config(neu_plugin_t *plugin, const uint32_t node_id, + const char *group_config_name); +neu_taggrp_config_t *common_get_group_config(neu_plugin_t * plugin, + const uint32_t node_id, + const char * group_config_name); +void common_group_configs_freach(neu_plugin_t *plugin, const uint32_t node_id, + void *context, + void visit_func(neu_taggrp_config_t *, + void *)); + +neu_datatag_table_t *common_get_datatags_table(neu_plugin_t * plugin, + const uint32_t node_id); + #ifdef __cplusplus } #endif diff --git a/plugins/mqtt/plugin_mqtt.c b/plugins/mqtt/plugin_mqtt.c index d850be93..704e1734 100644 --- a/plugins/mqtt/plugin_mqtt.c +++ b/plugins/mqtt/plugin_mqtt.c @@ -24,6 +24,8 @@ #include #include +#include "command_datatag.h" +#include "command_group_config.h" #include "command_rw.h" #include "message_handle.h" @@ -32,6 +34,17 @@ #define UNUSED(x) (void) (x) +#define MQTT_SEND(client, topic, qos, json_str) \ + { \ + if (NULL != json_str) { \ + client_error error = paho_client_publish( \ + client, topic, qos, (unsigned char *) json_str, \ + strlen(json_str)); \ + log_debug("Publish error code:%d, json:%s", error, json_str); \ + free(json_str); \ + } \ + } + const neu_plugin_module_t neu_plugin_module; struct neu_plugin { @@ -81,34 +94,54 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, } switch (mqtt->function) { - case NEU_MQTT_OP_GET_GROUP_CONFIG: - message_handle_get_group_config( - plugin, mqtt, (neu_parse_get_group_config_req_t *) result); - neu_parse_decode_get_group_config_free(result); + case NEU_MQTT_OP_GET_GROUP_CONFIG: { + rc = neu_parse_decode_get_group_config( + json_str, (neu_parse_get_group_config_req_t **) &result); + if (0 == rc) { + char *ret_str = NULL; + ret_str = command_get_group_configs( + plugin, mqtt, (neu_parse_get_group_config_req_t *) result); + neu_parse_decode_get_group_config_free(result); + MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + } break; - case NEU_MQTT_OP_ADD_GROUP_CONFIG: - message_handle_add_group_config( - plugin, mqtt, (neu_parse_add_group_config_req_t *) result); - neu_parse_decode_add_group_config_free(result); + } + case NEU_MQTT_OP_ADD_GROUP_CONFIG: { + rc = neu_parse_decode_add_group_config( + json_str, (neu_parse_add_group_config_req_t **) &result); + if (0 == rc) { + char *ret_str = NULL; + ret_str = command_add_group_config( + plugin, mqtt, (neu_parse_add_group_config_req_t *) result); + neu_parse_decode_add_group_config_free(result); + MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + } break; - case NEU_MQTT_OP_UPDATE_GROUP_CONFIG: - message_handle_update_group_config( - plugin, mqtt, (neu_parse_update_group_config_req_t *) result); - neu_parse_decode_update_group_config_free(result); + } + case NEU_MQTT_OP_UPDATE_GROUP_CONFIG: { + rc = neu_parse_decode_update_group_config( + json_str, (neu_parse_update_group_config_req_t **) &result); + if (0 == rc) { + char *ret_str = NULL; + ret_str = command_update_group_config( + plugin, mqtt, (neu_parse_update_group_config_req_t *) result); + neu_parse_decode_update_group_config_free(result); + MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + } break; - case NEU_MQTT_OP_DELETE_GROUP_CONFIG: - message_handle_delete_group_config( - plugin, mqtt, (neu_parse_del_group_config_req_t *) result); - neu_parse_decode_del_group_config_free(result); - break; - case NEU_MQTT_OP_ADD_DATATAG_IDS_CONFIG: - // message_handle_add_datatag_ids( - // plugin, (neu_parse_add_tag_ids_req_t *) result); - break; - case NEU_MQTT_OP_DELETE_DATATAG_IDS_CONFIG: - // message_handle_delete_datatag_ids( - // plugin, (struct neu_parse_delete_tag_ids_req *) result); + } + case NEU_MQTT_OP_DELETE_GROUP_CONFIG: { + rc = neu_parse_decode_del_group_config( + json_str, (neu_parse_del_group_config_req_t **) &result); + if (0 == rc) { + char *ret_str = NULL; + ret_str = command_delete_group_config( + plugin, mqtt, (neu_parse_del_group_config_req_t *) result); + neu_parse_decode_del_group_config_free(result); + MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + } break; + } case NEU_MQTT_OP_READ: { rc = neu_parse_decode_read(json_str, (neu_parse_read_req_t **) &result); if (0 == rc) { @@ -128,16 +161,54 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, } break; } - case NEU_MQTT_OP_GET_TAGS: - message_handle_get_tag_list(plugin, mqtt, - (neu_parse_get_tags_req_t *) result); - neu_parse_decode_get_tags_free(result); + case NEU_MQTT_OP_GET_TAGS: { + rc = neu_parse_decode_get_tags(json_str, + (neu_parse_get_tags_req_t **) &result); + if (0 == rc) { + char *ret_str = NULL; + ret_str = command_get_tags(plugin, mqtt, + (neu_parse_get_tags_req_t *) result); + neu_parse_decode_get_tags_free(result); + MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + } break; - case NEU_MQTT_OP_ADD_TAGS: - message_handle_add_tag(plugin, mqtt, - (neu_parse_add_tags_req_t *) result); - neu_parse_decode_add_tags_free(result); + } + case NEU_MQTT_OP_ADD_TAGS: { + rc = neu_parse_decode_add_tags(json_str, + (neu_parse_add_tags_req_t **) &result); + if (0 == rc) { + char *ret_str = NULL; + ret_str = command_add_tags(plugin, mqtt, + (neu_parse_add_tags_req_t *) result); + neu_parse_decode_add_tags_free(result); + MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + } break; + } + case NEU_MQTT_OP_UPDATE_TAGS: { + rc = neu_parse_decode_update_tags( + json_str, (neu_parse_update_tags_req_t **) &result); + if (0 == rc) { + char *ret_str = NULL; + ret_str = command_update_tags( + plugin, mqtt, (neu_parse_update_tags_req_t *) result); + neu_parse_decode_update_tags_free(result); + MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + } + break; + } + case NEU_MQTT_OP_DELETE_TAGS: { + rc = neu_parse_decode_del_tags(json_str, + (neu_parse_del_tags_req_t **) &result); + if (0 == rc) { + char *ret_str = NULL; + ret_str = command_delete_tags(plugin, mqtt, + (neu_parse_del_tags_req_t *) result); + neu_parse_decode_del_tags_free(result); + MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str); + } + break; + } default: break; } @@ -258,13 +329,7 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req) 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); - } + MQTT_SEND(plugin->paho, "neuronlite/response", 0, json_str); break; } case NEU_REQRESP_WRITE_RESP: { @@ -272,33 +337,17 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req) 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); - } + MQTT_SEND(plugin->paho, "neuronlite/response", 0, json_str); break; } case NEU_REQRESP_TRANS_DATA: { neu_reqresp_data_t *neu_data; 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("------------------------------------"); + MQTT_SEND(plugin->paho, "neuronlite/response", 0, json_str); break; } - case NEU_REQRESP_WRITE_DATA: { break; }