Describe the bug
I took inspiration from demo/quic_mqttv5 for a very simple MQTTv5 over QUIC client, which connects to the broker, and subscribe on connection.
#include "nng/mqtt/mqtt_client.h"
#include "nng/mqtt/mqtt_quic.h"
#include <nng/nng.h>
#include <nng/protocol/pubsub0/pub.h>
#include <nng/protocol/pubsub0/sub.h>
#include <nng/supplemental/util/platform.h>
#include <nng/transport/ipc/ipc.h>
#include <stdio.h>
#include <unistd.h>
// NOLINTBEGIN
nng_msg *
AllocateMessage()
{
nng_msg *message = NULL;
nng_mqtt_msg_alloc(&message, 0);
return message;
}
int
SendMessage(nng_msg *message, nng_socket *socket)
{
const int outcome = nng_sendmsg(*socket, message, NNG_FLAG_ALLOC);
if (outcome == 0) {
return 0;
}
nng_msg_free(message);
fprintf(stderr, "Failed to send MQTT message: %s\n",
nng_strerror(outcome));
return 1;
}
nng_msg *
BuildConnectionMessage()
{
nng_msg *const connection_message = AllocateMessage();
nng_mqtt_msg_set_connect_proto_version(
connection_message, MQTT_PROTOCOL_VERSION_v5);
nng_mqtt_msg_set_packet_type(connection_message, NNG_MQTT_CONNECT);
nng_mqtt_msg_set_connect_clean_session(connection_message, true);
nng_mqtt_msg_set_connect_keep_alive(connection_message, 20);
property *properties = mqtt_property_alloc();
mqtt_property_append(properties,
mqtt_property_set_value_str(ASSIGNED_CLIENT_IDENTIFIER, "client",
strlen("client"), false));
mqtt_property_append(properties,
mqtt_property_set_value_u32(SESSION_EXPIRY_INTERVAL, 0));
mqtt_property_append(
properties, mqtt_property_set_value_u16(TOPIC_ALIAS_MAXIMUM, 0));
nng_mqtt_msg_set_connect_property(connection_message, properties);
return connection_message;
}
#define TOPIC "test_topic"
void
Subscribe(nng_socket *socket)
{
printf("Subscribing\n");
nng_mqtt_topic_qos topic_qos = {
.topic =
{
.length = strlen(TOPIC),
.buf = (uint8_t *)TOPIC,
},
.qos = 0,
.nolocal = 0,
.rap = 0,
.retain_handling = 0,
};
nng_msg *const subscribe_message = AllocateMessage();
nng_mqtt_msg_set_packet_type(subscribe_message, NNG_MQTT_SUBSCRIBE);
nng_mqtt_msg_set_subscribe_topics(subscribe_message, &topic_qos, 1);
SendMessage(subscribe_message, socket);
}
int
OnConnect(void *rmsg, void *arg)
{
printf("Connected to MQTT broker! %p\n", rmsg);
nng_socket *socket = (nng_socket *) arg;
Subscribe(socket);
return 0;
}
int
OnDisconnect(void *rmsg, void *arg)
{
printf("Disconnected from MQTT broker! %p\n", rmsg);
return 0;
}
int
OnReceiveMessage(void *rmsg, void *arg)
{
nng_msg *message = (nng_msg *) rmsg;
printf("Received a message! %p\n", (void *) message);
return 0;
}
int
OnSendMessage(void *rmsg, void *arg)
{
nng_msg *message = (nng_msg *) rmsg;
printf("Message sent successfully! %p\n", (void *) message);
return 0;
}
int
main(int argc, char **argv)
{
nng_socket socket;
const char *url = "mqtt-quic://127.0.0.1:14567";
conf_quic config_user = {
.tls =
{
.enable = false,
.url = "",
.cafile = "",
.certfile = "",
.keyfile = "",
.ca = "",
.cert = "",
.key = "",
.key_password = "",
.verify_peer = true,
.set_fail = true,
},
.qos_first = false, // We might consider to bump it to true when using QoS > 0
.multi_stream = false,
.qkeepalive = 10, // QUIC keepalive (seconds)
.qconnect_timeout = 30, // QUIC handshake idle timeout (seconds)
.qdiscon_timeout = 1, // QUIC disconnect timeout (seconds)
.qidle_timeout = 30, // QUIC disconnect after idle (seconds)
.qcongestion_control = 1 // 0: cubic, 1: bbr
};
int outcome;
outcome = nng_mqttv5_quic_client_open_conf(&socket, url, &config_user);
if (outcome != 0) {
fprintf(stderr, "Failed to open MQTT client socket: %s\n",
nng_strerror(outcome));
return 1;
}
outcome = nng_mqtt_quic_set_connect_cb(&socket, OnConnect, &socket);
if (outcome != 0) {
fprintf(stderr,
"Failed to set MQTT client connect callback: %s\n",
nng_strerror(outcome));
}
outcome =
nng_mqtt_quic_set_disconnect_cb(&socket, OnDisconnect, &socket);
if (outcome != 0) {
fprintf(stderr,
"Failed to set MQTT client disconnect callback: %s\n",
nng_strerror(outcome));
}
outcome =
nng_mqtt_quic_set_msg_recv_cb(&socket, OnReceiveMessage, &socket);
if (outcome != 0) {
fprintf(stderr,
"Failed to set MQTT client message received callback: "
"%s\n",
nng_strerror(outcome));
}
outcome =
nng_mqtt_quic_set_msg_send_cb(&socket, OnSendMessage, &socket);
if (outcome != 0) {
fprintf(stderr,
"Failed to set MQTT client message send callback: "
"%s\n",
nng_strerror(outcome));
}
nng_msg *const connection_message = BuildConnectionMessage();
SendMessage(connection_message, &socket);
nng_msleep(5000);
printf("Closing socket\n");
nng_close(socket);
printf("Done\n");
}
// NOLINTENDExpected behavior
The program:
- Connects to the broker
- Subscribes to
test_topic
- Gently quits
Actual Behavior
The program:
- Connects to the broker
- Subscribe to
test_topic
- Sometimes holds infinitely on
Closing socket
To Reproduce
while true do; ./build/demo/dummy_quic/dummy_quic; done
After few iterations, the program will block on nng_close.
Output:
Message sent successfully! (nil)
Connected to MQTT broker! 0x73d06c000b70
Subscribing
Message sent successfully! (nil)
Closing socket
Environment Details
- NanoSDK/NNG version: using NanoSDK main
6cb73c84
- Operating system and version: Ubuntu 24
- Compiler and language used: clang-20
- Shared or static library: Static
- Remote MQTT Broker info: EMQX 5.8.8
Additional context
Add any other context about the problem here.
Describe the bug
I took inspiration from
demo/quic_mqttv5for a very simple MQTTv5 over QUIC client, which connects to the broker, and subscribe on connection.Expected behavior
The program:
test_topicActual Behavior
The program:
test_topicClosing socketTo Reproduce
After few iterations, the program will block on
nng_close.Output:
Environment Details
6cb73c84Additional context
Add any other context about the problem here.