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

Merge pull request #2200 from hxy7yx/mqtt5

Browse Source 
mqtt supports otel
This commit is contained in:
fengzeroz
2024-09-06 10:07:54 +08:00
committed by GitHub
parent 2eb8801db2 ef2099414e
commit bcbdeb3433
4 changed files with 13 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/json/neu_json_mqtt.h
+1
View File
@@ -35,6 +35,7 @@ typedef struct {
char *uuid;
char *traceparent;
char *tracestate;
char *payload;
} neu_json_mqtt_t;
typedef struct {
plugins/mqtt/mqtt_config.c
+1 -1
View File
@@ -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 };
plugins/mqtt/mqtt_handle.c
+11 -10
View File
@@ -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);
src/connection/mqtt_client.c
-29
View File
@@ -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);