diff --git a/include/neuron/json/json.h b/include/neuron/json/json.h index 96f09091..58004be0 100644 --- a/include/neuron/json/json.h +++ b/include/neuron/json/json.h @@ -28,6 +28,14 @@ extern "C" { #include #include +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); diff --git a/include/neuron/json/neu_json_rw.h b/include/neuron/json/neu_json_rw.h index 187d106b..97d02438 100644 --- a/include/neuron/json/neu_json_rw.h +++ b/include/neuron/json/neu_json_rw.h @@ -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 { diff --git a/plugins/mqtt/azure_iot_plugin.c b/plugins/mqtt/azure_iot_plugin.c index a806ffe2..bd470edc 100644 --- a/plugins/mqtt/azure_iot_plugin.c +++ b/plugins/mqtt/azure_iot_plugin.c @@ -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; diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index 570e2f8a..5bd59bf0 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -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": "写请求主题", diff --git a/plugins/mqtt/mqtt_config.c b/plugins/mqtt/mqtt_config.c index 2654c3dc..2505724a 100644 --- a/plugins/mqtt/mqtt_config.c +++ b/plugins/mqtt/mqtt_config.c @@ -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", diff --git a/plugins/mqtt/mqtt_config.h b/plugins/mqtt/mqtt_config.h index c28ae7cf..1de354b7 100644 --- a/plugins/mqtt/mqtt_config.h +++ b/plugins/mqtt/mqtt_config.h @@ -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 diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 77a9308b..40c0f10c 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -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; diff --git a/plugins/mqtt/mqtt_handle.h b/plugins/mqtt/mqtt_handle.h index fd1dabde..31df7881 100644 --- a/plugins/mqtt/mqtt_handle.h +++ b/plugins/mqtt/mqtt_handle.h @@ -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); diff --git a/src/parser/neu_json_rw.c b/src/parser/neu_json_rw.c index 5cddaf10..4cb8eb20 100644 --- a/src/parser/neu_json_rw.c +++ b/src/parser/neu_json_rw.c @@ -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); diff --git a/src/utils/json.c b/src/utils/json.c index b402ba8b..6f920983 100644 --- a/src/utils/json.c +++ b/src/utils/json.c @@ -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; diff --git a/tests/ft/app/test_mqtt.py b/tests/ft/app/test_mqtt.py index 87e170fb..842eba8e 100644 --- a/tests/ft/app/test_mqtt.py +++ b/tests/ft/app/test_mqtt.py @@ -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",