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

Merge pull request #2306 from fengzeroz/main

Browse Source 
mqtt filter tag & driver cmd
This commit is contained in:
fengzeroz
2024-11-11 11:40:51 +08:00
committed by GitHub
parent 141b3f56b1 b291e49611
commit d60e9293fb
20 changed files with 348 additions and 27 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/json.h
+7
View File
@@ -48,6 +48,7 @@ typedef enum neu_json_type {
NEU_JSON_ARRAY_FLOAT,
NEU_JSON_ARRAY_DOUBLE,
NEU_JSON_ARRAY_BOOL,
NEU_JSON_ARRAY_STR,
NEU_JSON_VALUE = NEU_JSON_UNDEFINE
} neu_json_type_e;
@@ -106,6 +107,11 @@ typedef struct {
uint8_t length;
} neu_json_value_array_bool_t;
typedef struct {
char ** p_strs;
uint16_t length;
} neu_json_value_array_str_t;
typedef union neu_json_value {
int64_t val_int;
uint8_t val_bit;
@@ -125,6 +131,7 @@ typedef union neu_json_value {
neu_json_value_array_float_t val_array_float;
neu_json_value_array_double_t val_array_double;
neu_json_value_array_bool_t val_array_bool;
neu_json_value_array_str_t val_array_str;
} neu_json_value_u;
typedef enum neu_json_attribute {
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/json/neu_json_rw.h
+7 -5
View File
@@ -149,11 +149,13 @@ int neu_json_decode_write(char *buf, neu_json_write_t **result);
void neu_json_decode_write_free(neu_json_write_t *req);
typedef struct {
char *group;
char *node;
char *name;
char *desc;
bool sync;
char * group;
char * node;
char * name;
char * desc;
bool sync;
int n_tags;
char **tags;
} neu_json_read_req_t;
int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result);
include/neuron/msg.h
+22 -5
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;
@@ -637,11 +646,13 @@ typedef struct {
} neu_resp_get_nodes_state_t, neu_reqresp_nodes_state_t;
typedef struct neu_req_read_group {
char *driver;
char *group;
char *name;
char *desc;
bool sync;
char * driver;
char * group;
char * name;
char * desc;
bool sync;
uint16_t n_tag;
char ** tags;
} neu_req_read_group_t;
static inline void neu_req_read_group_fini(neu_req_read_group_t *req)
@@ -650,6 +661,12 @@ static inline void neu_req_read_group_fini(neu_req_read_group_t *req)
free(req->group);
free(req->name);
free(req->desc);
if (req->n_tag > 0) {
for (uint16_t i = 0; i < req->n_tag; i++) {
free(req->tags[i]);
}
free(req->tags);
}
}
typedef struct neu_req_read_group_paginate {
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
+53 -2
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)
{
@@ -198,8 +219,14 @@ static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
cmd.driver = req->node;
cmd.group = req->group;
cmd.sync = req->sync;
req->node = NULL; // ownership moved
req->group = NULL; // ownership moved
cmd.n_tag = req->n_tags;
cmd.tags = req->tags;
req->node = NULL; // ownership moved
req->group = NULL; // ownership moved
req->n_tags = 0;
req->tags = NULL;
if (0 != neu_plugin_op(plugin, header, &cmd)) {
neu_req_read_group_fini(&cmd);
plog_error(plugin, "neu_plugin_op(NEU_REQ_READ_GROUP) fail");
@@ -527,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
+15 -2
View File
@@ -568,7 +568,8 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
neu_resp_read_group_t resp = { 0 };
neu_group_t * group = g->group;
UT_array *tags = neu_group_query_read_tag(group, cmd->name, cmd->desc);
UT_array *tags = neu_group_query_read_tag(group, cmd->name, cmd->desc,
cmd->n_tag, cmd->tags);
utarray_new(resp.tags, neu_resp_tag_value_meta_icd());
@@ -651,7 +652,7 @@ void neu_adapter_driver_read_group_paginate(neu_adapter_driver_t *driver,
group, cmd->name, cmd->desc, cmd->current_page, cmd->page_size,
&resp.total_count);
} else {
tags = neu_group_query_read_tag(group, cmd->name, cmd->desc);
tags = neu_group_query_read_tag(group, cmd->name, cmd->desc, 0, NULL);
resp.total_count = utarray_len(tags);
}
@@ -3020,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/group.c
+21 -7
View File
@@ -256,15 +256,26 @@ static inline bool description_contains(const neu_datatag_t *tag, void *data)
}
struct query {
char *name;
char *desc;
char * name;
char * desc;
uint16_t n_tagname;
char ** tagnames;
};
static inline bool match_query(const neu_datatag_t *tag, void *data)
{
struct query *q = data;
return (!q->name || name_contains(tag, q->name)) &&
struct query *q = data;
bool filter = (!q->name || name_contains(tag, q->name)) &&
(!q->desc || description_contains(tag, q->desc));
if (filter && q->n_tagname > 0) {
for (uint16_t i = 0; i < q->n_tagname; i++) {
if (strcmp(tag->name, q->tagnames[i]) == 0) {
return true;
}
}
return false;
}
return filter;
}
static inline bool is_readable_and_match_query(const neu_datatag_t *tag,
@@ -285,12 +296,15 @@ UT_array *neu_group_query_tag(neu_group_t *group, const char *name)
}
UT_array *neu_group_query_read_tag(neu_group_t *group, const char *name,
const char *desc)
const char *desc, uint16_t n_tagname,
char **tagnames)
{
UT_array * array = NULL;
struct query q = {
.name = (char *) name,
.desc = (char *) desc,
.name = (char *) name,
.desc = (char *) desc,
.n_tagname = n_tagname,
.tagnames = tagnames,
};
pthread_mutex_lock(&group->mtx);
src/base/group.h
+2 -1
View File
@@ -42,7 +42,8 @@ UT_array * neu_group_get_tag(neu_group_t *group);
UT_array * neu_group_query_tag(neu_group_t *group, const char *name);
UT_array * neu_group_get_read_tag(neu_group_t *group);
UT_array * neu_group_query_read_tag(neu_group_t *group, const char *name,
const char *desc);
const char *desc, uint16_t n_tagname,
char **tagnames);
UT_array * neu_group_query_read_tag_paginate(neu_group_t *group,
const char *name, const char *desc,
int current_page, int page_size,
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/parser/neu_json_rw.c
+20 -3
View File
@@ -670,6 +670,13 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
.t = NEU_JSON_OBJECT,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
{
.name = "tags",
.t = NEU_JSON_ARRAY_STR,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
.v.val_array_str.length = 0,
.v.val_array_str.p_strs = NULL,
},
};
neu_json_elem_t query_elems[] = {
@@ -691,9 +698,11 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
goto error;
}
req->node = req_elems[0].v.val_str;
req->group = req_elems[1].v.val_str;
req->sync = req_elems[2].v.val_bool;
req->node = req_elems[0].v.val_str;
req->group = req_elems[1].v.val_str;
req->sync = req_elems[2].v.val_bool;
req->n_tags = req_elems[4].v.val_array_str.length;
req->tags = req_elems[4].v.val_array_str.p_strs;
if (req_elems[3].v.val_object) {
ret = neu_json_decode_by_json(req_elems[3].v.val_object,
@@ -726,6 +735,14 @@ void neu_json_decode_read_req_free(neu_json_read_req_t *req)
{
free(req->group);
free(req->node);
if (req->n_tags > 0) {
for (int i = 0; i < req->n_tags; i++) {
if (req->tags[i] != NULL) {
free(req->tags[i]);
}
}
free(req->tags);
}
free(req);
}
src/utils/json.c
+26 -1
View File
@@ -296,6 +296,10 @@ static int decode_object(json_t *root, neu_json_elem_t *ele)
ele->t = NEU_JSON_ARRAY_INT64;
json_array_foreach(ob, index, value)
{
if (json_is_string(value)) {
ele->t = NEU_JSON_ARRAY_STR;
break;
}
if (json_is_real(value)) {
ele->t = NEU_JSON_ARRAY_DOUBLE;
break;
@@ -315,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;
@@ -327,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);
@@ -344,6 +349,26 @@ static int decode_object(json_t *root, neu_json_elem_t *ele)
case NEU_JSON_BOOL:
ele->v.val_bool = json_boolean_value(ob);
break;
case NEU_JSON_ARRAY_STR: {
json_t *value = NULL;
ele->v.val_array_str.length = json_array_size(ob);
if (ele->v.val_array_str.length > 0) {
int index = 0;
ele->v.val_array_str.p_strs =
calloc(ele->v.val_array_str.length, sizeof(char *));
json_array_foreach(ob, index, value)
{
const char *str_val = json_string_value(value);
if (str_val != NULL) {
ele->v.val_array_str.p_strs[index] = strdup(str_val);
}
}
}
break;
}
case NEU_JSON_ARRAY_BOOL: {
json_t *value = NULL;