Merge pull request #2340 from hxy7yx/main-1

mqtt: add logs for ecp-format

include/neuron/json/neu_json_fn.h

@@ -36,6 +36,9 @@ int neu_json_encode_by_fn(void *param, neu_json_encode_fn fn, char **result);
 int neu_json_encode_with_mqtt(void *param, neu_json_encode_fn fn,
                               void *mqtt_param, neu_json_encode_fn mqtt_fn,
                               char **result);
+int neu_json_encode_with_mqtt_ecp(void *param, neu_json_encode_fn fn,
+                                  void *mqtt_param, neu_json_encode_fn mqtt_fn,
+                                  char **result);
 
 #define NEU_JSON_RESPONSE_ERROR(err, func)                             \
     {                                                                  \

plugins/mqtt/mqtt.json

@@ -79,7 +79,7 @@
 		}
 	},
 	"upload_err": {
-		"name": "Upload Tag Error Code ",
+		"name": "Upload Tag Error Code",
 		"name_zh": "上报点位错误码",
 		"description": "When data tag collection reports an error, report the tag error code.",
 		"description_zh": "点位采集报错时，上报点位错误码。",

plugins/mqtt/mqtt_handle.c

@@ -108,6 +108,8 @@ char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
         return NULL;
     }
 
+    int ret;
+
     switch (format) {
     case MQTT_UPLOAD_FORMAT_VALUES:
         neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp1, &header,
@@ -120,9 +122,14 @@ char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
                                   &json_str);
         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);
+        ret = neu_json_encode_with_mqtt_ecp(
+            &json, neu_json_encode_read_resp_ecp, &header,
+            neu_json_encode_read_periodic_resp, &json_str);
+        if (ret == -2) {
+            *skip = true;
+            plog_warn(plugin, "driver:%s group:%s, no valid tags", data->driver,
+                      data->group);
+        }
         break;
     default:
         plog_warn(plugin, "invalid upload format: %d", format);

src/parser/neu_json_fn.c

@@ -71,6 +71,36 @@ int neu_json_encode_with_mqtt(void *param, neu_json_encode_fn fn,
     return ret;
 }
 
+int neu_json_encode_with_mqtt_ecp(void *param, neu_json_encode_fn fn,
+                                  void *mqtt_param, neu_json_encode_fn mqtt_fn,
+                                  char **result)
+{
+    void *object = neu_json_encode_new();
+
+    if (mqtt_fn != NULL) {
+        if (mqtt_fn(object, mqtt_param) != 0) {
+            neu_json_encode_free(object);
+            return -1;
+        }
+    }
+
+    if (fn != NULL) {
+        int encode_ret = fn(object, param);
+        if (encode_ret == -2) {
+            neu_json_encode_free(object);
+            return encode_ret;
+        }
+        if (encode_ret != 0) {
+            neu_json_encode_free(object);
+            return -1;
+        }
+    }
+
+    int ret = neu_json_encode(object, result);
+    neu_json_decode_free(object);
+    return ret;
+}
+
 int neu_parse_param(const char *buf, char **err_param, int n,
                     neu_json_elem_t *ele, ...)
 {

src/parser/neu_json_rw.c

@@ -347,8 +347,9 @@ int neu_json_encode_read_resp2(void *json_object, void *param)
 
 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;
+    int                   ret      = 0;
+    int                   has_data = 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;
@@ -380,6 +381,7 @@ int neu_json_encode_read_resp_ecp(void *json_object, void *param)
 
         tag_array =
             neu_json_encode_array(tag_array, tag_elems, 3 + p_tag->n_meta);
+        has_data = 1;
         p_tag++;
     }
 
@@ -391,6 +393,10 @@ int neu_json_encode_read_resp_ecp(void *json_object, void *param)
     ret = neu_json_encode_field(json_object, resp_elems,
                                 NEU_JSON_ELEM_SIZE(resp_elems));
 
+    if (!has_data) {
+        return -2;
+    }
+
     return ret;
 }