Merge branch 'main' into jaylin/reconnect_issue

CMakeLists.txt

@@ -306,6 +306,16 @@ if(NNG_ENABLE_QUIC)
   add_definitions(-DSUPP_QUIC)
 endif()
 
+if (NNG_ENABLE_SCRAM)
+    find_package(OpenSSL)
+    if (NNG_ENABLE_QUIC)
+    else ()
+        target_link_libraries(nng PRIVATE OpenSSL::SSL OpenSSL:Crypto)
+    endif (NNG_ENABLE_QUIC)
+
+    add_definitions(-DSUPP_SCRAM)
+endif ()
+
 add_subdirectory(src)
 
 if (NNG_TESTS)

include/nng/mqtt/mqtt_client.h

@@ -71,6 +71,8 @@ extern "C" {
 
 #define NNG_OPT_MQTT_SQLITE "mqtt-sqlite-option"
 
+#define NNG_OPT_MQTT_ENABLE_SCRAM "mqtt-scram-option"
+
 // NNG_OPT_MQTT_QOS is a byte (only lower two bits significant) representing
 // the quality of service.  At this time, only level zero is supported.
 // TODO: level 1 and level 2 QoS

include/nng/supplemental/nanolib/conf.h

@@ -74,8 +74,10 @@ struct conf_tls {
 
 typedef struct conf_tls conf_tls;
 
+// TODO: params for one single tcp node should be in here.
 typedef struct {
 	bool     enable;
+	char    *url;
 	uint8_t  nodelay;
 	uint8_t  keepalive;
 	uint8_t  quickack;
@@ -86,6 +88,16 @@ typedef struct {
 	uint16_t recvtimeo;
 } conf_tcp;
 
+typedef struct {
+	size_t     count;
+	conf_tls **nodes;
+} conf_tls_list;
+
+typedef struct {
+	size_t     count;
+	conf_tcp **nodes;
+} conf_tcp_list;
+
 struct conf_sqlite {
 	bool   enable;
 	size_t disk_cache_size;   // specify the max rows of sqlite table
@@ -548,15 +560,17 @@ typedef enum {
 
 struct conf {
 	char      *vin;
+	char      *cmd_ipc_url;
+	char      *hook_ipc_url;
 	char      *conf_file;
 	char      *url;
 	bool       enable;
 	int        property_size;
 	int        msq_len;
 	uint32_t   num_taskq_thread;
 	uint32_t   max_taskq_thread;
-	uint32_t   parallel;				// broker ctx
-	uint64_t   total_ctx;		// Total ctx of work (bridge + AWS + broker + HTTP)
+	uint32_t   parallel;			   // broker ctx
+	uint64_t   total_ctx;		       // Total ctx of work (bridge + AWS + broker + HTTP)
 	uint64_t   max_packet_size;        // byte
 	uint32_t   client_max_packet_size; // byte
 	uint32_t   max_inflight_window;
@@ -570,6 +584,8 @@ struct conf {
 	bool       ipc_internal;
 	bool       bridge_mode;
 
+	conf_tcp_list        tcp_list;
+	conf_tls_list        tls_list;
 	conf_sqlite          sqlite;
 	conf_tls             tls;
 	conf_http_server     http_server;

include/nng/supplemental/nanolib/env.h

@@ -54,8 +54,10 @@
 #define NANOMQ_CONF_PATH "NANOMQ_CONF_PATH"
 
 #define NANOMQ_VIN "NANOMQ_VIN"
+#define NANOMQ_PID_FILE "NANOMQ_PID_FILE"
 
 NNG_DECL void read_env_conf(conf *config);
 NNG_DECL char *read_env_vin();
+NNG_DECL char *read_env_pid_file();
 
 #endif

src/mqtt/protocol/mqtt/mqtt_client.c

@@ -432,6 +432,8 @@ mqtt_send_msg(nni_aio *aio, mqtt_ctx_t *arg)
 		nni_mqtt_msg_set_aio(msg, aio);
 		packet_id = nni_mqtt_msg_get_packet_id(msg);
 		taio = nni_id_get(&p->sent_unack, packet_id);
+		// pass proto_data to cached aio, either it is freed in ack or in cancel
+		nni_aio_set_prov_data(aio, nni_msg_get_proto_data(msg));
 		if (taio != NULL) {
 			log_warn("Warning : msg %d lost due to "
 			                "packetID duplicated!", packet_id);
@@ -530,9 +532,7 @@ static void
 mqtt_pipe_stop(void *arg)
 {
 	mqtt_pipe_t *p = arg;
-	// nni_aio_abort(&p->send_aio, NNG_ECANCELED);
 	nni_aio_stop(&p->send_aio);
-	// nni_aio_abort(&p->recv_aio, NNG_ECANCELED);
 	nni_aio_stop(&p->recv_aio);
 	nni_aio_stop(&p->time_aio);
 }