From c5c507a16abc9166fb7b88d779536231f863e5cf Mon Sep 17 00:00:00 2001 From: Andrii Fil Date: Sat, 19 Sep 2026 10:33:06 +0300 Subject: [PATCH 1/2] timeout pingresp; ping if nit send --- nmqtt.nim | 79 ++++++++++++++++++++++++++++++-------------- tests/connection.nim | 15 +++++++++ tests/ping.nim | 42 +++++++++++++++++------ 3 files changed, 101 insertions(+), 35 deletions(-) diff --git a/nmqtt.nim b/nmqtt.nim index d0ac56e..71ce74c 100644 --- a/nmqtt.nim +++ b/nmqtt.nim @@ -2,12 +2,13 @@ ## ## zevv (https://github.com/zevv) & ThomasTJdev (https://github.com/ThomasTJdev) & python36 (https://github.com/python36) -import - strutils, - asyncnet, - net, - asyncDispatch, - tables +import std/[strutils, + asyncnet, + net, + asyncdispatch, + tables, + monotimes, + times] when defined(broker): import @@ -49,6 +50,11 @@ type inWork: bool hasNewWorks: bool keepAlive: uint16 + pingRunned: bool + lastSend: MonoTime + awaitingPingResp: bool + timeoutPingRespTime: MonoTime + pingRespTimeout: Duration maxInflightMessages: int willFlag: bool willQoS: QoS @@ -423,13 +429,15 @@ proc send(ctx: MqttCtx, pkt: Pkt): Future[bool] {.async.} = let hdrlen = buf.len buf.setLen(hdrlen + pkt.data.len) copyMem(buf[hdrlen].addr, pkt.data[0].unsafeAddr, pkt.data.len) + + ctx.lastSend = getMonoTime() await ctx.s.send(buf[0].unsafeAddr, buf.len) return true proc recv(ctx: MqttCtx): Future[Pkt] {.async.} = ## Receive and parse the packet - if ctx.state notin {Connecting,Connected}: + if ctx.state notin {Connecting, Connected}: return var r: int @@ -511,7 +519,7 @@ proc sendConnect(ctx: MqttCtx): Future[bool] = pkt.put("MQTT", true) pkt.put(4.uint8) pkt.put(flags) - pkt.put(ctx.keepAlive.uint16) + pkt.put(ctx.keepAlive) pkt.put(ctx.clientId, true) if ctx.willFlag: @@ -523,7 +531,6 @@ proc sendConnect(ctx: MqttCtx): Future[bool] = pkt.put(ctx.username, true) if ctx.password != "": pkt.put(ctx.password, true) - ctx.state = Connecting result = ctx.send(pkt) proc sendDisconnect(ctx: MqttCtx): Future[bool] = @@ -1069,7 +1076,7 @@ proc onPingReq(ctx: MqttCtx, pkt: Pkt) {.async.} = await ctx.work() proc onPingResp(ctx: MqttCtx, pkt: Pkt) {.async.} = - discard + ctx.awaitingPingResp = false proc handle(ctx: MqttCtx, pkt: Pkt) {.async.} = when defined(broker): @@ -1109,12 +1116,29 @@ proc runRx(ctx: MqttCtx) {.async.} = ctx.wrn("Boom, socket is closed") proc runPing(ctx: MqttCtx) {.async.} = + if ctx.pingRunned: + return + ctx.pingRunned = true while true: - await sleepAsync(ctx.keepAlive.int * 1000) - let ok = await ctx.sendPingReq() - if not ok: + await sleepAsync(1000) + + if ctx.state notin {Connecting, Connected, Disconnecting}: break - await ctx.work() + if ctx.state != Connected: + continue + + let now = getMonoTime() + if ctx.awaitingPingResp: + if now >= ctx.timeoutPingRespTime: + await ctx.close("ping response timeout") + break + + elif (now - ctx.lastSend).inSeconds >= int64(ctx.keepAlive): + discard await ctx.sendPingReq() + ctx.timeoutPingRespTime = now + ctx.pingRespTimeout + ctx.awaitingPingResp = true + + ctx.pingRunned = false proc connectBroker(ctx: MqttCtx) {.async.} = ## Connect to the broker. @@ -1124,16 +1148,21 @@ proc connectBroker(ctx: MqttCtx) {.async.} = if ctx.verbosity >= 1: ctx.dbg("Connecting to " & ctx.host & ":" & $ctx.port) - ctx.state = Error # set to Connecting by sendConnect + ctx.state = Connecting - ctx.s = await asyncnet.dial(ctx.host, ctx.port) - if ctx.sslOn: - when defined(ssl): - ctx.ssl = newContext(protSSLv23, CVerifyNone, ctx.sslCert, ctx.sslKey) - wrapConnectedSocket(ctx.ssl, ctx.s, handshakeAsClient) - else: - ctx.wrn("Requested SSL session but ssl is not enabled") - await ctx.close("SSL not enabled") + try: + ctx.s = await asyncnet.dial(ctx.host, ctx.port) + if ctx.sslOn: + when defined(ssl): + ctx.ssl = newContext(protSSLv23, CVerifyNone, ctx.sslCert, ctx.sslKey) + wrapConnectedSocket(ctx.ssl, ctx.s, handshakeAsClient) + else: + ctx.wrn("Requested SSL session but ssl is not enabled") + await ctx.close("SSL not enabled") + raise newException(SslError, "SSL session requested but compiler flag -d:ssl is missing") + except CatchableError as e: + ctx.state = Error + raise let ok = await ctx.sendConnect() if ok: @@ -1155,7 +1184,6 @@ proc runConnect(ctx: MqttCtx) {.async.} = ctx.dbg("Error connecting to " & ctx.host) if ctx.verbosity >= 2: echo e.msg - ctx.state = Error # If the client has been disconnect, it is necessary to tell the broker, # that we still want to be Subscribed. PubCallbacks still holds the @@ -1177,7 +1205,8 @@ proc runConnect(ctx: MqttCtx) {.async.} = proc newMqttCtx*(clientId: string): MqttCtx = ## Initiate a new MQTT client. - MqttCtx(clientId: clientId, state: Disconnected, maxInflightMessages: 20) + MqttCtx(clientId: clientId, state: Disconnected, maxInflightMessages: 20, + keepAlive: 60, pingRespTimeout: initDuration(seconds = 10)) proc setPingInterval*(ctx: MqttCtx, txInterval: int = 60) = ## Set the clients ping interval in seconds. Default is 60 seconds. diff --git a/tests/connection.nim b/tests/connection.nim index 8fb37bb..631b658 100644 --- a/tests/connection.nim +++ b/tests/connection.nim @@ -1,6 +1,21 @@ suite "test suite for connections": + test "connection error": + let (tpc, _) = tdata("connection non-exists broker") + proc conn() {.async.} = + let ctx = newMqttCtx("nmqttTestConn" & tpc) # unique clientid for public broker + ctx.setHost("localhost", 1993) + var hasError: bool + try: + await ctx.connect() + except OsError: + hasError = true + check(ctx.state == Error) + check(hasError == true) + + waitFor conn() + test "connection public broker": let (tpc, msg) = tdata("connection public broker") diff --git a/tests/ping.nim b/tests/ping.nim index 72f93fc..309b3de 100644 --- a/tests/ping.nim +++ b/tests/ping.nim @@ -7,17 +7,39 @@ suite "test suite for ping": (tpc, msg) = tdata("set ping interval") proc conn() {.async.} = + var + pingCount: int + pingResp: int + + proc empty(topic: string, message: string) = + discard + + proc findPings() = + for i in testDmp: + if i[0] == "tx> PingReq(00):": pingCount += 1 + if i[0] == "rx> PingResp(00):": pingResp += 1 ctxMain.setPingInterval(1) await ctxMain.connect() + + for i in 1 .. 4: + await sleepAsync(500) + await ctxMain.subscribe(tpc, 0, empty) + await sleepAsync(500) + await ctxMain.publish(tpc, msg, 0) + await sleepAsync(500) + await ctxMain.unsubscribe(tpc) + + findPings() + checkpoint("Ping only if no other messages were sent") + check(pingCount == 0) + check(pingCount == 0) + await sleepAsync(6000) - var - pingCount: int - pingResp: int - for ping in testDmp: - if ping[0] == "tx> PingReq(00):": pingCount += 1 - if ping[0] == "rx> PingResp(00):": pingResp += 1 + for i in testDmp: + if i[0] == "tx> PingReq(00):": pingCount += 1 + if i[0] == "rx> PingResp(00):": pingResp += 1 checkpoint("Ping with 1 second interval during 6 seconds") check(pingCount > 3) @@ -33,12 +55,12 @@ suite "test suite for ping": pingCount = 0 pingResp = 0 - for ping in testDmp: - if ping[0] == "tx> PingReq(00):": pingCount += 1 - if ping[0] == "rx> PingResp(00):": pingResp += 1 + for i in testDmp: + if i[0] == "tx> PingReq(00):": pingCount += 1 + if i[0] == "rx> PingResp(00):": pingResp += 1 checkpoint("Ping with 60 second interval during 6 seconds") check(pingCount == 0) check(pingResp == 0) - waitFor conn() \ No newline at end of file + waitFor conn() From ad699c284c7c624a0eed99deb181f97f87f0caf6 Mon Sep 17 00:00:00 2001 From: Andrii Fil Date: Thu, 24 Sep 2026 13:07:11 +0300 Subject: [PATCH 2/2] fix: sslerror to ioerror --- nmqtt.nim | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nmqtt.nim b/nmqtt.nim index 71ce74c..d2d2b87 100644 --- a/nmqtt.nim +++ b/nmqtt.nim @@ -1159,7 +1159,7 @@ proc connectBroker(ctx: MqttCtx) {.async.} = else: ctx.wrn("Requested SSL session but ssl is not enabled") await ctx.close("SSL not enabled") - raise newException(SslError, "SSL session requested but compiler flag -d:ssl is missing") + raise newException(IOError, "SSL session requested but compiler flag -d:ssl is missing") except CatchableError as e: ctx.state = Error raise