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

Merge pull request #2338 from hxy7yx/main-1

Browse Source 
feat(mqtt):filter tags with errors when uploading
This commit is contained in:
fengzeroz
2024-12-16 15:08:28 +08:00
committed by GitHub
parent 2499657f8d a087f53d04
commit cab2f92e70
11 changed files with 199 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/json/json.h
+10
View File
@@ -28,6 +28,14 @@ extern "C" {
#include <stdint.h>
#include <stdlib.h>
typedef enum neu_json_type_ecp {
NEU_JSON_ECP_UNDEFINE = 0,
NEU_JSON_ECP_BOOL = 1,
NEU_JSON_ECP_INT = 2,
NEU_JSON_ECP_FLOAT = 3,
NEU_JSON_ECP_STRING = 4,
} neu_json_type_ecp_e;
typedef enum neu_json_type {
NEU_JSON_UNDEFINE = 0,
NEU_JSON_INT = 1,
@@ -157,6 +165,8 @@ void neu_json_elem_free(neu_json_elem_t *elem);
/* New a empty josn array */
void *neu_json_array();
int neu_json_type_transfer(neu_json_type_e type);
int neu_json_decode_by_json(void *json, int size, neu_json_elem_t *ele);
int neu_json_decode(char *buf, int size, neu_json_elem_t *ele);
int neu_json_decode_array_size_by_json(void *json, char *child);
include/neuron/json/neu_json_rw.h
+3 -2
View File
@@ -62,8 +62,9 @@ typedef struct {
} neu_json_read_resp_t;
int neu_json_encode_read_resp(void *json_object, void *param);
int neu_json_encode_read_resp1(void *json_object, void *param);
int neu_json_encode_read_resp2(void *json_object, void *param);
int neu_json_encode_read_resp1(void *json_object, void *param); // values
int neu_json_encode_read_resp2(void *json_object, void *param); // tags
int neu_json_encode_read_resp_ecp(void *json_object, void *param);
int neu_json_encode_read_paginate_resp(void *json_object, void *param);
typedef struct {
plugins/mqtt/azure_iot_plugin.c
+1 -1
View File
@@ -300,7 +300,7 @@ static int azure_handle_trans_data(neu_plugin_t * plugin,
}
char *json_str =
generate_upload_json(plugin, trans_data, plugin->config.format);
generate_upload_json(plugin, trans_data, plugin->config.format, NULL);
if (NULL == json_str) {
plog_error(plugin, "generate upload json fail");
return NEU_ERR_EINTERNAL;
plugins/mqtt/mqtt.json
+14
View File
@@ -70,10 +70,24 @@
{
"key": "tags-format",
"value": 1
},
{
"key": "ECP-format",
"value": 2
}
]
}
},
"upload_err": {
"name": "Upload Tag Error Code ",
"name_zh": "上报点位错误码",
"description": "When data tag collection reports an error, report the tag error code.",
"description_zh": "点位采集报错时，上报点位错误码。",
"attribute": "optional",
"type": "bool",
"default": true,
"valid": {}
},
"write-req-topic": {
"name": "Write Request Topic",
"name_zh": "写请求主题",
plugins/mqtt/mqtt_config.c
+15 -1
View File
@@ -315,6 +315,12 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
neu_json_elem_t upload_drv_state_interval = {
.name = "upload_drv_state_interval", .t = NEU_JSON_INT
};
neu_json_elem_t upload_err = {
.name = "upload_err",
.t = NEU_JSON_BOOL,
.v.val_bool = true,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
};
if (NULL == setting || NULL == config) {
plog_error(plugin, "invalid argument, null pointer");
@@ -348,7 +354,8 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
// format, required
if (MQTT_UPLOAD_FORMAT_VALUES != format.v.val_int &&
MQTT_UPLOAD_FORMAT_TAGS != format.v.val_int) {
MQTT_UPLOAD_FORMAT_TAGS != format.v.val_int &&
MQTT_UPLOAD_FORMAT_ECP != format.v.val_int) {
plog_error(plugin, "setting invalid format: %" PRIi64,
format.v.val_int);
goto error;
@@ -435,6 +442,11 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
goto error;
}
ret = neu_parse_param(setting, NULL, 1, &upload_err);
if (0 != ret) {
plog_notice(plugin, "setting upload_err failed");
}
config->version = version.v.val_int;
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
@@ -459,6 +471,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
config->upload_drv_state = upload_drv_state.v.val_bool;
config->heartbeat_topic = upload_drv_state_topic.v.val_str;
config->heartbeat_interval = upload_drv_state_interval.v.val_int;
config->upload_err = upload_err.v.val_bool;
plog_notice(plugin, "config MQTT version : %d", config->version);
plog_notice(plugin, "config client-id : %s", config->client_id);
@@ -474,6 +487,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
config->driver_action_resp_topic);
plog_notice(plugin, "config upload-drv-state: %d",
config->upload_drv_state);
plog_notice(plugin, "config upload-err: %d", config->upload_err);
if (config->upload_drv_state) {
if (config->heartbeat_topic) {
plog_notice(plugin, "config upload-drv-state-topic: %s",
plugins/mqtt/mqtt_config.h
+4
View File
@@ -33,6 +33,7 @@ extern "C" {
typedef enum {
MQTT_UPLOAD_FORMAT_VALUES = 0,
MQTT_UPLOAD_FORMAT_TAGS = 1,
MQTT_UPLOAD_FORMAT_ECP = 2,
} mqtt_upload_format_e;
static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
@@ -42,6 +43,8 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
return "format-values";
case MQTT_UPLOAD_FORMAT_TAGS:
return "format-tags";
case MQTT_UPLOAD_FORMAT_ECP:
return "ECP-format";
default:
return NULL;
}
@@ -56,6 +59,7 @@ typedef struct {
char * write_resp_topic; // write response topic
char * driver_action_req_topic; // driver action request topic
char * driver_action_resp_topic; // driver action response topic
bool upload_err; // Upload tag error code flag
bool upload_drv_state; // upload driver state flag
char * heartbeat_topic; // upload driver state topic
uint16_t heartbeat_interval; // upload driver state interval
plugins/mqtt/mqtt_handle.c
+49 -6
View File
@@ -64,8 +64,29 @@ static int tag_values_to_json(UT_array *tags, neu_json_read_resp_t *json)
return 0;
}
void filter_error_tags(neu_reqresp_trans_data_t *data)
{
if (!data || !data->tags) {
return;
}
UT_array *filtered_tags;
utarray_new(filtered_tags, neu_resp_tag_value_meta_icd());
neu_resp_tag_value_meta_t *tag_ptr = NULL;
while ((tag_ptr = (neu_resp_tag_value_meta_t *) utarray_next(data->tags,
tag_ptr))) {
if (tag_ptr->value.type != NEU_TYPE_ERROR) {
utarray_push_back(filtered_tags, tag_ptr);
}
}
utarray_free(data->tags);
data->tags = filtered_tags;
}
char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
mqtt_upload_format_e format)
mqtt_upload_format_e format, bool *skip)
{
char * json_str = NULL;
neu_json_read_periodic_t header = { .group = (char *) data->group,
@@ -73,21 +94,39 @@ char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
.timestamp = global_timestamp };
neu_json_read_resp_t json = { 0 };
if (!plugin->config.upload_err && skip != NULL) {
filter_error_tags(data);
if (utarray_len(data->tags) == 0) {
*skip = true;
return NULL;
}
}
if (0 != tag_values_to_json(data->tags, &json)) {
plog_error(plugin, "tag_values_to_json fail");
return NULL;
}
if (MQTT_UPLOAD_FORMAT_VALUES == format) { // values
switch (format) {
case MQTT_UPLOAD_FORMAT_VALUES:
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp1, &header,
neu_json_encode_read_periodic_resp,
&json_str);
} else if (MQTT_UPLOAD_FORMAT_TAGS == format) { // tags
break;
case MQTT_UPLOAD_FORMAT_TAGS:
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp2, &header,
neu_json_encode_read_periodic_resp,
&json_str);
} else {
break;
case MQTT_UPLOAD_FORMAT_ECP:
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp_ecp, &header,
neu_json_encode_read_periodic_resp,
&json_str);
break;
default:
plog_warn(plugin, "invalid upload format: %d", format);
break;
}
for (int i = 0; i < json.n_tag; i++) {
@@ -802,8 +841,12 @@ int handle_trans_data(neu_plugin_t * plugin,
break;
}
char *json_str =
generate_upload_json(plugin, trans_data, plugin->config.format);
bool skip_none = false;
char *json_str = generate_upload_json(
plugin, trans_data, plugin->config.format, &skip_none);
if (skip_none) {
break;
}
if (NULL == json_str) {
plog_error(plugin, "generate upload json fail");
rv = NEU_ERR_EINTERNAL;
plugins/mqtt/mqtt_handle.h
+1 -1
View File
@@ -58,7 +58,7 @@ int handle_driver_action_response(neu_plugin_t * plugin,
neu_resp_driver_action_t *data);
char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
mqtt_upload_format_e format);
mqtt_upload_format_e format, bool *skip);
int handle_trans_data(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *trans_data);
src/parser/neu_json_rw.c
+49
View File
@@ -345,6 +345,55 @@ int neu_json_encode_read_resp2(void *json_object, void *param)
return ret;
}
int neu_json_encode_read_resp_ecp(void *json_object, void *param)
{
int ret = 0;
neu_json_read_resp_t *resp = (neu_json_read_resp_t *) param;
void * tag_array = neu_json_array();
neu_json_read_resp_tag_t *p_tag = resp->tags;
for (int i = 0; i < resp->n_tag; i++) {
neu_json_elem_t tag_elems[2 + NEU_TAG_META_SIZE] = { 0 };
tag_elems[0].name = "name";
tag_elems[0].t = NEU_JSON_STR;
tag_elems[0].v.val_str = p_tag->name;
if (p_tag->error != 0) {
continue;
} else {
tag_elems[1].name = "value";
tag_elems[1].t = p_tag->t;
tag_elems[1].v = p_tag->value;
tag_elems[1].precision = p_tag->precision;
tag_elems[2].name = "type";
tag_elems[2].t = NEU_JSON_INT;
tag_elems[2].v.val_int = neu_json_type_transfer(p_tag->t);
}
for (int k = 0; k < p_tag->n_meta; k++) {
tag_elems[3 + k].name = p_tag->metas[k].name;
tag_elems[3 + k].t = p_tag->metas[k].t;
tag_elems[3 + k].v = p_tag->metas[k].value;
}
tag_array =
neu_json_encode_array(tag_array, tag_elems, 3 + p_tag->n_meta);
p_tag++;
}
neu_json_elem_t resp_elems[] = { {
.name = "tags",
.t = NEU_JSON_OBJECT,
.v.val_object = tag_array,
} };
ret = neu_json_encode_field(json_object, resp_elems,
NEU_JSON_ELEM_SIZE(resp_elems));
return ret;
}
int neu_json_decode_write_req(char *buf, neu_json_write_req_t **result)
{
void *json_obj = neu_json_decode_new(buf);
src/utils/json.c
+20
View File
@@ -155,6 +155,26 @@ static json_t *encode_object_value(neu_json_elem_t *ele)
return ob;
}
int neu_json_type_transfer(neu_json_type_e type)
{
switch (type) {
case NEU_JSON_BOOL:
return NEU_JSON_ECP_BOOL;
case NEU_JSON_INT:
case NEU_JSON_BIT:
return NEU_JSON_ECP_INT;
case NEU_JSON_FLOAT:
case NEU_JSON_DOUBLE:
return NEU_JSON_ECP_FLOAT;
case NEU_JSON_STR:
return NEU_JSON_ECP_STRING;
default:
break;
}
return NEU_JSON_ECP_STRING;
}
static json_t *encode_object(json_t *object, neu_json_elem_t ele)
{
json_t *ob = object;
tests/ft/app/test_mqtt.py
+33 -1
View File
@@ -24,7 +24,13 @@ GROUP = "group"
INTERVAL = 100
TAG0 = {
"name": "tag0",
"address": "1!400001.",
"address": "1!400001",
"attribute": config.NEU_TAG_ATTRIBUTE_RW,
"type": config.NEU_TYPE_INT16,
}
TAG_ERR = {
"name": "tag_err",
"address": "100!400001",
"attribute": config.NEU_TAG_ATTRIBUTE_RW,
"type": config.NEU_TYPE_INT16,
}
@@ -203,6 +209,14 @@ def conf_upload_driver_state(conf_base):
}
@pytest.fixture(scope="function")
def conf_upload_err(conf_base):
return {
**conf_base,
"upload_err": False,
}
@pytest.fixture(scope="function")
def conf_bad_host(conf_base):
return {
@@ -462,6 +476,24 @@ class TestMQTT:
finally:
api.del_tags(DRIVER, GROUP, [tag["name"] for tag in TAGS])
@description(
given="MQTT node",
when="read an error tag and set upload_err false",
then="broker should not receive error data on upload topic",
)
def test_mqtt_upload_err(self, mocker, conf_upload_err):
api.node_setting_check(NODE, conf_upload_err)
api.node_ctl(NODE, config.NEU_CTL_START)
try:
api.add_tags_check(DRIVER, GROUP, tags=[TAG_ERR])
msg = mocker.get(UPLOAD_TOPIC, timeout=3)
assert msg is not None
assert 0 == len(msg["errors"])
finally:
api.del_tags(DRIVER, GROUP, [TAG_ERR["name"]])
@description(
given="MQTT node",
when="after renaming subscribed driver",