new feature: nng pub/sub bridging

include/nng/protocol/mqtt/mqtt_parser.h

@@ -147,4 +147,6 @@ NNG_DECL int nmq_auth_http_sub_pub(
     conn_param *cparam, bool is_sub, topic_queue *topics, conf_auth_http *conf);
 
 NNG_DECL int mqtt_get_remaining_length(uint8_t *, uint32_t, uint32_t *, uint8_t *);
+NNG_DECL nng_msg *nng_sub0_msg_adapter(nng_msg *, char *);
+
 #endif // NNG_MQTT_H

include/nng/supplemental/nanolib/conf.h

@@ -598,6 +598,40 @@ struct conf_web_hook {
 
 typedef struct conf_web_hook  conf_web_hook;
 
+struct conf_nng_pub_node {
+	char       *name;
+	nng_socket  pub_sock;
+	char       *pub_url;
+	char       *clientid;
+	conn_param *client; // For ACL usage
+	topics    **pub_list;
+	size_t      forwards_count;
+};
+typedef struct conf_nng_pub_node conf_nng_pub_node;
+
+struct conf_nng_sub_node {
+	char      *name;
+	nng_socket sub_sock;
+	char      *sub_url;
+	char       *clientid;
+	conn_param *client; // For ACL usage
+	// NNG topics must be topic/payload, no multiple "/"
+	topics **sub_list;
+	size_t   inwards_count;
+};
+typedef struct conf_nng_sub_node conf_nng_sub_node;
+
+struct conf_nng_bridge {
+	bool                pub_enable;
+	bool                sub_enable;
+	conf_nng_pub_node **pnodes;
+	conf_nng_sub_node **snodes;
+	size_t              pub_count;
+	size_t              sub_count;
+};
+
+typedef struct conf_nng_bridge  conf_nng_bridge;
+
 typedef enum {
 	memory,
 	sqlite,
@@ -616,7 +650,7 @@ struct conf {
 	int        num_taskq_thread;
 	int        max_taskq_thread;
 	uint32_t   parallel;			   // broker ctx
-	uint64_t   total_ctx;		       // Total ctx of work (bridge + AWS + broker + HTTP)
+	uint64_t   total_ctx;		       // aio number of any ctx will send msg to bridge. (bridge + AWS + broker + HTTP + nng_sub)
 	uint64_t   max_packet_size;        // byte
 	uint32_t   client_max_packet_size; // byte
 	uint32_t   max_inflight_window;
@@ -641,7 +675,7 @@ struct conf {
 	conf_bridge          aws_bridge;	// AWS IoT Core
 	conf_exchange        exchange;
 	conf_parquet         parquet;
-	conf_blf             blf;
+	conf_nng_bridge      nng_proxy;
 #if defined(SUPP_PLUGIN)
 	conf_plugin          plugin;
 #endif

include/nng/supplemental/nanolib/nanolib.h

@@ -8,20 +8,22 @@
 #include <stdlib.h>
 #include "nng/supplemental/nanolib/conf.h"
 
-extern void conf_tls_init(conf_tls *tls);
-extern void conf_tls_destroy(conf_tls *tls);
-extern void conf_tls_parse(
+NNG_DECL void conf_tls_init(conf_tls *tls);
+NNG_DECL void conf_tls_destroy(conf_tls *tls);
+NNG_DECL void conf_tls_parse(
     conf_tls *tls, const char *path, const char *prefix1, const char *prefix2);
 
-extern void conf_http_server_init(conf_http_server *http, uint16_t port);
-extern void conf_http_server_destroy(conf_http_server *http);
+NNG_DECL void conf_http_server_init(conf_http_server *http, uint16_t port);
+NNG_DECL void conf_http_server_destroy(conf_http_server *http);
 
-extern void conf_session_node_init(conf_session_node *node);
-extern void conf_bridge_node_init(conf_bridge_node *node);
-extern void conf_bridge_sub_properties_init(conf_bridge_sub_properties *prop);
-extern void conf_bridge_conn_properties_init(
+NNG_DECL void conf_session_node_init(conf_session_node *node);
+NNG_DECL void conf_bridge_node_init(conf_bridge_node *node);
+NNG_DECL void conf_bridge_snode_init(conf_nng_sub_node *node);
+NNG_DECL void conf_bridge_pnode_init(conf_nng_pub_node *node);
+NNG_DECL void conf_bridge_sub_properties_init(conf_bridge_sub_properties *prop);
+NNG_DECL void conf_bridge_conn_properties_init(
     conf_bridge_conn_properties *prop);
-extern void conf_bridge_conn_will_properties_init(
+NNG_DECL void conf_bridge_conn_will_properties_init(
     conf_bridge_conn_will_properties *prop);
 
 #endif

src/sp/protocol/mqtt/mqtt_parser.c

@@ -877,6 +877,7 @@ conn_param_init(conn_param *cparam)
 	cparam->properties      = NULL;
 	cparam->will_prop_len   = 0;
 	cparam->will_properties = NULL;
+	cparam->nano_qos_db     = NULL;
 }
 
 int
@@ -2138,3 +2139,64 @@ mqtt_get_remaining_length(uint8_t *packet, uint32_t len,
 
 	return MQTT_ERR_INVAL;
 }
+
+/**
+ * @brief convert NNG sub0 msg to standard MQTT V4 msg.
+ * 
+ * @param origin original NNG sub msg
+ * @param topic  Default topic of NNG sub msg
+ * @return nng_msg
+ */
+nng_msg *
+nng_sub0_msg_adapter(nng_msg *origin, char *topic)
+{
+	nng_msg *mqtt_msg = NULL;
+	const uint8_t *body     = nng_msg_body(origin);
+	size_t         body_len = nng_msg_len(origin);
+
+	if (body == NULL || body_len == 0) {
+		log_error("Empty origin message");
+		return NULL;
+	}
+
+	// Define '/' to separate topic and payload.
+	// NNG SUB msg format: "topic/path/payload" -> Topic: "topic", Payload: "path/payload"
+	const char *ptr = (const char *)body;
+	char       *sep = NULL;
+
+	for (size_t i = 0; i < body_len; i++) {
+		if (ptr[i] == '/') {
+			sep = &ptr[i];
+			break;
+		}
+	}
+
+	// If no separator found, or starts with '/', fallback to using the whole message as payload.
+	if (sep == NULL || sep == ptr) {
+		log_warn("No valid topic/payload separator found in NNG sub0 msg.");
+		mqtt_msg = nano_encode_publish_msg(MQTT_PROTOCOL_VERSION_v311,
+		    0, false, false, body, body_len, NULL, topic, NULL);
+		return mqtt_msg;
+	}
+
+	size_t   topic_len    = sep - ptr;
+	char    *topic_buf    = nng_zalloc(topic_len + 1);
+	uint8_t *payload_data = (uint8_t *) sep + 1;
+	size_t   payload_len  = body_len - (topic_len + 1);
+
+	if (!topic_buf) {
+		log_error("Failed to allocate memory for topic");
+		return NULL;
+	}
+
+	memcpy(topic_buf, ptr, topic_len);
+
+	mqtt_msg = nano_encode_publish_msg(MQTT_PROTOCOL_VERSION_v311, 0,
+	    false, false, payload_data, payload_len, NULL, topic_buf, NULL);
+
+	nng_free(topic_buf, topic_len + 1);
+	if (mqtt_msg == NULL) {
+		log_error("Build MQTT msg from NNG sub0 msg failed");
+	}
+	return mqtt_msg;
+}

src/sp/protocol/mqtt/nmq_mqtt.c

@@ -231,7 +231,12 @@ nano_pipe_timer_cb(void *arg)
 		return;
 	}
 	nni_mtx_lock(&p->lk);
-
+	if (nni_atomic_get_bool(&p->closed)) {
+		// stop sending msg to closed pipe
+		nni_mtx_unlock(&p->lk);
+		log_warn("pipe is already closed");
+		return;
+	}
 	// According to the MQTT protocol, when keepalive is 0, the server should not check the keepalive timeout.
 	if (p->keepalive != 0) {
 		qos_backoff = p->ka_refresh * (qos_duration) * 1000 -

src/sp/transport/mqtt/broker_tcp.c

@@ -150,13 +150,6 @@ tcptran_pipe_close(void *arg)
 		nni_free(p->npipe->subinfol, sizeof(nni_list));
 		p->npipe->subinfol = NULL;
 	}
-	void *nano_qos_db = p->npipe->nano_qos_db;
-	if (!p->conf->sqlite.enable && nano_qos_db != NULL) {
-		nni_qos_db_remove_all_msg(
-		    false, nano_qos_db, tran_close_unack_msg_cb);
-		nni_qos_db_fini_id_hash(nano_qos_db);
-		p->npipe->nano_qos_db = NULL;
-	}
 	nni_lmq_flush(&p->rslmq);
 	nni_mtx_unlock(&p->mtx);
 
@@ -228,6 +221,14 @@ tcptran_pipe_fini(void *arg)
 		nni_mtx_unlock(&ep->mtx);
 	}
 	nni_mtx_lock(&p->mtx);
+	void *nano_qos_db = p->npipe->nano_qos_db;
+	if (p->npipe != NULL && p->conf != NULL &&
+	    !p->conf->sqlite.enable && nano_qos_db != NULL) {
+		nni_qos_db_remove_all_msg(
+		    false, nano_qos_db, tran_close_unack_msg_cb);
+		nni_qos_db_fini_id_hash(nano_qos_db);
+		p->npipe->nano_qos_db = NULL;
+	}
 	if (p->tcp_cparam) {
 		conn_param_free(p->tcp_cparam);	//reap thread working
 		p->tcp_cparam = NULL;

src/sp/transport/mqtts/broker_tls.c

@@ -153,13 +153,6 @@ tlstran_pipe_close(void *arg)
 		nni_free(p->npipe->subinfol, sizeof(nni_list));
 		p->npipe->subinfol = NULL;
 	}
-	void *nano_qos_db = p->npipe->nano_qos_db;
-	if (!p->conf->sqlite.enable && nano_qos_db != NULL) {
-		nni_qos_db_remove_all_msg(
-		    false, nano_qos_db, tran_close_unack_msg_cb);
-		nni_qos_db_fini_id_hash(nano_qos_db);
-		p->npipe->nano_qos_db = NULL;
-	}
 	nni_lmq_flush(&p->rslmq);
 	nni_mtx_unlock(&p->mtx);
 
@@ -235,6 +228,14 @@ tlstran_pipe_fini(void *arg)
 		nni_mtx_unlock(&ep->mtx);
 	}
 	nni_mtx_lock(&p->mtx);
+	void *nano_qos_db = p->npipe->nano_qos_db;
+	if (p->npipe != NULL && p->conf != NULL &&
+		!p->conf->sqlite.enable && nano_qos_db != NULL) {
+		nni_qos_db_remove_all_msg(
+		    false, nano_qos_db, tran_close_unack_msg_cb);
+		nni_qos_db_fini_id_hash(nano_qos_db);
+		p->npipe->nano_qos_db = NULL;
+	}
 	if (p->tcp_cparam) {
     		conn_param_free(p->tcp_cparam);
     		p->tcp_cparam = NULL;

src/sp/transport/mqttws/nmq_websocket.c

@@ -616,7 +616,8 @@ wstran_pipe_recv_cb(void *arg)
 		nni_aio_finish_error(p->ep_aio, rv);
 	} else if (uaio != NULL) {
 		nni_aio_set_msg(uaio, NULL);
-		nni_aio_finish_error(uaio, p->err_code == MQTT_SUCCESS ? rv : p->err_code);
+		nni_aio_finish_error(uaio,
+		    p->err_code == MQTT_SUCCESS ? rv : (int) p->err_code);
 	} else {
 		// Let protocol layer do the close first.
 		nng_stream_close(p->ws);
@@ -1333,6 +1334,15 @@ wstran_pipe_fini(void *arg)
 	nni_msg_free(p->tmp_msg);
 	nni_lmq_flush(&p->recvlmq);
 	nng_free(p->qos_buf, 16 + NNI_NANO_MAX_PACKET_SIZE);
+	// must free nano_qos_db in fini for safety
+	void *nano_qos_db = p->npipe->nano_qos_db;
+	if (p->npipe != NULL && p->conf != NULL &&
+		!p->conf->sqlite.enable && nano_qos_db != NULL) {
+		nni_qos_db_remove_all_msg(
+		    false, nano_qos_db, tran_close_unack_msg_cb);
+		nni_qos_db_fini_id_hash(nano_qos_db);
+		p->npipe->nano_qos_db = NULL;
+	}
 	nni_mtx_unlock(&p->mtx);
 
 	nng_stream_free(p->ws);
@@ -1375,13 +1385,7 @@ wstran_pipe_close(void *arg)
 		nni_free(p->npipe->subinfol, sizeof(nni_list));
 		p->npipe->subinfol = NULL;
 	}
-	void *nano_qos_db = p->npipe->nano_qos_db;
-	if (!p->conf->sqlite.enable && nano_qos_db != NULL) {
-		nni_qos_db_remove_all_msg(
-		    false, nano_qos_db, tran_close_unack_msg_cb);
-		nni_qos_db_fini_id_hash(nano_qos_db);
-		p->npipe->nano_qos_db = NULL;
-	}
+
 	nni_lmq_flush(&p->rslmq);
 	nni_mtx_unlock(&p->mtx);
 	nng_stream_close(p->ws);

src/supplemental/nanolib/CMakeLists.txt

@@ -46,5 +46,4 @@ endif (SUPP_RULE_ENGINE)
 add_subdirectory(linkedlist)
 add_subdirectory(ringbuffer)
 add_subdirectory(parquet)
-add_subdirectory(blf)