* NEW [conf] add ctx_msgs as a app layer cache mq for bridging

Signed-off-by: jaylin <jaylin@emqx.io>
Signed-off-by: JaylinYu <letrangerjaylin@gmail.com>

include/nng/supplemental/nanolib/conf.h

@@ -258,6 +258,7 @@ typedef struct {
 struct conf_bridge_node {
 	bool         enable;
 	bool         dynamic;
+	bool		 busy;
 	bool         clean_start;
 	bool         transparent;
 	bool         will_flag;
@@ -291,6 +292,9 @@ struct conf_bridge_node {
 	conf_tcp     tcp;
 	conf_sqlite *sqlite;
 	nng_aio    **bridge_aio;
+	nng_aio		*resend_aio;
+	nng_dialer   dialer;	// in order to postpone bridging client start after local broker
+	nng_lmq		*ctx_msgs;  // only cache qos msg blocked by aio busy
 	nng_mtx     *mtx;
 
 	bool         hybrid;  // enable/disable hybrid bridging

src/supplemental/nanolib/conf.c

@@ -2738,6 +2738,7 @@ conf_bridge_node_init(conf_bridge_node *node)
 {
 	node->sock           = NULL;
 	node->name           = NULL;
+	node->busy 			 = false;
 	node->enable         = false;
 	node->parallel       = 2;
 	node->address        = NULL;
@@ -2789,6 +2790,8 @@ conf_bridge_node_init(conf_bridge_node *node)
 	node->conn_properties = NULL;
 	node->will_properties = NULL;
 	node->sub_properties  = NULL;
+	// TODO compatible with bridge reload
+	node->ctx_msgs = NULL;
 }
 
 static void
@@ -2964,6 +2967,7 @@ conf_bridge_node_parse_with_name(const char *path,const char *key_prefix, const
 		} else if ((value = get_conf_value_with_prefix2(line, sz,
 		                key_prefix, name, ".max_send_queue_len")) != NULL) {
 			node->max_send_queue_len = atoi(value);
+			nng_lmq_alloc(&node->ctx_msgs, node->max_send_queue_len);
 			free(value);
 		} else if ((value = get_conf_value_with_prefix2(line, sz,
 		                key_prefix, name, ".resend_interval")) != NULL) {
@@ -3167,6 +3171,10 @@ conf_bridge_node_destroy(conf_bridge_node *node)
 		free(node->will_payload);
 		node->will_payload = NULL;
 	}
+	if (node->ctx_msgs) {
+		nng_lmq_flush(node->ctx_msgs);
+		nng_lmq_free(node->ctx_msgs);
+	}
 	if (node->forwards_count > 0 && node->forwards_list) {
 		for (size_t i = 0; i < node->forwards_count; i++) {
 			topics *s = node->forwards_list[i];