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

Refactor the mqtt driver management interface

Browse Source 
1. Refactor the group config operation interface
2. Refactor the datatag table operation interface
This commit is contained in:
gc87
2021-10-14 17:00:11 +08:00
committed by Parker Liu
parent 2393f52626
commit 83884fd7c6
9 changed files with 942 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/CMakeLists.txt
+10 -1
View File
@@ -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
plugins/mqtt/command_datatag.c
+402
View File
@@ -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 <stdlib.h>
#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;
}
plugins/mqtt/command_datatag.h
+43
View File
@@ -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 <neuron.h>
#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
plugins/mqtt/command_group_config.c
+258
View File
@@ -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 <stdlib.h>
#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;
}
plugins/mqtt/command_group_config.h
+43
View File
@@ -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 <neuron.h>
#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
plugins/mqtt/command_rw.c
+3 -3
View File
@@ -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;
}
plugins/mqtt/common.c
+61
View File
@@ -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;
}
plugins/mqtt/common.h
+16 -1
View File
@@ -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
plugins/mqtt/plugin_mqtt.c
+106 -57
View File
@@ -24,6 +24,8 @@
#include <config.h>
#include <neuron.h>
#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;
}