Add mqtt command interface about adapter

This commit is contained in:
gc87
2021-10-14 01:46:34 +08:00
committed by Parker Liu
parent 0acbf3e34c
commit 90721f520d
10 changed files with 272 additions and 106 deletions
+2 -1
View File
@@ -11,7 +11,8 @@ set(src
common.c
command_rw.c
command_group_config.c
command_datatag.c)
command_datatag.c
command_node.c)
add_library(plugin-mqtt SHARED ${src})
target_include_directories(plugin-mqtt PRIVATE
+1 -1
View File
@@ -103,7 +103,7 @@ static int get_all_tag(neu_plugin_t *plugin, const uint32_t node_id,
}
struct visit_context ctx = { .table = table, .tags = tags };
common_group_configs_freach(plugin, node_id, &ctx, visit_func);
common_group_configs_foreach(plugin, node_id, &ctx, visit_func);
return 0;
}
+4 -8
View File
@@ -109,15 +109,11 @@ char *command_get_group_configs(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
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;
return -1;
}
vector_t configs = neu_system_get_group_configs(plugin, node_id);
@@ -125,17 +121,17 @@ static int add_group_config(neu_plugin_t *plugin, const char *config_name,
GROUP_CONFIGS_UNINIT(configs);
if (0 == rc) {
return -4;
return -2;
}
neu_taggrp_config_t *grp_config = neu_taggrp_cfg_new((char *) config_name);
if (NULL == grp_config) {
return -5;
return -3;
}
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 -4;
}
return 0;
}
+102
View File
@@ -0,0 +1,102 @@
/**
* 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_node.h"
char *command_get_nodes(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_get_nodes_req_t *req)
{
log_info("Get node list uuid:%s, node type:%d", mqtt->uuid, req->node_type);
neu_node_type_e node_type = req->node_type;
neu_parse_get_nodes_res_t res = { 0 };
int index = 0;
vector_t nodes = neu_system_get_nodes(plugin, node_type);
res.n_node = nodes.size;
res.nodes = malloc(res.n_node * sizeof(neu_parse_get_nodes_res_node_t));
memset(res.nodes, 0, res.n_node * sizeof(neu_parse_get_nodes_res_node_t));
VECTOR_FOR_EACH(&nodes, iter)
{
neu_node_info_t *info = (neu_node_info_t *) iterator_get(&iter);
res.nodes[index].id = info->node_id;
res.nodes[index].name = info->node_name;
index += 1;
}
char *json_str = NULL;
int rc = neu_json_encode_with_mqtt(&res, neu_parse_encode_get_tags, mqtt,
neu_parse_encode_mqtt_param, &json_str);
if (0 == rc) {
return json_str;
}
return NULL;
}
char *command_add_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_add_node_req_t *req)
{
log_info("Add node uuid:%s, node type:%d", mqtt->uuid, req->node_type);
intptr_t rc = neu_system_add_node(plugin, req->node_type, req->node_name,
req->plugin_name);
char * json_str = NULL;
if (rc != 0) {
json_str = strdup("{\"error\": 1}");
} else {
json_str = strdup("{\"error\": 0}");
}
return json_str;
}
char *command_update_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_update_node_req_t *req)
{
log_info("Update node uuid:%s, node type:%d", mqtt->uuid, req->node_type);
intptr_t rc = neu_system_update_node(plugin, req->node_type, req->node_name,
req->plugin_name);
char * json_str = NULL;
if (rc != 0) {
json_str = strdup("{\"error\": 1}");
} else {
json_str = strdup("{\"error\": 0}");
}
return json_str;
}
char *command_delete_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_del_node_req_t *req)
{
log_info("Delete node uuid:%s, node id:%d", mqtt->uuid, req->node_id);
intptr_t rc = neu_system_del_node(plugin, req->node_id);
char * json_str = NULL;
if (rc != 0) {
json_str = strdup("{\"error\": 1}");
} else {
json_str = strdup("{\"error\": 0}");
}
return json_str;
}
+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_NODE
#define NEURON_PLUGIN_MQTT_COMMAND_NODE
#ifdef __cplusplus
extern "C" {
#endif
#include <neuron.h>
#include "common.h"
char *command_get_nodes(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_get_nodes_req_t *req);
char *command_add_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_add_node_req_t *req);
char *command_update_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_update_node_req_t *req);
char *command_delete_node(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_del_node_req_t *req);
#ifdef __cplusplus
}
#endif
#endif
+19 -16
View File
@@ -27,22 +27,20 @@ void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req)
{
log_info("READ uuid:%s, node id:%u", mqtt->uuid, req->node_id);
vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER);
int rc = common_node_id_exist(&nodes, req->node_id);
vector_uninit(&nodes);
int rc = common_has_node(plugin, req->node_id);
if (0 != rc) {
log_debug("The requested node does not exist");
return;
}
vector_t configs = neu_system_get_group_configs(plugin, req->node_id);
neu_taggrp_config_t *config;
neu_taggrp_config_t *c;
config = *(neu_taggrp_config_t **) vector_get(&configs, 0);
c = (neu_taggrp_config_t *) neu_taggrp_cfg_ref(config);
uint32_t req_id = neu_plugin_send_read_cmd(plugin, req->node_id, c);
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;
}
GROUP_CONFIGS_UNINIT(configs);
uint32_t req_id = neu_plugin_send_read_cmd(plugin, req->node_id, config);
UNUSED(req_id);
}
@@ -182,12 +180,13 @@ static void clean_read_response_json_object(neu_parse_read_res_t *json)
free(json->tags);
}
char *command_read_once_response(neu_plugin_t *plugin, const char *uuid,
neu_data_val_t *resp_val)
char *command_read_once_response(neu_plugin_t * plugin,
neu_parse_mqtt_t *parse_header,
neu_data_val_t * resp_val)
{
UNUSED(plugin);
UNUSED(uuid);
UNUSED(resp_val);
UNUSED(parse_header);
neu_fixed_array_t *array;
int rc = neu_dvalue_get_ref_array(resp_val, &array);
@@ -229,8 +228,12 @@ char *command_read_once_response(neu_plugin_t *plugin, const char *uuid,
char *command_read_cycle_response(neu_plugin_t * plugin,
neu_data_val_t *resp_val)
{
// TODO: Generate UUID
return command_read_once_response(plugin, "", resp_val);
neu_parse_mqtt_t parse_header = { 0 };
parse_header.function = NEU_MQTT_OP_READ;
char uuid4_str[40] = { '\0' };
neu_uuid_v4_gen(uuid4_str);
parse_header.uuid = uuid4_str;
return command_read_once_response(plugin, &parse_header, resp_val);
}
static bool match_name_grp_config(const void *key, const void *item)
+3 -2
View File
@@ -30,8 +30,9 @@ extern "C" {
void command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt,
neu_parse_read_req_t *req);
char *command_read_once_response(neu_plugin_t *plugin, const char *uuid,
neu_data_val_t *resp_val);
char *command_read_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,
+4 -3
View File
@@ -95,9 +95,10 @@ neu_taggrp_config_t *common_get_group_config(neu_plugin_t * plugin,
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 *))
void common_group_configs_foreach(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;
+5 -4
View File
@@ -28,6 +28,7 @@ extern "C" {
#include "parser/neu_json_fn.h"
#include "parser/neu_json_group_config.h"
#include "parser/neu_json_mqtt.h"
#include "parser/neu_json_node.h"
#include "parser/neu_json_rw.h"
#include "parser/neu_json_tag.h"
@@ -54,10 +55,10 @@ int common_has_group_config(neu_plugin_t *plugin, const uint32_t node_id,
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 *));
void common_group_configs_foreach(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);
+89 -71
View File
@@ -26,6 +26,7 @@
#include "command_datatag.h"
#include "command_group_config.h"
#include "command_node.h"
#include "command_rw.h"
#include "option.h"
@@ -51,6 +52,7 @@ struct neu_plugin {
option_t option;
paho_client_t * paho;
neu_list context_list;
neu_parse_mqtt_t * parse_header;
};
static int plugin_subscribe(neu_plugin_t *plugin, const char *topic,
@@ -83,128 +85,141 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len,
memset(json_str, 0x00, len + 1);
memcpy(json_str, payload, len);
void * result = NULL;
neu_parse_mqtt_t *mqtt = NULL;
int rc = neu_parse_decode_mqtt_param(json_str, &mqtt);
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;
}
plugin->parse_header = mqtt;
switch (mqtt->function) {
case NEU_MQTT_OP_GET_GROUP_CONFIG: {
rc = neu_parse_decode_get_group_config(
json_str, (neu_parse_get_group_config_req_t **) &result);
neu_parse_get_group_config_req_t *req = NULL;
rc = neu_parse_decode_get_group_config(json_str, &req);
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);
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: {
rc = neu_parse_decode_add_group_config(
json_str, (neu_parse_add_group_config_req_t **) &result);
neu_parse_add_group_config_req_t *req = NULL;
rc = neu_parse_decode_add_group_config(json_str, &req);
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);
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: {
rc = neu_parse_decode_update_group_config(
json_str, (neu_parse_update_group_config_req_t **) &result);
neu_parse_update_group_config_req_t *req = NULL;
rc = neu_parse_decode_update_group_config(json_str, &req);
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);
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: {
rc = neu_parse_decode_del_group_config(
json_str, (neu_parse_del_group_config_req_t **) &result);
neu_parse_del_group_config_req_t *req;
rc = neu_parse_decode_del_group_config(json_str, &req);
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);
ret_str = command_delete_group_config(plugin, mqtt, req);
neu_parse_decode_del_group_config_free(req);
}
break;
}
case NEU_MQTT_OP_READ: {
rc = neu_parse_decode_read(json_str, (neu_parse_read_req_t **) &result);
neu_parse_read_req_t *req = NULL;
rc = neu_parse_decode_read(json_str, &req);
if (0 == rc) {
command_read_once_request(plugin, mqtt,
(neu_parse_read_req_t *) result);
neu_parse_decode_read_free((neu_parse_read_req_t *) result);
command_read_once_request(plugin, mqtt, req);
neu_parse_decode_read_free(req);
}
break;
}
case NEU_MQTT_OP_WRITE: {
rc = neu_parse_decode_write(json_str,
(neu_parse_write_req_t **) &result);
neu_parse_write_req_t *req = NULL;
rc = neu_parse_decode_write(json_str, &req);
if (0 == rc) {
command_write_request(plugin, mqtt,
(neu_parse_write_req_t *) result);
neu_parse_decode_write_free(result);
command_write_request(plugin, mqtt, req);
neu_parse_decode_write_free(req);
}
break;
}
case NEU_MQTT_OP_GET_TAGS: {
rc = neu_parse_decode_get_tags(json_str,
(neu_parse_get_tags_req_t **) &result);
neu_parse_get_tags_req_t *req;
rc = neu_parse_decode_get_tags(json_str, &req);
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);
ret_str = command_get_tags(plugin, mqtt, req);
neu_parse_decode_get_tags_free(req);
}
break;
}
case NEU_MQTT_OP_ADD_TAGS: {
rc = neu_parse_decode_add_tags(json_str,
(neu_parse_add_tags_req_t **) &result);
neu_parse_add_tags_req_t *req = NULL;
rc = neu_parse_decode_add_tags(json_str, &req);
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);
ret_str = command_add_tags(plugin, mqtt, req);
neu_parse_decode_add_tags_free(req);
}
break;
}
case NEU_MQTT_OP_UPDATE_TAGS: {
rc = neu_parse_decode_update_tags(
json_str, (neu_parse_update_tags_req_t **) &result);
neu_parse_update_tags_req_t *req = NULL;
rc = neu_parse_decode_update_tags(json_str, &req);
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);
ret_str = command_update_tags(plugin, mqtt, req);
neu_parse_decode_update_tags_free(req);
}
break;
}
case NEU_MQTT_OP_DELETE_TAGS: {
rc = neu_parse_decode_del_tags(json_str,
(neu_parse_del_tags_req_t **) &result);
neu_parse_del_tags_req_t *req = NULL;
rc = neu_parse_decode_del_tags(json_str, &req);
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);
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;
}
@@ -212,6 +227,7 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len,
break;
}
MQTT_SEND(plugin->paho, "neuronlite/response", 0, ret_str);
free(json_str);
UNUSED(topic_len);
@@ -277,6 +293,8 @@ static int mqtt_plugin_init(neu_plugin_t *plugin)
plugin->option.keepalive_interval = atoi(keepalive_interval);
plugin->option.clean_session = atoi(clean_session);
plugin->parse_header = NULL;
// Paho mqtt client setup
client_error error =
paho_client_open(&plugin->option, plugin, &plugin->paho);
@@ -322,9 +340,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
switch (req->req_type) {
case NEU_REQRESP_READ_RESP: {
neu_reqresp_read_resp_t *read_resp;
read_resp = (neu_reqresp_read_resp_t *) req->buf;
char *json_str =
command_read_once_response(plugin, "", read_resp->data_val);
read_resp = (neu_reqresp_read_resp_t *) req->buf;
char *json_str = command_read_once_response(
plugin, plugin->parse_header, read_resp->data_val);
MQTT_SEND(plugin->paho, "neuronlite/response", 0, json_str);
break;
}