From 78e63f49b9b8b0d4097656680374d36096f58c58 Mon Sep 17 00:00:00 2001 From: Andrii Fil Date: Sun, 16 Aug 2026 13:27:06 +0300 Subject: [PATCH 1/2] index on master: 07cec5b setMaxInflightMessages From 0cd4fb7b3d201575b7649ee4f1466c0e4375afa3 Mon Sep 17 00:00:00 2001 From: Andrii Fil Date: Sun, 16 Aug 2026 16:36:37 +0300 Subject: [PATCH 2/2] review tests --- README.md | 6 +- nmqtt.nim | 4 +- nmqtt/nmqtt_pub.nim | 2 +- tests/connection.nim | 66 ++-- tests/ping.nim | 19 +- tests/publish.nim | 164 +++------- tests/publish_qos.nim | 175 ++++------- tests/publish_retained.nim | 40 +-- tests/subscribe.nim | 616 +++++++++++++++++-------------------- tests/tester.nim | 49 ++- tests/tools.nim | 7 +- tests/unsubscribe.nim | 72 +++-- tests/willmsg.nim | 64 ++-- 13 files changed, 538 insertions(+), 746 deletions(-) diff --git a/README.md b/README.md index 63f75a3..f07649c 100644 --- a/README.md +++ b/README.md @@ -197,7 +197,7 @@ runForever() proc mqttPub() {.async.} = await ctx.start() await ctx.publish("nmqtt", "hallo", 2) - await sleepAsync 500 + await sleepAsync(500) await ctx.disconnect() waitFor mqttPub() @@ -214,11 +214,11 @@ proc mqttSubPub() {.async.} = # Subscribe to topic the topic `nmqtt` await ctx.subscribe("nmqtt", 2, onData) - await sleepAsync 500 + await sleepAsync(500) # Publish a message to the topic `nmqtt` await ctx.publish("nmqtt", "hallo", 2) - await sleepAsync 500 + await sleepAsync(500) # Disconnect await ctx.disconnect() diff --git a/nmqtt.nim b/nmqtt.nim index b67fcb6..8ef008d 100644 --- a/nmqtt.nim +++ b/nmqtt.nim @@ -1077,7 +1077,7 @@ proc runRx(ctx: MqttCtx) {.async.} = proc runPing(ctx: MqttCtx) {.async.} = while true: - await sleepAsync ctx.keepAlive.int * 1000 + await sleepAsync(ctx.keepAlive.int * 1000) let ok = await ctx.sendPingReq() if not ok: break @@ -1135,7 +1135,7 @@ proc runConnect(ctx: MqttCtx) {.async.} = for topic, cb in ctx.pubCallbacks: let msgId = ctx.nextMsgId() ctx.workQueue[msgId] = Work(wk: SubWork, msgId: msgId, topic: topic, qos: cb.qos, typ: Subscribe) - await sleepAsync 1000 + await sleepAsync(1000) # # Public API diff --git a/nmqtt/nmqtt_pub.nim b/nmqtt/nmqtt_pub.nim index 61d6c20..c316bcf 100644 --- a/nmqtt/nmqtt_pub.nim +++ b/nmqtt/nmqtt_pub.nim @@ -47,7 +47,7 @@ proc nmqttPub(host="127.0.0.1", port=1883, ssl=false, clientid="", username="", else: for i in 0..repeat-1: waitFor ctx.publish(topic, msg, qos, retain) - if repeatdelay > 0: waitFor sleepAsync (repeatdelay * 1000) + if repeatdelay > 0: waitFor sleepAsync(repeatdelay * 1000) # Check that the message has been succesfully send while ctx.workQueue.len() > 0: diff --git a/tests/connection.nim b/tests/connection.nim index 0c807a2..8fb37bb 100644 --- a/tests/connection.nim +++ b/tests/connection.nim @@ -5,8 +5,8 @@ suite "test suite for connections": let (tpc, msg) = tdata("connection public broker") proc conn() {.async.} = - let ctx = newMqttCtx("nmqttTestConn") - ctx.set_host("test.mosquitto.org", 1883) + let ctx = newMqttCtx("nmqttTestConn" & tpc) # unique clientid for public broker + ctx.setHost("broker-cn.emqx.io", 1883) await ctx.connect() await sleepAsync(1500) check(ctx.state == Connected) @@ -14,15 +14,15 @@ suite "test suite for connections": await sleepAsync(1500) await ctx.disconnect() check(ctx.state == Disabled) - waitFor conn() + waitFor conn() test "connection public broker SSL": let (tpc, msg) = tdata("connection public broker SSL") proc conn() {.async.} = - let ctx = newMqttCtx("nmqttTestConn") - ctx.set_host("test.mosquitto.org", 8883, true) + let ctx = newMqttCtx("nmqttTestConn" & tpc) # unique clientid for public broker + ctx.setHost("broker-cn.emqx.io", 8883, true) await ctx.connect() await sleepAsync(1500) check(ctx.state == Connected) @@ -30,79 +30,65 @@ suite "test suite for connections": await sleepAsync(1500) await ctx.disconnect() check(ctx.state == Disabled) - waitFor conn() - - - #[test "connection wrong port - timeout": - proc conn() {.async.} = - let ctx = newMqttCtx("nmqttTestConn") - ctx.set_host("test.mosquitto.org", 2222) - await ctx.start() - check(ctx.state == Error) - waitFor conn()]# + waitFor conn() test "connect() to broker": - ## FAILS. Due to `runConnect` it will reconnect forever + let ctxMain = newCtx() proc conn() {.async.} = - await ctxSlave.connect() + await ctxMain.connect() await sleepAsync(500) - check(ctxSlave.state == Connected) + check(ctxMain.state == Connected) # Do important stuff await sleepAsync(500) # Disconnect - await ctxSlave.disconnect() - check(ctxSlave.state == Disabled) + await ctxMain.disconnect() + check(ctxMain.state == Disabled) waitFor conn() - test "start() and reconnect": - ## FAILS. Due to `runConnect` it will reconnect forever + let ctxMain = newCtx() proc conn() {.async.} = await sleepAsync(500) - await ctxSlave.start() - await sleepAsync(500) - check(ctxSlave.state == Connected) + check(ctxMain.state == Connected) # Do important stuff await sleepAsync(500) # Close connection - ctxSlave.state = Disconnecting - ctxSlave.s.close() - echo(ctxSlave.state) # = Disconnected + ctxMain.state = Disconnecting + ctxMain.s.close() await sleepAsync(500) # Auto-reconnect goes on `Disconnected" - ctxSlave.state = Disconnected + ctxMain.state = Disconnected # Auto-reconnect loop is 1000ms, wait 2000ms to ensure loop await sleepAsync(2000) # Check reconnect - check(ctxSlave.state == Connected) + check(ctxMain.state == Connected) # Disconnect - await ctxSlave.disconnect() - check(ctxSlave.state == Disabled) + await ctxMain.disconnect() + check(ctxMain.state == Disabled) waitFor conn() - test "isConnected()": - ## FAILS. Due to `runConnect` it will reconnect forever + let ctxMain = newCtx() proc conn() {.async.} = - check(ctxSlave.isConnected() == false) - await ctxSlave.connect() + check(ctxMain.isConnected() == false) + await ctxMain.connect() await sleepAsync(500) - check(ctxSlave.isConnected() == true) + check(ctxMain.isConnected() == true) - await ctxSlave.disconnect() - check(ctxSlave.state == Disabled) + await ctxMain.disconnect() + check(ctxMain.state == Disabled) - waitFor conn() \ No newline at end of file + waitFor conn() diff --git a/tests/ping.nim b/tests/ping.nim index a6a1f66..72f93fc 100644 --- a/tests/ping.nim +++ b/tests/ping.nim @@ -2,12 +2,14 @@ suite "test suite for ping": test "set ping interval": - let (tpc, msg) = tdata("set ping interval") + let + ctxMain = newCtx() + (tpc, msg) = tdata("set ping interval") proc conn() {.async.} = - ctxSlave.set_ping_interval(1) - await ctxSlave.connect() + ctxMain.setPingInterval(1) + await ctxMain.connect() await sleepAsync(6000) var @@ -21,12 +23,12 @@ suite "test suite for ping": check(pingCount > 3) check(pingResp > 3) - await ctxSlave.disconnect() - await sleepAsync(1500) + await ctxMain.disconnect() + await sleepAsync(500) testDmp = @[] - ctxSlave.set_ping_interval(60) - await ctxSlave.connect() + ctxMain.setPingInterval(60) + await ctxMain.connect() await sleepAsync(6000) pingCount = 0 @@ -39,7 +41,4 @@ suite "test suite for ping": check(pingCount == 0) check(pingResp == 0) - await ctxSlave.disconnect() - await sleepAsync(500) - waitFor conn() \ No newline at end of file diff --git a/tests/publish.nim b/tests/publish.nim index b8c6a34..74e62b1 100644 --- a/tests/publish.nim +++ b/tests/publish.nim @@ -1,48 +1,21 @@ suite "test suite for publish": - test "publish multi line": - let (tpc, _) = tdata("publish multi line") - - const text = """1) If wishes were horses, beggars would ride. + test "publish multi line; json; long (757 chars); special chars": + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc1, _) = tdata("publish multi line") + (tpc2, _) = tdata("publish json") + (tpc3, _) = tdata("publish long (757 chars)") + (tpc4, _) = tdata("publish special chars") + + const text1 = """1) If wishes were horses, beggars would ride. 2) It’s easy to be wise after the event. 3) Watch the doughnut, and not the hole. 4) On a wing and a prayer""" - proc conn() {.async.} = - await sleepAsync 1000 - - var - msgFound: bool - timeout: int - msg: string - - proc on_data_pub1(topic: string, message: string) = - if topic == tpc: - msg = message - - await ctxListen.subscribe(tpc, 0, on_data_pub1) - - await sleepAsync 500 - await ctxMain.publish(tpc, text, 0) - - # Wait for final msg is found - while msg == "": - if timeout == 5: - break - await sleepAsync(1000) - timeout += 1 - - check(text == msg) - await ctxListen.unsubscribe(tpc) - await sleepAsync 500 - waitFor conn() - - - test "publish json": - let (tpc, _) = tdata("publish json") - - const text = """ + const text2 = """ { "Novo": { "priceLatest": 359.35, @@ -75,103 +48,46 @@ suite "test suite for publish": "success": true } }""" - proc conn() {.async.} = - await sleepAsync 1000 - - var - msgFound: bool - timeout: int - msg: string - - proc on_data_pub2(topic: string, message: string) = - if topic == tpc: - msg = message - - await ctxListen.subscribe(tpc, 0, on_data_pub2) - - await sleepAsync 500 - await ctxMain.publish(tpc, text, 0) - - # Wait for final msg is found - while msg == "": - if timeout == 5: - break - await sleepAsync(1000) - timeout += 1 - - check(text == msg) - await ctxListen.unsubscribe(tpc) - await sleepAsync 500 - waitFor conn() - + const text3 = "Nim code specifies a computation that acts on a memory consisting of components called locations. A variable is basically a name for a location. Each variable and location is of a certain type. The variable's type is called static type, the location's type is called dynamic type. If the static type is not the same as the dynamic type, it is a super-type or subtype of the dynamic type. An identifier is a symbol declared as a name for a variable, type, procedure, etc. The region of the program over which a declaration applies is called the scope of the declaration. Scopes can be nested. The meaning of an identifier is determined by the smallest enclosing scope in which the identifier is declared unless overloading resolution rules suggest otherwise." - test "publish long (757 chars)": - let (tpc, _) = tdata("publish long (757 chars)") - - const text = "Nim code specifies a computation that acts on a memory consisting of components called locations. A variable is basically a name for a location. Each variable and location is of a certain type. The variable's type is called static type, the location's type is called dynamic type. If the static type is not the same as the dynamic type, it is a super-type or subtype of the dynamic type. An identifier is a symbol declared as a name for a variable, type, procedure, etc. The region of the program over which a declaration applies is called the scope of the declaration. Scopes can be nested. The meaning of an identifier is determined by the smallest enclosing scope in which the identifier is declared unless overloading resolution rules suggest otherwise." + const text4 = "*~\"%?+#!öôéè|§½';£@$ 诶艾弗艾儿豆贝尔维 НимИсТчеБест æøå αγλρξψ mənʊʃjõəd̪ʱɪkaːɾõ 😆😎😍😘" proc conn() {.async.} = - await sleepAsync 1000 - var - msgFound: bool - timeout: int - msg: string + msgFound1: bool + msgFound2: bool + msgFound3: bool + msgFound4: bool - proc on_data_pub3(topic: string, message: string) = - if topic == tpc: - msg = message + proc onDataPub1(topic: string, message: string) = + msgFound1 = topic == tpc1 and message == text1 - await ctxListen.subscribe(tpc, 0, on_data_pub3) + proc onDataPub2(topic: string, message: string) = + msgFound2 = topic == tpc2 and message == text2 - await sleepAsync 500 - await ctxMain.publish(tpc, text, 0) + proc onDataPub3(topic: string, message: string) = + msgFound3 = topic == tpc3 and message == text3 - # Wait for final msg is found - while msg == "": - if timeout == 5: - break - await sleepAsync(1000) - timeout += 1 + proc onDataPub4(topic: string, message: string) = + msgFound4 = topic == tpc4 and message == text4 - check(text == msg) - await ctxListen.unsubscribe(tpc) - await sleepAsync 500 - waitFor conn() + await ctxListen.subscribe(tpc1, 0, onDataPub1) + await ctxListen.subscribe(tpc2, 0, onDataPub2) + await ctxListen.subscribe(tpc3, 0, onDataPub3) + await ctxListen.subscribe(tpc4, 0, onDataPub4) + await sleepAsync(500) + await ctxMain.publish(tpc1, text1, 0) + await ctxMain.publish(tpc2, text2, 0) + await ctxMain.publish(tpc3, text3, 0) + await ctxMain.publish(tpc4, text4, 0) - test "publish special chars": - let (tpc, _) = tdata("publish special chars") + await sleepAsync(500) - const text = "*~\"%?+#!öôéè|§½';£@$ 诶艾弗艾儿豆贝尔维 НимИсТчеБест æøå αγλρξψ mənʊʃjõəd̪ʱɪkaːɾõ 😆😎😍😘" + check(msgFound1 == true) + check(msgFound2 == true) + check(msgFound3 == true) + check(msgFound4 == true) - proc conn() {.async.} = - await sleepAsync 1000 - - var - msgFound: bool - timeout: int - msg: string - - proc on_data_pub3(topic: string, message: string) = - if topic == tpc: - msg = message - echo msg - - await ctxListen.subscribe(tpc, 0, on_data_pub3) - - await sleepAsync 500 - await ctxMain.publish(tpc, text, 0) - - # Wait for final msg is found - while msg == "": - if timeout == 5: - break - await sleepAsync(1000) - timeout += 1 - - check(text == msg) - await ctxListen.unsubscribe(tpc) - await sleepAsync 500 - waitFor conn() \ No newline at end of file + waitFor conn() diff --git a/tests/publish_qos.nim b/tests/publish_qos.nim index 8a6b148..79127a4 100644 --- a/tests/publish_qos.nim +++ b/tests/publish_qos.nim @@ -2,126 +2,79 @@ const msgCount = 500 suite "test suite for publish with qos": - test "publish multiple message fast qos=0": - let (tpc, _) = tdata("publish multiple message fast qos=0") + test "publish multiple message fast qos=0,1,2": + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc0, _) = tdata("publish multiple message fast qos=0") + (tpc1, _) = tdata("publish multiple message fast qos=1") + (tpc2, _) = tdata("publish multiple message fast qos=2") proc conn() {.async.} = + await sleepAsync(500) + var - msgFound: bool - timeout: int - msgRec: int - - await sleepAsync 1000 - - proc on_data_qos0(topic: string, message: string) = - if topic == tpc: - msgRec += 1 - if msgRec == msgCount: - msgFound = true - return - await ctxListen.subscribe(tpc, 0, on_data_qos0) - - # Send msg with no delay - var msg: int - for i in 0 .. msgCount-1: - await ctxMain.publish(tpc, $msg, 0) - msg += 1 - - check(msg == msgCount) - check(ctxMain.state == Connected) + msgs0: array[msgCount, bool] + msgs1: array[msgCount, bool] + msgs2: array[msgCount, bool] + receivedAllMsgs0: bool + receivedAllMsgs1: bool + receivedAllMsgs2: bool + + proc checkMsgs(msgs: array[msgCount, bool]): bool = + for m in msgs: + if not m: + return false + true + + proc onDataQoS0(topic: string, message: string) = + let i = parseInt(message) + check(i in 0 .. msgCount - 1) + msgs0[i] = true + + proc onDataQoS1(topic: string, message: string) = + let i = parseInt(message) + check(i in 0 .. msgCount - 1) + msgs1[i] = true + + proc onDataQoS2(topic: string, message: string) = + let i = parseInt(message) + check(i in 0 .. msgCount - 1) + msgs2[i] = true + + await ctxListen.subscribe(tpc0, 0, onDataQoS0) + await ctxListen.subscribe(tpc1, 0, onDataQoS1) + await ctxListen.subscribe(tpc2, 0, onDataQoS2) + + for i in 0 ..< msgCount: + await ctxMain.publish(tpc0, $i, 0) + await ctxMain.publish(tpc1, $i, 1) + await ctxMain.publish(tpc2, $i, 2) - # Wait for final msg is found - while not msgFound: - if timeout == 5: - break - await sleepAsync(1000) - timeout += 1 + check(ctxMain.state == Connected) - check(msgRec == msgCount) - check(ctxMain.workQueue.len == 0) # A ping could cause a failure - await ctxListen.unsubscribe(tpc) - await sleepAsync 500 - waitFor conn() + for i in 0 .. 9: + await sleepAsync(500) + if not receivedAllMsgs0: + receivedAllMsgs0 = checkMsgs(msgs0) + if not receivedAllMsgs1: + receivedAllMsgs1 = checkMsgs(msgs1) + if not receivedAllMsgs2: + receivedAllMsgs2 = checkMsgs(msgs2) - test "publish multiple message fast qos=1": - let (tpc, _) = tdata("publish multiple message fast qos=1") + if receivedAllMsgs0 and receivedAllMsgs1 and receivedAllMsgs2: + break - proc conn() {.async.} = - var - msgFound: bool - timeout: int - msgRec: int - - await sleepAsync(2000) - - proc on_data_qos1(topic: string, message: string) = - if topic == tpc: - msgRec += 1 - if msgRec == msgCount: - msgFound = true - return - await ctxListen.subscribe(tpc, 0, on_data_qos1) - - # Send msg with no delay - var msg: int - for i in 0 .. msgCount-1: - await ctxMain.publish(tpc, $msg, 1) - msg += 1 - - check(msg == msgCount) - check(ctxMain.state == Connected) + check(receivedAllMsgs0 == true) + check(receivedAllMsgs1 == true) + check(receivedAllMsgs2 == true) - # Wait for final msg is found - while not msgFound: - if timeout == 5: + var hasQueue: bool + for w in ctxMain.workQueue.values(): + if w.typ != PingReq: + hasQueue = true break - await sleepAsync(1000) - timeout += 1 + check(hasQueue == false) - check(msgRec == msgCount) - check(ctxMain.workQueue.len == 0) # A ping could cause a failure - await ctxListen.unsubscribe(tpc) - await sleepAsync 500 waitFor conn() - - - test "publish multiple message fast qos=2": - let (tpc, _) = tdata("publish multiple message fast qos=2") - - proc conn() {.async.} = - var - msgFound: bool - timeout: int - msgRec: int - - await sleepAsync(2000) - - proc on_data_qos2(topic: string, message: string) = - if topic == tpc: - msgRec += 1 - if msgRec == msgCount: - msgFound = true - return - await ctxListen.subscribe(tpc, 0, on_data_qos2) - - # Send msg with no delay - var msg: int - for i in 0 .. msgCount-1: - await ctxMain.publish(tpc, $msg, 2) - msg += 1 - - check(msg == msgCount) - check(ctxMain.state == Connected) - - while ctxMain.workQueue.len > 0: - if timeout == 5: - break - await sleepAsync(1000) - timeout += 1 - - check(msgRec == msgCount) - check(ctxMain.workQueue.len == 0) # A ping could cause a failure - await ctxListen.unsubscribe(tpc) - await sleepAsync 500 - waitFor conn() \ No newline at end of file diff --git a/tests/publish_retained.nim b/tests/publish_retained.nim index c37dd68..f1af265 100644 --- a/tests/publish_retained.nim +++ b/tests/publish_retained.nim @@ -2,33 +2,25 @@ suite "test suite for publish retained": test "publish retain msg": - let (tpc, msg) = tdata("publish retain msg") - waitFor ctxMain.publish(tpc, msg, qos=1, retain=true) - waitFor sleepAsync 500 + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("publish retain msg") proc conn() {.async.} = - var - msgFound: bool - timeout: int + var msgFound: bool - # Start listening slave - proc on_data_retain(topic: string, message: string) = + waitFor ctxMain.publish(tpc, msg, qos=1, retain=true) + waitFor sleepAsync(500) + + proc onDataRetain(topic: string, message: string) = if topic == tpc: check(message == msg) msgFound = true - return - await ctxListen.subscribe(tpc, 2, on_data_retain) - - # Wait for retained msg is found - while not msgFound: - if timeout == 5: - # In an ideal world this should take 0sec, but to include - # bad connections and latency we wait 5sec. - check(msgFound == true) - break - await sleepAsync(1000) - timeout += 1 - - await ctxListen.unsubscribe(tpc) - - waitFor conn() \ No newline at end of file + + await ctxListen.subscribe(tpc, 2, onDataRetain) + await sleepAsync(500) + + check(msgFound == true) + + waitFor conn() diff --git a/tests/subscribe.nim b/tests/subscribe.nim index 94d14ae..34fa2ca 100644 --- a/tests/subscribe.nim +++ b/tests/subscribe.nim @@ -2,162 +2,166 @@ suite "test suite for subscribe": test "subscribe to topic qos=0": - let (tpc, msg) = tdata("subscribe to topic qos=0") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to topic qos=0") proc conn() {.async.} = - proc on_data_sub_qos0(topic: string, message: string) = + proc onDataSubQoS0(topic: string, message: string) = if topic == tpc: check(message == msg) return - await ctxListen.subscribe(tpc, 0, on_data_sub_qos0) - await sleepAsync 500 + await ctxListen.subscribe(tpc, 0, onDataSubQoS0) + await sleepAsync(500) await ctxMain.publish(tpc, msg, 0) - await sleepAsync 500 + await sleepAsync(500) await ctxListen.unsubscribe(tpc) - await sleepAsync 500 - check(testDmp[0][0] == "tx> Subscribe(02):") - check(testDmp[1][0] == "rx> SubAck(00):") # and testDmp[1][1] == "00 01 00 ") - check(testDmp[2][0] == "tx> Publish(00):") - check(testDmp[3][0] == "rx> Publish(00):") + await sleepAsync(500) - waitFor conn() + check(hasAllInDmp(@["tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(00):", + "rx> Publish(00):"])) + waitFor conn() test "subscribe to topic qos=1": - let (tpc, msg) = tdata("subscribe to topic qos=1") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to topic qos=1") proc conn() {.async.} = - proc on_data_sub_qos1(topic: string, message: string) = + proc onDataSubQoS1(topic: string, message: string) = if topic == tpc: check(message == msg) return - await ctxListen.subscribe(tpc, 1, on_data_sub_qos1) - await sleepAsync 500 + + await ctxListen.subscribe(tpc, 1, onDataSubQoS1) + await sleepAsync(500) await ctxMain.publish(tpc, msg, 1) - await sleepAsync 500 + await sleepAsync(500) await ctxListen.unsubscribe(tpc) - await sleepAsync 500 - check(testDmp[0][0] == "tx> Subscribe(02):") - check(testDmp[1][0] == "rx> SubAck(00):") # and testDmp[1][1] == "00 02 01 ") # and testDmp[1][1] == "00 01 01 ") - check(testDmp[2][0] == "tx> Publish(02):") - check(testDmp[3][0] == "rx> PubAck(00):") # and testDmp[3][1] == "00 02 ") # and testDmp[3][1] == "00 01 ") - check(testDmp[4][0] == "rx> Publish(02):") - check(testDmp[5][0] == "tx> PubAck(02):") # and testDmp[5][1] == "00 01 ") # and testDmp[5][1] == "00 01 ") + await sleepAsync(500) - waitFor conn() + check(hasAllInDmp(@["tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(02):", + "rx> PubAck(00):", + "rx> Publish(02):", + "tx> PubAck(02):"])) + waitFor conn() test "subscribe to topic qos=2": - let (tpc, msg) = tdata("subscribe to topic qos=2") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to topic qos=2") proc conn() {.async.} = - # We need all these sleepAsync cause it's too fast in -d:release - await sleepAsync 500 - proc on_data_sub_qos2(topic: string, message: string) = + proc onDataSubQoS2(topic: string, message: string) = if topic == tpc: check(message == msg) return - await ctxListen.subscribe(tpc, 2, on_data_sub_qos2) - await sleepAsync 1000 + + await ctxListen.subscribe(tpc, 2, onDataSubQoS2) + await sleepAsync(500) await ctxMain.publish(tpc, msg, 2) - await sleepAsync 1000 + await sleepAsync(500) await ctxListen.unsubscribe(tpc) - await sleepAsync 1000 - check(testDmp[0][0] == "tx> Subscribe(02):") - check(testDmp[1][0] == "rx> SubAck(00):") # and testDmp[1][1] == "00 03 02 ") # and testDmp[1][1] == "00 01 02 ") - check(testDmp[2][0] == "tx> Publish(04):") - check(testDmp[3][0] == "rx> PubRec(00):") # and testDmp[3][1] == "00 03 ") # and testDmp[3][1] == "00 01 ") - check(testDmp[4][0] == "tx> PubRel(02):") # and testDmp[4][1] == "00 03 ") # and testDmp[4][1] == "00 01 ") - check(testDmp[5][0] == "rx> PubComp(00):") # and testDmp[5][1] == "00 03 ") # and testDmp[5][1] == "00 01 ") - check(testDmp[6][0] == "rx> Publish(04):") - check(testDmp[7][0] == "tx> PubRec(02):") # and testDmp[7][1] == "00 02 ") # and testDmp[7][1] == "00 01 ") - check(testDmp[8][0] == "rx> PubRel(02):") # and testDmp[8][1] == "00 02 ") # and testDmp[8][1] == "00 01 ") - check(testDmp[9][0] == "tx> PubComp(02):") # and testDmp[9][1] == "00 02 ") + await sleepAsync(500) - waitFor conn() + check(hasAllInDmp(@["tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(04):", + "rx> PubRec(00):", + "tx> PubRel(02):", + "rx> PubComp(00):", + "rx> Publish(04):", + "tx> PubRec(02):", + "rx> PubRel(02):", + "tx> PubComp(02):"])) + waitFor conn() test "subscribe to multiple topics": - let (tpc, msg) = tdata("subscribe to multiple topics") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to multiple topics") proc conn() {.async.} = - var - topic1: bool - topic2: bool + var topic1, topic2: bool - proc on_data_sub_mul1(topic: string, message: string) = + proc onDataSubMul1(topic: string, message: string) = check(message == msg & "-mul1") - if topic1 == true: - topic1 = false - else: - topic1 = true + check(not topic1) + topic1 = true - proc on_data_sub_mul2(topic: string, message: string) = + proc onDataSubMul2(topic: string, message: string) = check(message == msg & "-mul2") - if topic2 == true: - topic2 = false - else: - topic2 = true - - await ctxListen.subscribe(tpc & "-1", 0, on_data_sub_mul1) - # TODO: This failes without the sleepAsync due to `len(t) == L` the length of the table changed while iterating over it - await sleepAsync 500 - await ctxListen.subscribe(tpc & "-2", 0, on_data_sub_mul2) - await sleepAsync 500 + check(not topic2) + topic2 = true + + await ctxListen.subscribe(tpc & "-1", 0, onDataSubMul1) + await ctxListen.subscribe(tpc & "-2", 0, onDataSubMul2) + await sleepAsync(500) await ctxMain.publish(tpc & "-1", msg & "-mul1", 0) await ctxMain.publish(tpc & "-2", msg & "-mul2", 0) - await sleepAsync 500 + await sleepAsync(500) await ctxListen.unsubscribe(tpc & "-1") await ctxListen.unsubscribe(tpc & "-2") - await sleepAsync 500 - check(topic1 == true) - check(topic2 == true) - waitFor conn() + check(topic1) + check(topic2) + waitFor conn() test "subscribe to multiple with identical topic": - let (tpc, msg) = tdata("subscribe to multiple with identical topic") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to multiple with identical topic") proc conn() {.async.} = - var - sub1: int - sub2: int - sub3: int + var sub1, sub2, sub3: int - proc on_data_sub_mul1(topic: string, message: string) = + proc onDataSubMul1(topic: string, message: string) = if topic == tpc: sub1 += 1 - proc on_data_sub_mul2(topic: string, message: string) = + proc onDataSubMul2(topic: string, message: string) = if topic == tpc: sub2 += 1 - proc on_data_sub_mul3(topic: string, message: string) = + proc onDataSubMul3(topic: string, message: string) = if topic == tpc: sub3 += 1 check(ctxListen.pubCallbacks.len() == 0) # Add 1 msg to sub1 - await ctxListen.subscribe(tpc, 0, on_data_sub_mul1) + await ctxListen.subscribe(tpc, 0, onDataSubMul1) + await sleepAsync(500) await ctxMain.publish(tpc, msg, 0) check(ctxListen.pubCallbacks.len() == 1) - await sleepAsync 500 - await ctxListen.subscribe(tpc, 0, on_data_sub_mul2) + await sleepAsync(500) + await ctxListen.subscribe(tpc, 0, onDataSubMul2) # sub3 now overrides sub1 and sub2 - await ctxListen.subscribe(tpc, 0, on_data_sub_mul3) + await ctxListen.subscribe(tpc, 0, onDataSubMul3) check(ctxListen.pubCallbacks.len() == 1) - await sleepAsync 500 + await sleepAsync(500) await ctxMain.publish(tpc, msg, 0) await ctxMain.publish(tpc, msg, 0) await ctxMain.publish(tpc, msg, 0) - await sleepAsync 500 + await sleepAsync(500) await ctxListen.unsubscribe(tpc) check(ctxListen.pubCallbacks.len() == 0) @@ -165,290 +169,274 @@ suite "test suite for subscribe": check(sub1 == 1) check(sub2 == 0) check(sub3 == 3) - await sleepAsync 500 waitFor conn() test "subscribe to #": - let (_, msg) = tdata("subscribe to #") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to #") proc conn() {.async.} = var msgCount: int - proc on_data_sub_all(topic: string, message: string) = + + proc onDataSubAll(topic: string, message: string) = msgCount += 1 - await ctxListen.subscribe("#", 0, on_data_sub_all) - await sleepAsync 500 - await ctxMain.publish("random1", msg, 0) - await ctxMain.publish("random2", msg, 0) - await ctxMain.publish("random3", msg, 0) - await sleepAsync 500 - await ctxListen.unsubscribe("#") - await sleepAsync 500 + await ctxListen.subscribe(tpc & "/#", 0, onDataSubAll) + await sleepAsync(500) + await ctxMain.publish(tpc & "/random1", msg, 0) + await ctxMain.publish(tpc & "/random2/1", msg, 0) + await ctxMain.publish(tpc & "/random3/2/1/0", msg, 0) + await sleepAsync(500) + await ctxListen.unsubscribe(tpc & "/#") + check(msgCount == 3) waitFor conn() test "subscribe to test/#": - let (_, msg) = tdata("subscribe to test/#") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to test/#") proc conn() {.async.} = var msgCount: int - proc on_data_sub_wild(topic: string, message: string) = + + proc onDataSubWild(topic: string, message: string) = msgCount += 1 - await ctxListen.subscribe("test/#", 0, on_data_sub_wild) - await sleepAsync 500 - await ctxMain.publish("test/random1", msg, 0) - await ctxMain.publish("second/random2", msg, 0) - await ctxMain.publish("test", msg, 0) - await ctxMain.publish("test/random3", msg, 0) - await sleepAsync 500 - await ctxListen.unsubscribe("test/#") - await sleepAsync 500 + await ctxListen.subscribe(tpc & "/test/#", 0, onDataSubWild) + await sleepAsync(500) + await ctxMain.publish(tpc & "/test/random1", msg, 0) + await ctxMain.publish(tpc & "/second/random2", msg, 0) + await ctxMain.publish(tpc & "/test", msg, 0) + await ctxMain.publish(tpc & "/test/random3/2", msg, 0) + await sleepAsync(500) + await ctxListen.unsubscribe(tpc & "/test/#") + check(msgCount == 3) waitFor conn() test "subscribe to test/+": - let (_, msg) = tdata("subscribe to test/+") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to test/+") proc conn() {.async.} = var msgCount: int - proc on_data_sub_wild(topic: string, message: string) = + + proc onDataSubWild(topic: string, message: string) = msgCount += 1 - await ctxListen.subscribe("test/+", 0, on_data_sub_wild) - await sleepAsync 500 - await ctxMain.publish("test/random1", msg, 0) - await ctxMain.publish("second/random2", msg, 0) - await ctxMain.publish("test", msg, 0) - await ctxMain.publish("test/random3", msg, 0) - await sleepAsync 500 - await ctxListen.unsubscribe("test/+") - await sleepAsync 500 + await ctxListen.subscribe(tpc & "/test/+", 0, onDataSubWild) + await sleepAsync(500) + await ctxMain.publish(tpc & "/test/random1", msg, 0) + await ctxMain.publish(tpc & "/second/random2", msg, 0) + await ctxMain.publish(tpc & "/test", msg, 0) + await ctxMain.publish(tpc & "/test/random3", msg, 0) + await ctxMain.publish(tpc & "/test/random3/2", msg, 0) + await sleepAsync(500) + await ctxListen.unsubscribe(tpc & "/test/+") check(msgCount == 2) waitFor conn() test "subscribe to test/+/test": - let (_, msg) = tdata("subscribe to test/+/test") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to test/+/test") proc conn() {.async.} = var msgCount: int - proc on_data_sub_wild(topic: string, message: string) = + + proc onDataSubWild(topic: string, message: string) = msgCount += 1 - await ctxListen.subscribe("test/+/data", 0, on_data_sub_wild) - await sleepAsync 500 - await ctxMain.publish("test/random1/data", msg, 0) - await ctxMain.publish("second/random2/data", msg, 0) - await ctxMain.publish("test/random3", msg, 0) - await ctxMain.publish("test/random4/data", msg, 0) - await ctxMain.publish("test/random5/data/random6", msg, 0) - await sleepAsync 500 - await ctxListen.unsubscribe("test/+/data") - await sleepAsync 500 + await ctxListen.subscribe(tpc & "test/+/data", 0, onDataSubWild) + await sleepAsync(500) + await ctxMain.publish(tpc & "test/random1/data", msg, 0) + await ctxMain.publish(tpc & "second/random2/data", msg, 0) + await ctxMain.publish(tpc & "test/random3", msg, 0) + await ctxMain.publish(tpc & "test/random4/data", msg, 0) + await ctxMain.publish(tpc & "test/random5/data/random6", msg, 0) + await sleepAsync(500) + await ctxListen.unsubscribe(tpc & "test/+/data") check(msgCount == 2) waitFor conn() test "stay subscribed after disconnect with reconnect": - let (tpc, msg) = tdata("stay subscribed after disconnect with reconnect") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("stay subscribed after disconnect with reconnect") proc conn() {.async.} = - await ctxSlave.start() - await sleepAsync(1000) - testDmp = @[] - var msgCount: int - proc on_data_sub_keep(topic: string, message: string) = + + proc onDataSubKeep(topic: string, message: string) = msgCount += 1 - await ctxSlave.subscribe(tpc, 0, on_data_sub_keep) - await sleepAsync 500 + await ctxListen.subscribe(tpc, 0, onDataSubKeep) + await sleepAsync(500) await ctxMain.publish(tpc, msg, 0) # msg 1 - await sleepAsync 500 + await sleepAsync(500) # Disconnect - ctxSlave.state = Disconnecting - ctxSlave.s.close() + ctxListen.state = Disconnecting + ctxListen.s.close() await sleepAsync(500) - ctxSlave.state = Disconnected + ctxListen.state = Disconnected await sleepAsync(2000) # Auto-reconnect loop is 1000ms, wait 2000ms to ensure loop # We should automatic reconnect here await ctxMain.publish(tpc, msg, 0) # msg 2 await ctxMain.publish(tpc, msg, 0) # msg 3 - await sleepAsync 500 - await ctxSlave.unsubscribe(tpc) - await sleepAsync 500 + await sleepAsync(500) + await ctxListen.unsubscribe(tpc) + await sleepAsync(500) check(msgCount == 3) # A total of 3 msgs on this topic - check(testDmp[0][0] == "tx> Subscribe(02):") - check(testDmp[1][0] == "rx> SubAck(00):") - check(testDmp[2][0] == "tx> Publish(00):") - check(testDmp[3][0] == "rx> Publish(00):") - # Disconnected - check(testDmp[4][0] == "tx> Connect(00):") - check(testDmp[5][0] == "rx> ConnAck(00):") - # Re-subscribe - check(testDmp[6][0] == "tx> Subscribe(02):") - check(testDmp[7][0] == "rx> SubAck(00):") - check(testDmp[8][0] == "tx> Publish(00):") - check(testDmp[9][0] == "tx> Publish(00):") - check(testDmp[10][0] == "rx> Publish(00):") - check(testDmp[11][0] == "rx> Publish(00):") - # Unsub - check(testDmp[12][0] == "tx> Unsubscribe(02):") - check(testDmp[13][0] == "rx> Unsuback(00):") - - await ctxSlave.disconnect() - await sleepAsync(1000) - waitFor conn() + check(hasAllInDmp(@["tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(00):", + "rx> Publish(00):", + "tx> Connect(00):", + "rx> ConnAck(00):", + "tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(00):", + "tx> Publish(00):", + "rx> Publish(00):", + "rx> Publish(00):", + "tx> Unsubscribe(02):", + "rx> Unsuback(00):"])) + + await ctxListen.disconnect() + waitFor conn() test "stay subscribed after disconnect with reconnect with same qos=2": - let (tpc, msg) = tdata("stay subscribed after disconnect with reconnect with same qos=2") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("stay subscribed after disconnect with reconnect with same qos=2") proc conn() {.async.} = - await ctxSlave.start() - await sleepAsync(1000) - testDmp = @[] - var msgCount: int - proc on_data_sub_keep(topic: string, message: string) = + + proc onDataSubKeep(topic: string, message: string) = msgCount += 1 - await ctxSlave.subscribe(tpc, 2, on_data_sub_keep) - await sleepAsync 500 + await ctxListen.subscribe(tpc, 2, onDataSubKeep) + await sleepAsync(500) await ctxMain.publish(tpc, msg, 0) # msg 1 - await sleepAsync 500 + await sleepAsync(500) # Disconnect - ctxSlave.state = Disconnecting - ctxSlave.s.close() + ctxListen.state = Disconnecting + ctxListen.s.close() await sleepAsync(500) - ctxSlave.state = Disconnected + ctxListen.state = Disconnected await sleepAsync(2000) # Auto-reconnect loop is 1000ms, wait 2000ms to ensure loop # We should automatic reconnect here await ctxMain.publish(tpc, msg, 2) # msg 2 - #await ctxMain.publish(tpc, msg, 0) # msg 3 - await sleepAsync 500 - await ctxSlave.unsubscribe(tpc) - await sleepAsync 500 - - check(msgCount == 2) # A total of 3 msgs on this topic - - check(testDmp[0][0] == "tx> Subscribe(02):") - check(testDmp[1][0] == "rx> SubAck(00):") - check(testDmp[2][0] == "tx> Publish(00):") - check(testDmp[3][0] == "rx> Publish(00):") - - # Disconnected - check(testDmp[4][0] == "tx> Connect(00):") - check(testDmp[5][0] == "rx> ConnAck(00):") - - # Re-subscribe - check(testDmp[6][0] == "tx> Subscribe(02):") - check(testDmp[7][0] == "rx> SubAck(00):") - - # Publish - qos=2 - check(testDmp[8][0] == "tx> Publish(04):") - check(testDmp[9][0] == "rx> PubRec(00):") - check(testDmp[10][0] == "tx> PubRel(02):") - check(testDmp[11][0] == "rx> PubComp(00):") - - # Subscribe - receive - qos=2 - check(testDmp[12][0] == "rx> Publish(04):") - check(testDmp[13][0] == "tx> PubRec(02):") - check(testDmp[14][0] == "rx> PubRel(02):") - check(testDmp[15][0] == "tx> PubComp(02):") - - # Unsub - check(testDmp[16][0] == "tx> Unsubscribe(02):") - check(testDmp[17][0] == "rx> Unsuback(00):") + await sleepAsync(500) + await ctxListen.unsubscribe(tpc) + await sleepAsync(500) - await ctxSlave.disconnect() - await sleepAsync(1000) + check(msgCount == 2) # A total of 2 msgs on this topic + + check(hasAllInDmp(@["tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(00):", + "rx> Publish(00):", + "tx> Connect(00):", + "rx> ConnAck(00):", + "tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(04):", + "rx> PubRec(00):", + "tx> PubRel(02):", + "rx> PubComp(00):", + "rx> Publish(04):", + "tx> PubRec(02):", + "rx> PubRel(02):", + "tx> PubComp(02):", + "tx> Unsubscribe(02):", + "rx> Unsuback(00):"])) waitFor conn() - test "stay subscribed after multipe (2) disconnect with reconnect": - let (tpc, msg) = tdata("stay subscribed after multipe (2) disconnect with reconnect") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("stay subscribed after multipe (2) disconnect with reconnect") proc conn() {.async.} = - await ctxSlave.start() - await sleepAsync(1000) - testDmp = @[] - var msgCount: int - proc on_data_sub_keep_multiple(topic: string, message: string) = + + proc onDataSubKeepMultiple(topic: string, message: string) = msgCount += 1 - await ctxSlave.subscribe(tpc, 0, on_data_sub_keep_multiple) - await sleepAsync 500 + await ctxListen.subscribe(tpc, 0, onDataSubKeepMultiple) + await sleepAsync(500) await ctxMain.publish(tpc, msg, 0) # msg 1 - await sleepAsync 500 + await sleepAsync(500) # Disconnect 1/2 - ctxSlave.state = Disconnecting - ctxSlave.s.close() + ctxListen.state = Disconnecting + ctxListen.s.close() await sleepAsync(500) - ctxSlave.state = Disconnected + ctxListen.state = Disconnected await sleepAsync(2000) # Auto-reconnect loop is 1000ms, wait 2000ms to ensure loop # We should automatic reconnect here # Disconnect 2/2 - ctxSlave.state = Disconnecting - ctxSlave.s.close() + ctxListen.state = Disconnecting + ctxListen.s.close() await sleepAsync(500) - ctxSlave.state = Disconnected + ctxListen.state = Disconnected await sleepAsync(2000) # Auto-reconnect loop is 1000ms, wait 2000ms to ensure loop # We should automatic reconnect here await ctxMain.publish(tpc, msg, 0) # msg 2 await ctxMain.publish(tpc, msg, 0) # msg 3 - await sleepAsync 500 - await ctxSlave.unsubscribe(tpc) - await sleepAsync 500 + await sleepAsync(500) + await ctxListen.unsubscribe(tpc) + await sleepAsync(500) check(msgCount == 3) # A total of 3 msgs on this topic - check(testDmp[0][0] == "tx> Subscribe(02):") - check(testDmp[1][0] == "rx> SubAck(00):") - check(testDmp[2][0] == "tx> Publish(00):") - check(testDmp[3][0] == "rx> Publish(00):") - - # Disconnected 1/2 - check(testDmp[4][0] == "tx> Connect(00):") - check(testDmp[5][0] == "rx> ConnAck(00):") - # Re-subscribe - check(testDmp[6][0] == "tx> Subscribe(02):") - check(testDmp[7][0] == "rx> SubAck(00):") - - # Disconnected 1/2 - check(testDmp[8][0] == "tx> Connect(00):") - check(testDmp[9][0] == "rx> ConnAck(00):") - # Re-subscribe - check(testDmp[10][0] == "tx> Subscribe(02):") - check(testDmp[11][0] == "rx> SubAck(00):") - - check(testDmp[12][0] == "tx> Publish(00):") - check(testDmp[13][0] == "tx> Publish(00):") - check(testDmp[14][0] == "rx> Publish(00):") - check(testDmp[15][0] == "rx> Publish(00):") - # Unsub - check(testDmp[16][0] == "tx> Unsubscribe(02):") - check(testDmp[17][0] == "rx> Unsuback(00):") - - await ctxSlave.disconnect() - await sleepAsync(1000) - waitFor conn() + check(hasAllInDmp(@["tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(00):", + "rx> Publish(00):", + "tx> Connect(00):", + "rx> ConnAck(00):", + "tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Connect(00):", + "rx> ConnAck(00):", + "tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(00):", + "tx> Publish(00):", + "rx> Publish(00):", + "rx> Publish(00):", + "tx> Unsubscribe(02):", + "rx> Unsuback(00):"])) + waitFor conn() test "stay subscribed after long disconnect with reconnect": ## This test currently needs manual actions - you need to close/disconnect @@ -456,32 +444,25 @@ suite "test suite for subscribe": echo "\n\nTHIS TEST NEEDS MANUAL ACTIONS - STAY READY\n\n" - let (tpc, msg) = tdata("stay subscribed after long disconnect with reconnect") + let + ctxMain = newCtx() + ctxSlave = newCtx() + (tpc, msg) = tdata("stay subscribed after long disconnect with reconnect") proc conn() {.async.} = - # Disconnect main clients to avoid interference with result - await sleepAsync(1000) - waitFor ctxMain.disconnect() - waitFor ctxListen.disconnect() - await sleepAsync(1000) - - await ctxSlave.start() - ctxSlave.set_ping_interval(90) # Increas ping to avoid interference with package order - await sleepAsync(1000) - testDmp = @[] + ctxSlave.setPingInterval(90) # Increas ping to avoid interference with package order var msgCount: int - proc on_data_sub_keep_long(topic: string, message: string) = + proc onDataSubKeepLong(topic: string, message: string) = msgCount += 1 - await ctxSlave.subscribe(tpc, 0, on_data_sub_keep_long) - await sleepAsync 500 + await ctxSlave.subscribe(tpc, 0, onDataSubKeepLong) + await sleepAsync(500) await ctxSlave.publish(tpc, msg, 0) # msg 1 - await sleepAsync 500 # Disconnect - echo "\n\nDISCONNECT THE BROKER NOW\n\n" - await sleepAsync(2000) + echo "\n\nDISCONNECT THE BROKER NOW (you have 5 sec)\n\n" + await sleepAsync(5000) # Publish messages while the broker is down await ctxSlave.publish(tpc, msg, 0) # msg 2 @@ -491,65 +472,40 @@ suite "test suite for subscribe": await sleepAsync(5000) # Reconnect the broker - echo "\n\nCONNECT THE BROKER NOW\n\n" - await sleepAsync(4000) + echo "\n\nCONNECT THE BROKER NOW (you have 5 sec)\n\n" + await sleepAsync(5000) # Send messages when the broker is up again await ctxSlave.publish(tpc, msg, 0) # msg 2 await ctxSlave.publish(tpc, msg, 0) # msg 3 - await sleepAsync 1500 + await sleepAsync(500) await ctxSlave.unsubscribe(tpc) - await sleepAsync 500 + await sleepAsync(500) check(msgCount == 7) - # Connect and send 1 messages - check(testDmp[0][0] == "tx> Subscribe(02):") - check(testDmp[1][0] == "rx> SubAck(00):") - check(testDmp[2][0] == "tx> Publish(00):") - check(testDmp[3][0] == "rx> Publish(00):") - - # Manual disconnect - check(testDmp[4][0] == "tx> Disconnect(00):") - - # Connect - check(testDmp[5][0] == "tx> Connect(00):") - check(testDmp[6][0] == "rx> ConnAck(00):") - - # Re-subscribe - check(testDmp[7][0] == "tx> Subscribe(02):") - - # Publish all messages in queue - check(testDmp[8][0] == "tx> Publish(00):") - check(testDmp[9][0] == "tx> Publish(00):") - check(testDmp[10][0] == "tx> Publish(00):") - check(testDmp[11][0] == "tx> Publish(00):") - - # Receive subscribe ack - check(testDmp[12][0] == "rx> SubAck(00):") - - # Receive all messages - check(testDmp[13][0] == "rx> Publish(00):") - check(testDmp[14][0] == "rx> Publish(00):") - check(testDmp[15][0] == "rx> Publish(00):") - check(testDmp[16][0] == "rx> Publish(00):") - - # Publish the 2 last messages - check(testDmp[17][0] == "tx> Publish(00):") - check(testDmp[18][0] == "tx> Publish(00):") - - # Receive the 2 last messages - check(testDmp[19][0] == "rx> Publish(00):") - check(testDmp[20][0] == "rx> Publish(00):") - - # Unsub - check(testDmp[21][0] == "tx> Unsubscribe(02):") - check(testDmp[22][0] == "rx> Unsuback(00):") - - # Reconnect main clients - await ctxSlave.disconnect() - await ctxMain.start() - await ctxListen.start() - await sleepAsync(1000) + check(hasAllInDmp(@["tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(00):", + "rx> Publish(00):", + "tx> Disconnect(00):", + "tx> Connect(00):", + "rx> ConnAck(00):", + "tx> Subscribe(02):", + "tx> Publish(00):", + "tx> Publish(00):", + "tx> Publish(00):", + "tx> Publish(00):", + "rx> SubAck(00):", + "rx> Publish(00):", + "rx> Publish(00):", + "rx> Publish(00):", + "rx> Publish(00):", + "tx> Publish(00):", + "tx> Publish(00):", + "rx> Publish(00):", + "rx> Publish(00):", + "tx> Unsubscribe(02):", + "rx> Unsuback(00):"])) waitFor conn() diff --git a/tests/tester.nim b/tests/tester.nim index a5ff8e1..879fd76 100644 --- a/tests/tester.nim +++ b/tests/tester.nim @@ -10,35 +10,27 @@ include "../nmqtt.nim" randomize() -# Test client main: -# ctxMain is an open connection, which can used in all the test. This -# connection is not to be closed. -let ctxMain = newMqttCtx("nmqttTestMain") -#ctxMain.set_host("test.mosquitto.org", 1883) -ctxMain.set_host("127.0.0.1", 1883) -ctxMain.set_ping_interval(1200) -waitFor ctxMain.start() - -# Test clíent slave: -# ctxSlave is a client which may be closed and open. It should be closed -# after each test. -let ctxSlave = newMqttCtx("nmqttTestSlave") -#ctxSlave.set_host("test.mosquitto.org", 1883) -ctxSlave.set_host("127.0.0.1", 1883) -ctxSlave.set_ping_interval(1200) - -# Test clíent listen: -# ctxListen is a client which only should be used to make subscribe -# callbacks. Do not close it. -let ctxListen = newMqttCtx("nmqttTestListen") -#ctxSlave.set_host("test.mosquitto.org", 1883) -ctxListen.set_host("127.0.0.1", 1883) -ctxMain.set_ping_interval(1200) -waitFor ctxListen.start() +proc newCtx(): MqttCtx = + result = newMqttCtx("nmqttTest-" & $genOid()) + result.setHost("127.0.0.1", 1883) + result.setPingInterval(1200) + waitFor result.start() proc tout(t, m, s: string) = ## Print test data during test. - stderr.write " \e[17m" & t & " - " & m & " - " & s & "\e[0m\n" + echo " \e[17m" & t & " - " & m & " - " & s & "\e[0m\n" + +proc hasAllInDmp(s: seq[string]): bool = + var t = s + for rec in testDmp: + var + j = 0 + while j < len(t): + if rec[0] == t[j]: + t.del(j) + break + inc j + len(t) == 0 proc tdata(t: string): (string, string) = ## Generate the test topic and message @@ -48,8 +40,6 @@ proc tdata(t: string): (string, string) = testDmp = @[] return (topicTest, msg) -waitFor sleepAsync(1500) # Let the clients connect - include "connection.nim" include "subscribe.nim" include "unsubscribe.nim" @@ -64,6 +54,3 @@ include "publish_retained.nim" # Contains retained msgs # it needs to be the last test, since it # will store msg's, which will be caught # in the subscribe test on `#`. - -waitFor ctxMain.disconnect() -waitFor ctxListen.disconnect() \ No newline at end of file diff --git a/tests/tools.nim b/tests/tools.nim index 6813436..2d98941 100644 --- a/tests/tools.nim +++ b/tests/tools.nim @@ -2,7 +2,9 @@ suite "test suite for messages": test "msgQueue() wait for all messages in workqueue": - let (tpc, msg) = tdata("msgQueue() wait for all messages in workqueue") + let + ctxMain = newCtx() + (tpc, msg) = tdata("msgQueue() wait for all messages in workqueue") proc conn() {.async.} = @@ -16,7 +18,4 @@ suite "test suite for messages": await sleepAsync(1000) check(ctxMain.msgQueue == 0) - await ctxListen.unsubscribe(tpc) - waitFor conn() - diff --git a/tests/unsubscribe.nim b/tests/unsubscribe.nim index 17a2f6a..d2bcb87 100644 --- a/tests/unsubscribe.nim +++ b/tests/unsubscribe.nim @@ -1,79 +1,87 @@ suite "test suite for unsubscribe": test "unsubscribe from topic": - let (tpc, msg) = tdata("unsubscribe from topic") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("unsubscribe from topic") proc conn() {.async.} = - proc on_data_unsub(topic: string, message: string) = + proc onDataUnsub(topic: string, message: string) = if topic == tpc: check(message == msg) return check(ctxListen.pubCallbacks.len() == 0) - await ctxListen.subscribe(tpc, 0, on_data_unsub) + await ctxListen.subscribe(tpc, 0, onDataUnsub) check(ctxListen.pubCallbacks.len() == 1) - await sleepAsync 500 + await sleepAsync(500) await ctxMain.publish(tpc, msg, 0) - await sleepAsync 500 + await sleepAsync(500) await ctxListen.unsubscribe(tpc) check(ctxListen.pubCallbacks.len() == 0) - await sleepAsync 500 + await sleepAsync(500) await ctxMain.publish(tpc, "Msg must not be received", 0) - await sleepAsync 500 - - check(testDmp[0][0] == "tx> Subscribe(02):") - check(testDmp[1][0] == "rx> SubAck(00):") # and testDmp[1][1] == "00 01 00 ") - check(testDmp[2][0] == "tx> Publish(00):") - check(testDmp[3][0] == "rx> Publish(00):") - check(testDmp[4][0] == "tx> Unsubscribe(02):") - check(testDmp[5][0] == "rx> Unsuback(00):") - check(testDmp[6][0] == "tx> Publish(00):") - check(testDmp.len() == 7) + await sleepAsync(500) + check(hasAllInDmp(@["tx> Subscribe(02):", + "rx> SubAck(00):", + "tx> Publish(00):", + "rx> Publish(00):", + "tx> Unsubscribe(02):", + "rx> Unsuback(00):", + "tx> Publish(00):"])) waitFor conn() test "unsubscribe from one of many topics": - let (tpc, msg) = tdata("unsubscribe from one of many topics") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc1, msg1) = tdata("unsubscribe from one of many topics") + (tpc2, msg2) = tdata("unsubscribe from one of many topics") proc conn() {.async.} = var + topic1: bool topic2: bool - proc on_data_unsub1(topic: string, message: string) = - check(message == msg) + proc onDataUnsub1(topic: string, message: string) = + check(message == msg1) + topic1 = true - proc on_data_unsub2(topic: string, message: string) = - check(message == msg) + proc onDataUnsub2(topic: string, message: string) = + check(message == msg2) topic2 = true check(ctxListen.pubCallbacks.len() == 0) - await ctxListen.subscribe(tpc & "-1", 0, on_data_unsub1) - await ctxListen.subscribe(tpc & "-2", 0, on_data_unsub2) + await ctxListen.subscribe(tpc1, 0, onDataUnsub1) + await ctxListen.subscribe(tpc2, 0, onDataUnsub2) check(ctxListen.pubCallbacks.len() == 2) - await sleepAsync 500 - await ctxListen.unsubscribe(tpc & "-2") + await sleepAsync(500) + await ctxListen.unsubscribe(tpc2) check(ctxListen.pubCallbacks.len() == 1) - await sleepAsync 500 - await ctxMain.publish(tpc & "-1", msg, 0) - await ctxMain.publish(tpc & "-2", msg, 0) - await sleepAsync 500 - await ctxListen.unsubscribe(tpc & "-1") + await sleepAsync(500) + await ctxMain.publish(tpc1, msg1, 0) + await ctxMain.publish(tpc2, msg2, 0) + await sleepAsync(500) + await ctxListen.unsubscribe(tpc1) - await sleepAsync 500 + await sleepAsync(500) check(ctxListen.pubCallbacks.len() == 0) + check(topic1 == true) check(topic2 == false) - waitFor conn() \ No newline at end of file + waitFor conn() diff --git a/tests/willmsg.nim b/tests/willmsg.nim index 4400bd6..254843b 100644 --- a/tests/willmsg.nim +++ b/tests/willmsg.nim @@ -2,33 +2,38 @@ suite "test suite for will messages": test "send will msg with default values": - let (tpc, msg) = tdata("send will msg with default values") + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, _) = tdata("send will msg with default values") proc conn() {.async.} = const willMsg = "willmsg_qos0_retain=false" var willMsgCheck: bool - proc on_data_will(topic: string, message: string) = + proc onDataWill(topic: string, message: string) = if topic == tpc and message == willMsg: willMsgCheck = true - await ctxListen.subscribe(tpc, 2, on_data_will) - await sleepAsync 500 + await ctxListen.subscribe(tpc, 2, onDataWill) + + ctxMain.setWill(tpc, willMsg) + await ctxMain.connect() + await sleepAsync(500) # Wait for full connection + ctxMain.s.close() + await sleepAsync(500) # Wait for willMsg to be sent - ctxSlave.set_will(tpc, willMsg) - await ctxSlave.connect() - await sleepAsync 500 # Wait for full connection - ctxSlave.s.close() - await sleepAsync 500 # Wait for willMsg to be sent check(willMsgCheck == true) - ctxSlave.willFlag = true waitFor conn() - test "send will msg retained = true": - let (tpc, msg) = tdata("send will msg retained = true") + let + ctxMain = newCtx() + ctxListen = newCtx() + ctxDestroy = newCtx() + (tpc, _) = tdata("send will msg retained = true") proc conn() {.async.} = @@ -37,37 +42,28 @@ suite "test suite for will messages": willMsgCheck: bool willMsgRetain: bool - proc on_data_will(topic: string, message: string) = + proc onDataWill(topic: string, message: string) = if topic == tpc and message == willMsg: willMsgCheck = true - await ctxListen.subscribe(tpc, 2, on_data_will) - await sleepAsync 500 + await ctxListen.subscribe(tpc, 2, onDataWill) # Set will and send - ctxSlave.set_will(tpc, willMsg, retain=true) - await ctxSlave.connect() - await sleepAsync 500 # Wait for full connection - ctxSlave.s.close() - await sleepAsync 500 # Wait for willMsg to be sent - check(willMsgCheck == true) + ctxMain.setWill(tpc, willMsg, retain=true) + await ctxMain.connect() + await sleepAsync(500) # Wait for full connection + ctxMain.s.close() + await sleepAsync(500) # Wait for willMsg to be sent - # Check that the message is retained. - # New client but on same topic. - let ctxDestroy = newMqttCtx("nmqttTestWill") - ctxDestroy.set_host("127.0.0.1", 1883) - await ctxDestroy.connect() + check(willMsgCheck == true) - proc on_data_will_retain(topic: string, message: string) = + proc onDataWillRetain(topic: string, message: string) = if topic == tpc and message == willMsg: willMsgRetain = true - await ctxDestroy.subscribe(tpc, 2, on_data_will_retain) - await sleepAsync 500 + await ctxDestroy.subscribe(tpc, 2, onDataWillRetain) + await sleepAsync(500) + check(willMsgRetain == true) - await ctxDestroy.unsubscribe(tpc) - await ctxDestroy.disconnect() - ctxDestroy.willFlag = true - await sleepAsync 500 - waitFor conn() \ No newline at end of file + waitFor conn()