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

Reorganize the source code of the mqtt command

Browse Source
This commit is contained in:
gc87
2021-11-09 10:31:23 +08:00
committed by Parker Liu
parent c0d7e62640
commit bbbdb3007c
12 changed files with 270 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/CMakeLists.txt
+5 -4
View File
@@ -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})
plugins/mqtt/command/command.c
+184
View File
@@ -0,0 +1,184 @@
#include <stdio.h>
#include <stdlib.h>
#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);
}
}
plugins/mqtt/command/command.h
+53
View File
@@ -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 <neuron.h>
#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
plugins/mqtt/command/command_datatag.c → plugins/mqtt/command/datatag.c
+1 -1
View File
@@ -19,7 +19,7 @@
#include <stdlib.h>
#include "command_datatag.h"
#include "datatag.h"
typedef struct {
char * name;
plugins/mqtt/command/command_datatag.h → plugins/mqtt/command/datatag.h
View File
plugins/mqtt/command/command_group_config.c → plugins/mqtt/command/group_config.c
+1 -1
View File
@@ -19,7 +19,7 @@
#include <stdlib.h>
#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)
plugins/mqtt/command/command_group_config.h → plugins/mqtt/command/group_config.h
View File
plugins/mqtt/command/command_node.c → plugins/mqtt/command/node.c
+1 -1
View File
@@ -19,7 +19,7 @@
#include <stdlib.h>
#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)
plugins/mqtt/command/command_node.h → plugins/mqtt/command/node.h
View File
plugins/mqtt/command/command_rw.c → plugins/mqtt/command/read_write.c
+1 -1
View File
@@ -21,7 +21,7 @@
#include <stdio.h>
#include <stdlib.h>
#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)
plugins/mqtt/command/command_rw.h → plugins/mqtt/command/read_write.h
+2 -2
View File
@@ -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" {
plugins/mqtt/plugin_mqtt.c
+22 -185
View File
@@ -21,15 +21,10 @@
#include <pthread.h>
#include <stdlib.h>
#include <neuron.h>
#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 <neuron.h>
#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,