diff --git a/include/neuron/json/neu_json_mqtt.h b/include/neuron/json/neu_json_mqtt.h index fb850b51..a156cdf4 100644 --- a/include/neuron/json/neu_json_mqtt.h +++ b/include/neuron/json/neu_json_mqtt.h @@ -35,6 +35,7 @@ typedef struct { char *uuid; char *traceparent; char *tracestate; + char *payload; } neu_json_mqtt_t; typedef struct { diff --git a/plugins/mqtt/mqtt_config.c b/plugins/mqtt/mqtt_config.c index 6a9ced31..a8156bdb 100644 --- a/plugins/mqtt/mqtt_config.c +++ b/plugins/mqtt/mqtt_config.c @@ -221,7 +221,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting, neu_json_elem_t version = { .name = "version", .t = NEU_JSON_INT, - .v.val_int = NEU_MQTT_VERSION_V5, // default to V5 + .v.val_int = NEU_MQTT_VERSION_V311, // default to V311 .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility }; neu_json_elem_t client_id = { .name = "client-id", .t = NEU_JSON_STR }; diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 15bf06f0..1f94f43d 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -141,6 +141,7 @@ static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_reqresp_head_t header = { 0 }; header.ctx = mqtt; header.type = NEU_REQ_READ_GROUP; + // header.otel_trace_type = NEU_OTEL_TRACE_TYPE_MQTT; neu_req_read_group_t cmd = { 0 }; cmd.driver = req->node; cmd.group = req->group; @@ -202,8 +203,9 @@ static int send_write_tag_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_reqresp_head_t header = { 0 }; neu_req_write_tag_t cmd = { 0 }; - header.ctx = mqtt; - header.type = NEU_REQ_WRITE_TAG; + header.ctx = mqtt; + header.type = NEU_REQ_WRITE_TAG; + header.otel_trace_type = NEU_OTEL_TRACE_TYPE_MQTT; cmd.driver = req->node; cmd.group = req->group; @@ -245,8 +247,9 @@ static int send_write_tags_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, } neu_reqresp_head_t header = { - .ctx = mqtt, - .type = NEU_REQ_WRITE_TAGS, + .ctx = mqtt, + .type = NEU_REQ_WRITE_TAGS, + .otel_trace_type = NEU_OTEL_TRACE_TYPE_MQTT, }; neu_req_write_tags_t cmd = { 0 }; @@ -358,9 +361,8 @@ void handle_write_req(neu_mqtt_qos_e qos, const char *topic, if (trace_w3c && trace_w3c->traceparent) { mqtt->traceparent = strdup(trace_w3c->traceparent); - } - if (trace_w3c && trace_w3c->tracestate) { - mqtt->tracestate = strdup(trace_w3c->tracestate); + mqtt->tracestate = strdup(trace_w3c->tracestate); + mqtt->payload = json_str; } rv = neu_json_decode_write(json_str, &req); @@ -457,9 +459,8 @@ void handle_read_req(neu_mqtt_qos_e qos, const char *topic, if (trace_w3c && trace_w3c->traceparent) { mqtt->traceparent = strdup(trace_w3c->traceparent); - } - if (trace_w3c && trace_w3c->tracestate) { - mqtt->tracestate = strdup(trace_w3c->tracestate); + mqtt->tracestate = strdup(trace_w3c->tracestate); + mqtt->payload = json_str; } rv = neu_json_decode_read_req(json_str, &req); diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c index 1d94a914..a90e1b0e 100644 --- a/src/connection/mqtt_client.c +++ b/src/connection/mqtt_client.c @@ -814,23 +814,6 @@ static int client_send_sub_msg(neu_mqtt_client_t *client, nng_mqtt_msg_set_packet_type(sub_msg, NNG_MQTT_SUBSCRIBE); nng_mqtt_msg_set_subscribe_topics(sub_msg, topic_qos, topic_qos_count); - /*if (client->version == MQTT_PROTOCOL_VERSION_v5) { - property *plist = mqtt_property_alloc(); - property *p1 = mqtt_property_set_value_strpair(USER_PROPERTY, - "traceparent", strlen("traceparent"), "abcde", strlen("abcde"), true); - mqtt_property_append(plist, p1); - property *p2 = mqtt_property_set_value_strpair(USER_PROPERTY, - "tracestate", strlen("tracestate"), "abcde", strlen("abcde"), true); - mqtt_property_append(plist, p2); - - nng_mqtt_msg_set_subscribe_property(sub_msg, plist); - }*/ - - /*nng_mqttv5_msg_encode(sub_msg); - uint8_t print[1024] = {0}; - nng_mqtt_msg_dump(sub_msg, print, 1024, true); - printf("%s\n", print);*/ - task_t *task = client_alloc_task(client); if (NULL == task) { log(error, "client_alloc_task fail"); @@ -1595,18 +1578,6 @@ int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos, nng_mqtt_msg_set_publish_payload(pub_msg, (uint8_t *) payload, len); nng_mqtt_msg_set_publish_qos(pub_msg, qos); - /*if (client->version == MQTT_PROTOCOL_VERSION_v5) { - property *plist = mqtt_property_alloc(); - property *p1 = mqtt_property_set_value_strpair(USER_PROPERTY, - "traceparent", strlen("traceparent"), "abcde", strlen("abcde"), true); - mqtt_property_append(plist, p1); - property *p2 = mqtt_property_set_value_strpair(USER_PROPERTY, - "tracestate", strlen("tracestate"), "12345", strlen("12345"), true); - mqtt_property_append(plist, p2); - - nng_mqtt_msg_set_publish_property(pub_msg, plist); - }*/ - nng_mtx_lock(client->mtx); task = client_alloc_task(client); nng_mtx_unlock(client->mtx);