GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-29 07:06:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

mqtt: driver cmd

Browse Source
This commit is contained in:
fengzero
2024-11-11 03:31:09 +00:00
parent ebc91b0ce1
commit b291e49611
15 changed files with 243 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/errcodes.h
+1
View File
@@ -142,6 +142,7 @@ typedef enum {
NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE = 3020,
NEU_ERR_PLUGIN_NOT_SUPPORT_SCAN_TAGS = 3021,
NEU_ERR_PLUGIN_NOT_SUPPORT_TEST_READ_TAG = 3022,
NEU_ERR_PLUGIN_NOT_SUPPORT_CMD_CALL = 3023,
NEU_ERR_MQTT_FAILURE = 4000,
NEU_ERR_MQTT_NO_CERTFILESET = 4001,
include/neuron/json/neu_json_driver.h
+41
View File
@@ -0,0 +1,41 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 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 _NEU_JSON_API_DRIVER_CMD_H_
#define _NEU_JSON_API_DRIVER_CMD_H_
#include "define.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
char *driver;
char *cmd;
} neu_json_driver_cmd_t;
int neu_json_decode_driver_cmd_req(char *buf, neu_json_driver_cmd_t **result);
void neu_json_decode_driver_cmd_req_free(neu_json_driver_cmd_t *req);
#ifdef __cplusplus
}
#endif
#endif
include/neuron/msg.h
+9
View File
@@ -125,6 +125,8 @@ typedef enum neu_reqresp_type {
NEU_REQ_CHECK_SCHEMA,
NEU_RESP_CHECK_SCHEMA,
NEU_REQ_DRIVER_CMD,
} neu_reqresp_type_e;
static const char *neu_reqresp_type_string_t[] = {
@@ -210,6 +212,8 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_CHECK_SCHEMA] = "NEU_REQ_CHECK_SCHEMA",
[NEU_RESP_CHECK_SCHEMA] = "NEU_RESP_CHECK_SCHEMA",
[NEU_REQ_DRIVER_CMD] = "NEU_REQ_DRIVER_CMD",
};
inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type)
@@ -238,6 +242,11 @@ typedef struct neu_resp_error {
int error;
} neu_resp_error_t;
typedef struct neu_req_driver_cmd {
char driver[NEU_NODE_NAME_LEN];
char *cmd;
} neu_req_driver_cmd_t;
typedef struct neu_req_check_schema {
char schema[NEU_PLUGIN_NAME_LEN];
} neu_req_check_schema_t;
include/neuron/plugin.h
+1
View File
@@ -114,6 +114,7 @@ typedef struct neu_plugin_intf_funs {
char *ctx);
int (*test_read_tag)(neu_plugin_t *plugin, void *req,
neu_datatag_t tag);
int (*call)(neu_plugin_t *plugin, const char *cmd);
} driver;
};
plugins/mqtt/mqtt_handle.c
+45
View File
@@ -22,6 +22,7 @@
#include "otel/otel_manager.h"
#include "utils/asprintf.h"
#include "version.h"
#include "json/neu_json_driver.h"
#include "json/neu_json_mqtt.h"
#include "json/neu_json_rw.h"
@@ -179,6 +180,26 @@ static char *generate_heartbeat_json(neu_plugin_t *plugin, UT_array *states,
return json_str;
}
static inline int send_driver_cmd(neu_plugin_t * plugin,
neu_json_driver_cmd_t *req)
{
plog_notice(plugin, "driver cmd, driver:%s, cmd:%s", req->driver, req->cmd);
neu_reqresp_head_t header = { 0 };
header.ctx = NULL;
neu_req_driver_cmd_t cmd = { 0 };
strncpy(cmd.driver, req->driver, NEU_NODE_NAME_LEN);
cmd.cmd = strdup(req->cmd);
if (0 != neu_plugin_op(plugin, header, &cmd)) {
free(cmd.cmd);
plog_error(plugin, "neu_plugin_op(NEU_REQ_DRIVER_CMD) fail");
return -1;
}
return 0;
}
static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_read_req_t *req)
{
@@ -533,6 +554,30 @@ end:
return rv;
}
void handle_driver_cmd_req(neu_mqtt_qos_e qos, const char *topic,
const uint8_t *payload, uint32_t len, void *data,
trace_w3c_t *trace_w3c)
{
(void) qos;
(void) topic;
(void) trace_w3c;
neu_plugin_t * plugin = data;
neu_json_driver_cmd_t *req = NULL;
char *json_str = calloc(len + 1, sizeof(char));
memcpy(json_str, payload, len);
int rv = neu_json_decode_driver_cmd_req(json_str, &req);
if (rv != 0) {
plog_error(plugin, "neu_json_decode_driver_cmd_req failed");
free(json_str);
return;
}
neu_json_decode_driver_cmd_req_free(req);
free(json_str);
}
void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
const uint8_t *payload, uint32_t len, void *data,
trace_w3c_t *trace_w3c)
plugins/mqtt/mqtt_handle.h
+4
View File
@@ -50,6 +50,10 @@ void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
neu_resp_read_group_t *data);
void handle_driver_cmd_req(neu_mqtt_qos_e qos, const char *topic,
const uint8_t *payload, uint32_t len, void *data,
trace_w3c_t *trace_w3c);
char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
mqtt_upload_format_e format);
int handle_trans_data(neu_plugin_t * plugin,
plugins/mqtt/mqtt_plugin.h
+1
View File
@@ -52,6 +52,7 @@ struct neu_plugin {
char * read_req_topic;
char * read_resp_topic;
char * upload_topic;
char * driver_cmd_topic;
route_entry_t * route_tbl;
int (*parse_config)(neu_plugin_t *plugin, const char *setting,
plugins/mqtt/mqtt_plugin_intf.c
+19
View File
@@ -178,6 +178,8 @@ int mqtt_plugin_uninit(neu_plugin_t *plugin)
plugin->read_resp_topic = NULL;
free(plugin->upload_topic);
plugin->upload_topic = NULL;
free(plugin->driver_cmd_topic);
plugin->driver_cmd_topic = NULL;
route_tbl_free(plugin->route_tbl);
@@ -279,6 +281,14 @@ static int create_topic(neu_plugin_t *plugin)
return -1;
}
neu_asprintf(&plugin->driver_cmd_topic, "/neuron/%s/driver/cmd",
plugin->common.name);
if (NULL == plugin->driver_cmd_topic) {
free(plugin->driver_cmd_topic);
plugin->driver_cmd_topic = NULL;
return -1;
}
return 0;
}
@@ -306,6 +316,14 @@ static int subscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
}
if (0 !=
neu_mqtt_client_subscribe(plugin->client, config->qos,
plugin->driver_cmd_topic, plugin,
handle_driver_cmd_req)) {
plog_error(plugin, "subscribe [%s] fail", plugin->driver_cmd_topic);
return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
}
return 0;
}
@@ -313,6 +331,7 @@ static int unsubscribe(neu_plugin_t *plugin, const mqtt_config_t *config)
{
neu_mqtt_client_unsubscribe(plugin->client, plugin->read_req_topic);
neu_mqtt_client_unsubscribe(plugin->client, config->write_req_topic);
neu_mqtt_client_unsubscribe(plugin->client, plugin->driver_cmd_topic);
neu_msleep(100); // wait for message completion
return 0;
}
src/adapter/adapter.c
+12
View File
@@ -443,6 +443,11 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
strcpy(pheader->sender, adapter->name);
switch (pheader->type) {
case NEU_REQ_DRIVER_CMD: {
neu_req_driver_cmd_t *cmd = (neu_req_driver_cmd_t *) data;
strcpy(pheader->receiver, cmd->driver);
break;
}
case NEU_REQ_READ_GROUP: {
neu_req_read_group_t *cmd = (neu_req_read_group_t *) data;
strcpy(pheader->receiver, cmd->driver);
@@ -1399,6 +1404,13 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_msg_free(msg);
break;
}
case NEU_REQ_DRIVER_CMD: {
neu_req_driver_cmd_t *cmd = (neu_req_driver_cmd_t *) &header[1];
neu_adapter_driver_cmd((neu_adapter_driver_t *) adapter,
(const char *) cmd->cmd);
free(cmd->cmd);
break;
}
default:
nlog_warn("adapter: %s recv msg type error, type: %s", adapter->name,
neu_reqresp_type_string(header->type));
src/adapter/driver/driver.c
+12
View File
@@ -3021,4 +3021,16 @@ void neu_adapter_driver_test_read_tag(neu_adapter_driver_t *driver,
if (g != NULL && neu_group_tag_size(g->group) > 0) {
start_group_timer(driver, g);
}
}
int neu_adapter_driver_cmd(neu_adapter_driver_t *driver, const char *cmd)
{
if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) {
return NEU_ERR_PLUGIN_NOT_RUNNING;
}
if (driver->adapter.module->intf_funs->driver.call == NULL) {
return NEU_ERR_PLUGIN_NOT_SUPPORT_CMD_CALL;
}
return driver->adapter.module->intf_funs->driver.call(
driver->adapter.plugin, cmd);
}
src/adapter/driver/driver_internal.h
+3
View File
@@ -94,4 +94,7 @@ void neu_adapter_driver_unsubscribe(neu_adapter_driver_t * driver,
void neu_adapter_driver_scan_tags(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req);
int neu_adapter_driver_cmd(neu_adapter_driver_t *driver, const char *cmd);
#endif
src/base/msg_internal.h
+2 -1
View File
@@ -101,7 +101,8 @@ extern "C" {
XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t) \
XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t) \
XX(NEU_REQ_CHECK_SCHEMA, neu_req_check_schema_t) \
XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t)
XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t) \
XX(NEU_REQ_DRIVER_CMD, neu_req_driver_cmd_t)
static inline size_t neu_reqresp_size(neu_reqresp_type_e t)
{
src/core/manager.c
+1
View File
@@ -1062,6 +1062,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_WRITE_GTAGS:
case NEU_REQ_GET_TAG:
case NEU_REQ_NODE_CTL:
case NEU_REQ_DRIVER_CMD:
case NEU_REQ_ADD_GROUP: {
neu_otel_trace_ctx trace = NULL;
neu_otel_scope_ctx scope = NULL;
src/parser/neu_json_driver.c
+90
View File
@@ -0,0 +1,90 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 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 <string.h>
#include <jansson.h>
#include "json/json.h"
#include "json/neu_json_driver.h"
int neu_json_decode_driver_cmd_req_json(void * json_obj,
neu_json_driver_cmd_t *req)
{
int ret = 0;
neu_json_elem_t req_elems[] = {
{
.name = "driver",
.t = NEU_JSON_STR,
},
{
.name = "cmd",
.t = NEU_JSON_STR,
},
};
ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
req_elems);
if (ret != 0) {
return -1;
}
req->driver = req_elems[0].v.val_str;
req->cmd = req_elems[1].v.val_str;
return 0;
}
int neu_json_decode_driver_cmd_req(char *buf, neu_json_driver_cmd_t **result)
{
int ret = 0;
void * json_obj = NULL;
neu_json_driver_cmd_t *req = calloc(1, sizeof(neu_json_driver_cmd_t));
if (req == NULL) {
return -1;
}
json_obj = neu_json_decode_new(buf);
if (NULL == json_obj) {
free(req);
return -1;
}
ret = neu_json_decode_driver_cmd_req_json(json_obj, req);
if (0 == ret) {
*result = req;
} else {
free(req);
}
neu_json_decode_free(json_obj);
return ret;
return 0;
}
void neu_json_decode_driver_cmd_req_free(neu_json_driver_cmd_t *req)
{
if (req) {
free(req->driver);
free(req->cmd);
free(req);
}
}
src/utils/json.c
+2 -1
View File
@@ -319,7 +319,7 @@ static int decode_object(json_t *root, neu_json_elem_t *ele)
case NEU_JSON_INT:
ele->v.val_int = json_integer_value(ob);
break;
case NEU_JSON_STR:
case NEU_JSON_STR: {
if (!json_is_string(ob)) {
zlog_error(neuron, "json decode: %s failed", ele->name);
return -1;
@@ -331,6 +331,7 @@ static int decode_object(json_t *root, neu_json_elem_t *ele)
}
ele->v.val_str = strdup(str_val);
break;
}
case NEU_JSON_FLOAT:
if (json_is_integer(ob)) {
ele->v.val_float = (float) json_integer_value(ob);