From bbbdb3007c04bacc920dd22194cccf7624dcadee Mon Sep 17 00:00:00 2001 From: gc87 Date: Tue, 9 Nov 2021 10:31:23 +0800 Subject: [PATCH] Reorganize the source code of the mqtt command --- plugins/mqtt/CMakeLists.txt | 9 +- plugins/mqtt/command/command.c | 184 ++++++++++++++++ plugins/mqtt/command/command.h | 53 +++++ .../command/{command_datatag.c => datatag.c} | 2 +- .../command/{command_datatag.h => datatag.h} | 0 ...{command_group_config.c => group_config.c} | 2 +- ...{command_group_config.h => group_config.h} | 0 .../mqtt/command/{command_node.c => node.c} | 2 +- .../mqtt/command/{command_node.h => node.h} | 0 .../command/{command_rw.c => read_write.c} | 2 +- .../command/{command_rw.h => read_write.h} | 4 +- plugins/mqtt/plugin_mqtt.c | 207 ++---------------- 12 files changed, 270 insertions(+), 195 deletions(-) create mode 100644 plugins/mqtt/command/command.c create mode 100644 plugins/mqtt/command/command.h rename plugins/mqtt/command/{command_datatag.c => datatag.c} (99%) rename plugins/mqtt/command/{command_datatag.h => datatag.h} (100%) rename plugins/mqtt/command/{command_group_config.c => group_config.c} (99%) rename plugins/mqtt/command/{command_group_config.h => group_config.h} (100%) rename plugins/mqtt/command/{command_node.c => node.c} (99%) rename plugins/mqtt/command/{command_node.h => node.h} (100%) rename plugins/mqtt/command/{command_rw.c => read_write.c} (99%) rename plugins/mqtt/command/{command_rw.h => read_write.h} (95%) diff --git a/plugins/mqtt/CMakeLists.txt b/plugins/mqtt/CMakeLists.txt index 961616c9..b668fa2a 100644 --- a/plugins/mqtt/CMakeLists.txt +++ b/plugins/mqtt/CMakeLists.txt @@ -6,10 +6,11 @@ set(src mqttc_util.c mqttc_client.c command/common.c - command/command_rw.c - command/command_group_config.c - command/command_datatag.c - command/command_node.c) + command/read_write.c + command/group_config.c + command/datatag.c + command/node.c + command/command.c) add_library(plugin-mqtt SHARED ${src}) diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c new file mode 100644 index 00000000..a43276b3 --- /dev/null +++ b/plugins/mqtt/command/command.c @@ -0,0 +1,184 @@ +#include +#include + +#include "command.h" + +void command_response_handle(mqtt_response_t *response) +{ + neu_plugin_t *plugin = (neu_plugin_t *) response->context; + + if (NULL == response->topic_name || NULL == response->payload) { + return; + } + + char *json_str = malloc(response->len + 1); + if (NULL == json_str) { + return; + } + + memset(json_str, 0x00, response->len + 1); + memcpy(json_str, response->payload, response->len); + + neu_parse_mqtt_t *mqtt = NULL; + char * ret_str = NULL; + int rc = neu_parse_decode_mqtt_param(json_str, &mqtt); + + if (0 != rc) { + log_error("JSON parsing mqtt failed"); + return; + } + + switch (mqtt->function) { + case NEU_MQTT_OP_GET_GROUP_CONFIG: { + neu_parse_get_group_config_req_t *req = NULL; + rc = neu_parse_decode_get_group_config(json_str, &req); + if (0 == rc) { + ret_str = command_get_group_configs(plugin, mqtt, req); + neu_parse_decode_get_group_config_free(req); + } + break; + } + case NEU_MQTT_OP_ADD_GROUP_CONFIG: { + neu_parse_add_group_config_req_t *req = NULL; + rc = neu_parse_decode_add_group_config(json_str, &req); + if (0 == rc) { + ret_str = command_add_group_config(plugin, mqtt, req); + neu_parse_decode_add_group_config_free(req); + } + break; + } + case NEU_MQTT_OP_UPDATE_GROUP_CONFIG: { + neu_parse_update_group_config_req_t *req = NULL; + rc = neu_parse_decode_update_group_config(json_str, &req); + if (0 == rc) { + ret_str = command_update_group_config(plugin, mqtt, req); + neu_parse_decode_update_group_config_free(req); + } + break; + } + case NEU_MQTT_OP_DELETE_GROUP_CONFIG: { + neu_parse_del_group_config_req_t *req = NULL; + rc = neu_parse_decode_del_group_config(json_str, &req); + if (0 == rc) { + ret_str = command_delete_group_config(plugin, mqtt, req); + neu_parse_decode_del_group_config_free(req); + } + break; + } + case NEU_MQTT_OP_READ: { + neu_parse_read_req_t *req = NULL; + rc = neu_parse_decode_read(json_str, &req); + if (0 == rc) { + uint32_t req_id = neu_plugin_get_event_id(plugin); + if (0 < req_id) { + response->context_add(plugin, req_id, mqtt); + } + + command_read_once_request(plugin, mqtt, req, req_id); + neu_parse_decode_read_free(req); + } + break; + } + case NEU_MQTT_OP_WRITE: { + neu_parse_write_req_t *req = NULL; + rc = neu_parse_decode_write(json_str, &req); + if (0 == rc) { + uint32_t req_id = neu_plugin_get_event_id(plugin); + if (0 < req_id) { + response->context_add(plugin, req_id, mqtt); + } + + command_write_request(plugin, mqtt, req, req_id); + neu_parse_decode_write_free(req); + } + break; + } + case NEU_MQTT_OP_GET_TAGS: { + neu_parse_get_tags_req_t *req = NULL; + rc = neu_parse_decode_get_tags(json_str, &req); + if (0 == rc) { + ret_str = command_get_tags(plugin, mqtt, req); + neu_parse_decode_get_tags_free(req); + } + break; + } + case NEU_MQTT_OP_ADD_TAGS: { + neu_parse_add_tags_req_t *req = NULL; + rc = neu_parse_decode_add_tags(json_str, &req); + if (0 == rc) { + ret_str = command_add_tags(plugin, mqtt, req); + neu_parse_decode_add_tags_free(req); + } + break; + } + case NEU_MQTT_OP_UPDATE_TAGS: { + neu_parse_update_tags_req_t *req = NULL; + rc = neu_parse_decode_update_tags(json_str, &req); + if (0 == rc) { + ret_str = command_update_tags(plugin, mqtt, req); + neu_parse_decode_update_tags_free(req); + } + break; + } + case NEU_MQTT_OP_DELETE_TAGS: { + neu_parse_del_tags_req_t *req = NULL; + rc = neu_parse_decode_del_tags(json_str, &req); + if (0 == rc) { + ret_str = command_delete_tags(plugin, mqtt, req); + neu_parse_decode_del_tags_free(req); + } + break; + } + case NEU_MQTT_OP_GET_NODES: { + neu_parse_get_nodes_req_t *req = NULL; + rc = neu_parse_decode_get_nodes(json_str, &req); + if (0 == rc) { + ret_str = command_get_nodes(plugin, mqtt, req); + neu_parse_decode_get_nodes_free(req); + } + break; + } + case NEU_MQTT_OP_ADD_NODES: { + neu_parse_add_node_req_t *req = NULL; + rc = neu_parse_decode_add_node(json_str, &req); + if (0 == rc) { + ret_str = command_add_node(plugin, mqtt, req); + neu_parse_decode_add_node_free(req); + } + break; + } + case NEU_MQTT_OP_UPDATE_NODES: { + neu_parse_update_node_req_t *req = NULL; + rc = neu_parse_decode_update_node(json_str, &req); + if (0 == rc) { + ret_str = command_update_node(plugin, mqtt, req); + neu_parse_decode_update_node_free(req); + } + break; + } + case NEU_MQTT_OP_DELETE_NODES: { + neu_parse_del_node_req_t *req = NULL; + rc = neu_parse_decode_del_node(json_str, &req); + if (0 == rc) { + ret_str = command_delete_node(plugin, mqtt, req); + neu_parse_decode_del_node_free(req); + } + break; + } + default: + break; + } + + response->mqtt_send(plugin, ret_str); + + if (NULL != mqtt) { + if (NULL != mqtt->uuid) { + free(mqtt->uuid); + } + free(mqtt); + } + + if (NULL != json_str) { + free(json_str); + } +} diff --git a/plugins/mqtt/command/command.h b/plugins/mqtt/command/command.h new file mode 100644 index 00000000..82746deb --- /dev/null +++ b/plugins/mqtt/command/command.h @@ -0,0 +1,53 @@ +/** + * 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 +#define NEURON_PLUGIN_MQTT_COMMAND + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#include "datatag.h" +#include "group_config.h" +#include "node.h" +#include "read_write.h" + +typedef void (*mqtt_send_callback)(neu_plugin_t *plugin, char *json_str); +typedef void (*context_add_callback)(neu_plugin_t *plugin, uint32_t req_id, + neu_parse_mqtt_t *parse_head); + +typedef struct { + const char * topic_name; + size_t topic_len; + void * payload; + size_t len; + void * context; + mqtt_send_callback mqtt_send; + context_add_callback context_add; +} mqtt_response_t; + +void command_response_handle(mqtt_response_t *response); + +#ifdef __cplusplus +} +#endif +#endif \ No newline at end of file diff --git a/plugins/mqtt/command/command_datatag.c b/plugins/mqtt/command/datatag.c similarity index 99% rename from plugins/mqtt/command/command_datatag.c rename to plugins/mqtt/command/datatag.c index b8e405b4..1d0c5b49 100644 --- a/plugins/mqtt/command/command_datatag.c +++ b/plugins/mqtt/command/datatag.c @@ -19,7 +19,7 @@ #include -#include "command_datatag.h" +#include "datatag.h" typedef struct { char * name; diff --git a/plugins/mqtt/command/command_datatag.h b/plugins/mqtt/command/datatag.h similarity index 100% rename from plugins/mqtt/command/command_datatag.h rename to plugins/mqtt/command/datatag.h diff --git a/plugins/mqtt/command/command_group_config.c b/plugins/mqtt/command/group_config.c similarity index 99% rename from plugins/mqtt/command/command_group_config.c rename to plugins/mqtt/command/group_config.c index 208da822..faa8fc5a 100644 --- a/plugins/mqtt/command/command_group_config.c +++ b/plugins/mqtt/command/group_config.c @@ -19,7 +19,7 @@ #include -#include "command_group_config.h" +#include "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) diff --git a/plugins/mqtt/command/command_group_config.h b/plugins/mqtt/command/group_config.h similarity index 100% rename from plugins/mqtt/command/command_group_config.h rename to plugins/mqtt/command/group_config.h diff --git a/plugins/mqtt/command/command_node.c b/plugins/mqtt/command/node.c similarity index 99% rename from plugins/mqtt/command/command_node.c rename to plugins/mqtt/command/node.c index 46301284..bbcb2997 100644 --- a/plugins/mqtt/command/command_node.c +++ b/plugins/mqtt/command/node.c @@ -19,7 +19,7 @@ #include -#include "command_node.h" +#include "node.h" char *command_get_nodes(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, neu_parse_get_nodes_req_t *req) diff --git a/plugins/mqtt/command/command_node.h b/plugins/mqtt/command/node.h similarity index 100% rename from plugins/mqtt/command/command_node.h rename to plugins/mqtt/command/node.h diff --git a/plugins/mqtt/command/command_rw.c b/plugins/mqtt/command/read_write.c similarity index 99% rename from plugins/mqtt/command/command_rw.c rename to plugins/mqtt/command/read_write.c index 60e589e9..db8372b9 100644 --- a/plugins/mqtt/command/command_rw.c +++ b/plugins/mqtt/command/read_write.c @@ -21,7 +21,7 @@ #include #include -#include "command_rw.h" +#include "read_write.h" int command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, neu_parse_read_req_t *req, uint32_t req_id) diff --git a/plugins/mqtt/command/command_rw.h b/plugins/mqtt/command/read_write.h similarity index 95% rename from plugins/mqtt/command/command_rw.h rename to plugins/mqtt/command/read_write.h index e718dd22..86f5f642 100644 --- a/plugins/mqtt/command/command_rw.h +++ b/plugins/mqtt/command/read_write.h @@ -17,8 +17,8 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ -#ifndef NEURON_PLUGIN_MQTT_COMMAND_RW -#define NEURON_PLUGIN_MQTT_COMMAND_RW +#ifndef NEURON_PLUGIN_MQTT_COMMAND_READ_WRITE +#define NEURON_PLUGIN_MQTT_COMMAND_READ_WRITE #ifdef __cplusplus extern "C" { diff --git a/plugins/mqtt/plugin_mqtt.c b/plugins/mqtt/plugin_mqtt.c index 535abb92..ab163aab 100644 --- a/plugins/mqtt/plugin_mqtt.c +++ b/plugins/mqtt/plugin_mqtt.c @@ -21,15 +21,10 @@ #include #include -#include - -#include "command/command_datatag.h" -#include "command/command_group_config.h" -#include "command/command_node.h" -#include "command/command_rw.h" - +#include "command/command.h" #include "mqttc_client.h" #include "mqttc_util.h" +#include #define UNUSED(x) (void) (x) @@ -130,188 +125,30 @@ static void context_list_destroy(neu_list *list) } } +static void mqtt_send(neu_plugin_t *plugin, char *json_str) +{ + MQTT_SEND(plugin->mqtt_client, "neuronlite/response", 0, json_str); +} + +static void context_add(neu_plugin_t *plugin, uint32_t req_id, + neu_parse_mqtt_t *parse_header) +{ + context_list_add(&plugin->context_list, req_id, parse_header); +} + static void plugin_response_handle(const char *topic_name, size_t topic_len, void *payload, const size_t len, void *context) { - UNUSED(topic_len); - - neu_plugin_t *plugin = (neu_plugin_t *) context; - - if (NULL == topic_name || NULL == payload) { - return; - } - - char *json_str = malloc(len + 1); - if (NULL == json_str) { - return; - } - - memset(json_str, 0x00, len + 1); - memcpy(json_str, payload, len); - - neu_parse_mqtt_t *mqtt = NULL; - char * ret_str = NULL; - int rc = neu_parse_decode_mqtt_param(json_str, &mqtt); - - if (0 != rc) { - log_error("JSON parsing mqtt failed"); - return; - } - - switch (mqtt->function) { - case NEU_MQTT_OP_GET_GROUP_CONFIG: { - neu_parse_get_group_config_req_t *req = NULL; - rc = neu_parse_decode_get_group_config(json_str, &req); - if (0 == rc) { - ret_str = command_get_group_configs(plugin, mqtt, req); - neu_parse_decode_get_group_config_free(req); - } - break; - } - case NEU_MQTT_OP_ADD_GROUP_CONFIG: { - neu_parse_add_group_config_req_t *req = NULL; - rc = neu_parse_decode_add_group_config(json_str, &req); - if (0 == rc) { - ret_str = command_add_group_config(plugin, mqtt, req); - neu_parse_decode_add_group_config_free(req); - } - break; - } - case NEU_MQTT_OP_UPDATE_GROUP_CONFIG: { - neu_parse_update_group_config_req_t *req = NULL; - rc = neu_parse_decode_update_group_config(json_str, &req); - if (0 == rc) { - ret_str = command_update_group_config(plugin, mqtt, req); - neu_parse_decode_update_group_config_free(req); - } - break; - } - case NEU_MQTT_OP_DELETE_GROUP_CONFIG: { - neu_parse_del_group_config_req_t *req = NULL; - rc = neu_parse_decode_del_group_config(json_str, &req); - if (0 == rc) { - ret_str = command_delete_group_config(plugin, mqtt, req); - neu_parse_decode_del_group_config_free(req); - } - break; - } - case NEU_MQTT_OP_READ: { - neu_parse_read_req_t *req = NULL; - rc = neu_parse_decode_read(json_str, &req); - if (0 == rc) { - uint32_t req_id = neu_plugin_get_event_id(plugin); - if (0 < req_id) { - context_list_add(&plugin->context_list, req_id, mqtt); - } - - command_read_once_request(plugin, mqtt, req, req_id); - neu_parse_decode_read_free(req); - } - break; - } - case NEU_MQTT_OP_WRITE: { - neu_parse_write_req_t *req = NULL; - rc = neu_parse_decode_write(json_str, &req); - if (0 == rc) { - uint32_t req_id = neu_plugin_get_event_id(plugin); - if (0 < req_id) { - context_list_add(&plugin->context_list, req_id, mqtt); - } - - command_write_request(plugin, mqtt, req, req_id); - neu_parse_decode_write_free(req); - } - break; - } - case NEU_MQTT_OP_GET_TAGS: { - neu_parse_get_tags_req_t *req = NULL; - rc = neu_parse_decode_get_tags(json_str, &req); - if (0 == rc) { - ret_str = command_get_tags(plugin, mqtt, req); - neu_parse_decode_get_tags_free(req); - } - break; - } - case NEU_MQTT_OP_ADD_TAGS: { - neu_parse_add_tags_req_t *req = NULL; - rc = neu_parse_decode_add_tags(json_str, &req); - if (0 == rc) { - ret_str = command_add_tags(plugin, mqtt, req); - neu_parse_decode_add_tags_free(req); - } - break; - } - case NEU_MQTT_OP_UPDATE_TAGS: { - neu_parse_update_tags_req_t *req = NULL; - rc = neu_parse_decode_update_tags(json_str, &req); - if (0 == rc) { - ret_str = command_update_tags(plugin, mqtt, req); - neu_parse_decode_update_tags_free(req); - } - break; - } - case NEU_MQTT_OP_DELETE_TAGS: { - neu_parse_del_tags_req_t *req = NULL; - rc = neu_parse_decode_del_tags(json_str, &req); - if (0 == rc) { - ret_str = command_delete_tags(plugin, mqtt, req); - neu_parse_decode_del_tags_free(req); - } - break; - } - case NEU_MQTT_OP_GET_NODES: { - neu_parse_get_nodes_req_t *req = NULL; - rc = neu_parse_decode_get_nodes(json_str, &req); - if (0 == rc) { - ret_str = command_get_nodes(plugin, mqtt, req); - neu_parse_decode_get_nodes_free(req); - } - break; - } - case NEU_MQTT_OP_ADD_NODES: { - neu_parse_add_node_req_t *req = NULL; - rc = neu_parse_decode_add_node(json_str, &req); - if (0 == rc) { - ret_str = command_add_node(plugin, mqtt, req); - neu_parse_decode_add_node_free(req); - } - break; - } - case NEU_MQTT_OP_UPDATE_NODES: { - neu_parse_update_node_req_t *req = NULL; - rc = neu_parse_decode_update_node(json_str, &req); - if (0 == rc) { - ret_str = command_update_node(plugin, mqtt, req); - neu_parse_decode_update_node_free(req); - } - break; - } - case NEU_MQTT_OP_DELETE_NODES: { - neu_parse_del_node_req_t *req = NULL; - rc = neu_parse_decode_del_node(json_str, &req); - if (0 == rc) { - ret_str = command_delete_node(plugin, mqtt, req); - neu_parse_decode_del_node_free(req); - } - break; - } - default: - break; - } - - MQTT_SEND(plugin->mqtt_client, "neuronlite/response", 0, ret_str); - - if (NULL != mqtt) { - if (NULL != mqtt->uuid) { - free(mqtt->uuid); - } - free(mqtt); - } - - if (NULL != json_str) { - free(json_str); - } + mqtt_response_t response = { .topic_name = topic_name, + .topic_len = topic_len, + .payload = payload, + .len = len, + .context = context, + .mqtt_send = mqtt_send, + .context_add = context_add }; + command_response_handle(&response); + return; } static neu_plugin_t *mqtt_plugin_open(neu_adapter_t * adapter,