* MDF [test conf] enable tests with different conf.

cmake/NanoMQHelpers.cmake

@@ -11,6 +11,6 @@ function(nanomq_test NAME)
         target_include_directories(${NAME} PRIVATE
                 ${PROJECT_SOURCE_DIR}/include)
         add_test(NAME nanomq.${NAME} COMMAND ${NAME} -t -v)
-        set_tests_properties(nanomq.${NAME} PROPERTIES TIMEOUT 180)
+        set_tests_properties(nanomq.${NAME} PROPERTIES TIMEOUT 60)
     endif ()
 endfunction()

nanomq/apps/broker.c

@@ -1122,7 +1122,7 @@
 			conf *conf = works[0]->config;
 			if(is_testing == true && conf->bridge.count > 0) {
 				// bridge might need more time to response to the resquest
-				nng_msleep(4000); 
+				nng_msleep(8 * 1000); 
 			}
 			for (size_t t = 0; t < conf->bridge.count; t++) {
 				conf_bridge_node *node = conf->bridge.nodes[t];
@@ -1649,10 +1649,10 @@
 
 	if (nanomq_conf == NULL) {
 		if ((nanomq_conf = nng_zalloc(sizeof(conf))) == NULL) {
 			fprintf(stderr,
 			    "Cannot allocate storge for configuration, "
 			    "quit\n");
 			exit(EXIT_FAILURE);
 		}
 
 		conf_init(nanomq_conf);
@@ -1685,7 +1685,7 @@
 	}
 	// Active the configure for nanomq
 	if ((rc = active_conf(nanomq_conf)) != 0) {
 		broker_start_rc = rc;
 	}
 #if defined(ENABLE_LOG)
 	if ((rc = log_init(&nanomq_conf->log)) != 0) {

nanomq/tests/bridge_test.c

@@ -4,11 +4,11 @@
 int
 main()
 {
-	char *cmd_sub_emqx[] = {"mosquitto_sub", "-h", "broker.emqx.io", "-p", "1883", "-t", "forward1/test", "-V", "mqttv5", NULL};
-	char *cmd_sub_nmq[] = {"mosquitto_sub", "-h", "127.0.0.1", "-p", "1881", "-t", "recv/topic1", "-V", "mqttv5", NULL};
+	char *cmd_sub_emqx[] = {"mosquitto_sub", "-h", "broker.emqx.io", "-p", "1883", "-t", "forward1/test", "-V", "mqttv5", "-q", "2", NULL};
+	char *cmd_sub_nmq[] = {"mosquitto_sub", "-h", "127.0.0.1", "-p", "1881", "-t", "recv/topic1", "-V", "mqttv5", "-q", "2", NULL};
 
-	char *cmd_pub_nmq = "mosquitto_pub -h 127.0.0.1 -p 1881 -t forward1/test -m message-to-emqx -V mqttv5";
-	char *cmd_pub_emqx = "mosquitto_pub -h broker.emqx.io -p 1883 -t recv/topic1 -m message-to-nmq -V mqttv5";
+	char *cmd_pub_nmq = "mosquitto_pub -h 127.0.0.1 -p 1881 -t forward1/test -m message-to-emqx -V mqttv5 -q 2";
+	char *cmd_pub_emqx = "mosquitto_pub -h broker.emqx.io -p 1883 -t recv/topic1 -m message-to-nmq -V mqttv5 -q 2";
 
 	nng_thread *nmq;
 	pid_t       pid_sub_nmq;
@@ -21,22 +21,25 @@ main()
 	int  outfp_nmq, outfp_emqx;
 	char buf_nmq[buf_size];
 	char buf_emqx[buf_size];
+	memset(buf_nmq, 0, buf_size);
+	memset(buf_emqx, 0, buf_size);
 
 	// create nmq thread
-	conf = get_test_conf();
+	conf = get_test_conf(BRIDGE_CONF);
 	assert(conf != NULL);
 	nng_thread_create(&nmq, (void *) broker_start_with_conf, (void *) conf);
-	nng_msleep(50); // wait a while before sub
-
+	nng_msleep(1000); // wait a while before sub
 	pid_sub_nmq = popen_sub_with_cmd(&outfp_nmq, cmd_sub_nmq);
 	pid_sub_emqx = popen_sub_with_cmd(&outfp_emqx, cmd_sub_emqx);
-	nng_msleep(500);
+	nng_msleep(2000);
 	p_pub_emqx = popen(cmd_pub_emqx, "r");
 	p_pub_nmq= popen(cmd_pub_nmq, "r");
 	// check recv msg
 	assert(read(outfp_nmq, buf_nmq, buf_size) != -1);
+	printf("get the msg in nmq:%s\n", buf_nmq);
 	assert(strncmp(buf_nmq, "message-to-nmq", 14) == 0);
 	assert(read(outfp_emqx, buf_emqx, buf_size) != -1);
+	printf("get the msg in emqx:%s\n", buf_emqx);
 	assert(strncmp(buf_emqx, "message-to-emqx", 15) == 0);
 
 	kill(pid_sub_nmq, SIGKILL);

nanomq/tests/http_server_test.c

@@ -592,7 +592,7 @@ main()
 	conf       *conf;
 	FILE       *fd;
 
-	conf = get_test_conf();
+	conf = get_test_conf(ALL_FEATURE_CONF);
 	assert(conf != NULL);
 	nng_thread_create(&nmq, (void *) broker_start_with_conf, (void *) conf);
 	// nng_msleep(100);  // wait a while for broker to init
@@ -662,6 +662,7 @@ main()
 	// // tested now.
 	// assert(test_restart());
 	// assert(test_stop());
+	pclose(fd);
 
 	nng_thread_destroy(nmq);
 }

nanomq/tests/nanomq_bridge_test.conf

@@ -0,0 +1,114 @@
+# NanoMQ Configuration 0.18.0
+
+# #============================================================
+# # NanoMQ Broker
+# #============================================================
+
+mqtt {
+    property_size = 32
+    max_packet_size = 10KB
+    max_mqueue_len = 2048
+    retry_interval = 10s
+    keepalive_multiplier = 1.25
+
+    # Three of below, unsupported now
+    max_inflight_window = 2048
+    max_awaiting_rel = 10s
+    await_rel_timeout = 10s
+}
+
+listeners.tcp {
+    bind = "0.0.0.0:1881"
+}
+
+# listeners.ssl {
+# 	bind = "0.0.0.0:8883"
+# 	keyfile = "/etc/certs/key.pem"
+# 	certfile = "/etc/certs/cert.pem"
+# 	cacertfile = "/etc/certs/cacert.pem"
+# 	verify_peer = false
+# 	fail_if_no_peer_cert = false
+# }
+
+listeners.ws {
+    bind = "0.0.0.0:8083/mqtt"
+}
+
+http_server {
+    port = 8081
+    limit_conn = 2
+    username = admin_test
+    password = pw_test
+    auth_type = basic
+    jwt {
+        public.keyfile = "/etc/certs/jwt/jwtRS256.key.pub"
+    }
+}
+
+log {
+    to = [file, console]
+    level = error
+    dir = "/tmp"
+    file = "nanomq.log"
+    rotation {
+        size = 10MB
+        count = 5
+    }
+}
+
+auth {
+    allow_anonymous = true
+    no_match = allow
+    deny_action = ignore
+
+    cache = {
+        max_size = 32
+        ttl = 1m
+    }
+
+    # password = {include "/etc/nanomq_pwd.conf"}
+    # acl = {include "/etc/nanomq_acl.conf"}
+}
+
+bridges.mqtt.emqx {
+	## TCP URL 格式:  mqtt-tcp://host:port
+	## TLS URL 格式:  tls+mqtt-tcp://host:port
+	## QUIC URL 格式: mqtt-quic://host:port
+	server = "mqtt-tcp://broker.emqx.io:1883"
+	# server = "mqtt-tcp://116.205.239.134:1883"
+	## MQTT 协议版本 （ 4 ｜ 5 ）
+	proto_ver = 5
+	# username = admin
+	# password = public
+	clean_start = true
+	keepalive = 60s
+	## 如果通过 TLS 桥接将下面的代码取消注释
+	## ssl {
+	## 	keyfile = "/etc/certs/key.pem"
+	## 	certfile = "/etc/certs/cert.pem"
+	## 	cacertfile = "/etc/certs/cacert.pem"
+	## }
+	forwards = ["forward1/test","forward2/test"]
+	subscription = [
+		{
+			topic = "recv/topic1"
+			qos = 1
+		},
+		{
+			topic = "recv/topic2"
+			qos = 2
+		},
+		{
+			topic = "cmd/topic1"
+			qos = 1
+		},
+		{
+			topic = "cmd/topic2"
+			qos = 2
+		}
+	]
+	max_parallel_processes = 2 
+	max_send_queue_len = 1024
+	max_recv_queue_len = 1024
+}
+

nanomq/tests/rule_engine_test.c

@@ -11,7 +11,7 @@ main()
 	conf       *nmq_conf = NULL;
 
 	// create nmq thread
-	nmq_conf = get_test_conf();
+	nmq_conf = get_test_conf(ALL_FEATURE_CONF);
 	assert(nmq_conf != NULL);
 	nng_thread_create(&nmq, (void *) broker_start_with_conf, (void *) nmq_conf);
 	nng_msleep(50); // wait a while before sub

nanomq/tests/tests_api.h

@@ -1,6 +1,8 @@
 // This is a test only Scenario for advanced features of NanoMQ, like webhook, etc.
 #define INPROC_TEST_URL "inproc://test"
 #define REST_TEST_URL "http://0.0.0.0:%u/hook"
+#define	ALL_FEATURE_CONF_PATH "../../../nanomq/tests/nanomq_test.conf"
+#define	BRIDGE_CONF_PATH "../../../nanomq/tests/nanomq_bridge_test.conf"
 
 int webhook_msg_cnt = 0; // this is a silly signal to indicate whether the webhook tests pass
 
@@ -50,6 +52,10 @@ int webhook_msg_cnt = 0; // this is a silly signal to indicate whether the webho
 //
 // The above flow is pretty linear, and so we use contexts (nng_ctx) to
 // obtain parallelism.
+typedef enum {
+	ALL_FEATURE_CONF,
+	BRIDGE_CONF,
+} conf_type;
 
 typedef enum {
 	SEND_REQ, // Sending REQ request
@@ -458,14 +464,25 @@ get_webhook_conf()
 }
 
 conf *
-get_test_conf()
+get_test_conf(conf_type type)
 {
 	// get conf from file
-	conf *nmq_conf = NULL;
-	if((nmq_conf = nng_zalloc(sizeof(conf))) == NULL) {
+	conf *nmq_conf  = NULL;
+	char *conf_path = NULL;
+
+	if ((nmq_conf = nng_zalloc(sizeof(conf))) == NULL) {
 		return nmq_conf;
 	}
-	char *conf_path = "../../../nanomq/tests/nanomq_test.conf";
+	switch (type) {
+	case ALL_FEATURE_CONF:
+		conf_path = ALL_FEATURE_CONF_PATH;
+		break;
+	case BRIDGE_CONF:
+		conf_path = BRIDGE_CONF_PATH;
+		break;
+	default:
+		break;
+	}
 	conf_init(nmq_conf);
 	nmq_conf->conf_file = conf_path;
 	conf_parse_ver2(nmq_conf);