willFlag; keepAliveMonitor; handle ping response; test keep alive

nmqtt.nim

@@ -49,7 +49,9 @@ type
     pubCallbacks: Table[string, TopicCb]
     inWork: bool
     hasNewWorks: bool
-    keepAlive: uint16
+    keepAlive: Duration
+    keepAliveTimeout: Duration
+    keepAliveMonitorRunned: bool
     pingRunned: bool
     lastSend: MonoTime
     awaitingPingResp: bool
@@ -67,7 +69,7 @@ type
     version: uint8
     connFlags: string
     subscribed: Table[string, QoS] # Topic, QoS
-    lastAction: float # Check keepAlive
+    lastAction: MonoTime # Check keepAlive
 
   #when defined(broker):
   ConnAckFlag = enum
@@ -324,7 +326,7 @@ proc splitTopic(topic: string): seq[string] =
 
 when defined(broker):
   proc retainMsg(topic: string, message: string, qos: QoS, clientid: string) =
-    mqttbroker.retained[topic] = RetainedMsg(msg: message, qos: qos, time: epochTime(), clientid: clientid)
+    mqttbroker.retained[topic] = RetainedMsg(msg: message, qos: qos, clientid: clientid)
 
 when defined(broker):
   proc addSubscriber*(ctx: MqttCtx, topic: string) {.async.} =
@@ -519,7 +521,7 @@ proc sendConnect(ctx: MqttCtx): Future[bool] =
   pkt.put("MQTT", true)
   pkt.put(4.uint8)
   pkt.put(flags)
-  pkt.put(ctx.keepAlive)
+  pkt.put(uint16(ctx.keepAlive.inSeconds))
   pkt.put(ctx.clientId, true)
 
   if ctx.willFlag:
@@ -715,14 +717,19 @@ proc work(ctx: MqttCtx) {.async.} =
 when defined(broker):
   proc sendWill(ctx: MqttCtx) {.async.} =
     ## Send the will
-    if ctx.willTopic != "":
-      if ctx.willRetain:
-        retainMsg(ctx.willTopic, ctx.willMsg, ctx.willQoS, ctx.clientid)
-      for c in mqttbroker.subscribers[ctx.willTopic].ctxs:
-        let msgId = c.nextMsgId()
-        let qos = qosAlign(ctx.willQoS, c.subscribed[ctx.willTopic])
-        c.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: ctx.willTopic, qos: qos, retain: ctx.willRetain, message: ctx.willMsg, typ: Publish)
-        await c.work()
+    if not ctx.willFlag:
+      return
+    if ctx.willRetain:
+      retainMsg(ctx.willTopic, ctx.willMsg, ctx.willQoS, ctx.clientid)
+
+    if not mqttbroker.subscribers.hasKey(ctx.willTopic):
+      return
+    for c in mqttbroker.subscribers[ctx.willTopic].ctxs:
+      let msgId = c.nextMsgId()
+      let qos = qosAlign(ctx.willQoS, c.subscribed[ctx.willTopic])
+      c.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: ctx.willTopic, qos: qos,
+                                retain: ctx.willRetain, message: ctx.willMsg, typ: Publish)
+      await c.work()
 
 when defined(broker):
   proc publishToSubscribers(seqctx: seq[MqttCtx], pkt: Pkt, matchedTopic, topic, message: string, qos: QoS, retain: bool, senderId: string) {.async.} =
@@ -756,19 +763,24 @@ proc onConnect(ctx: MqttCtx, pkt: Pkt) {.async.} =
     var
       offset: int
       nextLen: uint16
+      keepAlive: uint16
 
     # Main data
     (ctx.proto, offset)     = pkt.getstring(0, true)
     (ctx.version, offset)   = pkt.getu8(offset)
     (ctx.connFlags, offset) = getbin(pkt, offset)
-    (ctx.keepAlive, offset) = pkt.getu16(offset)
+    (keepAlive, offset)     = pkt.getu16(offset)
+    ctx.keepAlive           = initDuration(seconds = int64(keepAlive))
+    ctx.keepAliveTimeout    = initDuration(milliseconds = int64(float(ctx.keepAlive.inMilliseconds) * 1.5))
     (nextLen, offset)       = pkt.getu16(offset)
     (ctx.clientId, offset)  = pkt.getstring(offset, parseInt($nextLen))
     if not mqttbroker.spacesInClientId:
       ctx.clientid = ctx.clientid.replace(" ", "")
 
     # Will Topic
-    if ctx.connFlags[5] == '1':
+    ctx.willFlag = ctx.connFlags[5] == '1'
+    if ctx.willFlag:
+      ctx.willFlag = true
       (nextLen, offset)         = pkt.getu16(offset)
       (ctx.willTopic, offset)   = pkt.getstring(offset,  parseInt($nextLen))
       (nextLen, offset)         = pkt.getu16(offset)
@@ -1058,8 +1070,6 @@ proc onDisconnect(ctx: MqttCtx, pkt: Pkt) {.async.} =
   when not defined(broker):
     ctx.wrn("Packet type only supported for broker: " & $pkt.typ)
   else:
-    #await removeSubscriber(ctx)
-    #await sendWill(ctx)
     ctx.state = Disconnected
     if mqttbroker.verbosity >= 1:
       verbose("Connections >> " & ctx.clientid & " has disconnected")
@@ -1080,7 +1090,7 @@ proc onPingResp(ctx: MqttCtx, pkt: Pkt) {.async.} =
 
 proc handle(ctx: MqttCtx, pkt: Pkt) {.async.} =
   when defined(broker):
-    ctx.lastAction = epochTime()
+    ctx.lastAction = getMonoTime()
 
   case pkt.typ
     of ConnAck: await ctx.onConnAck(pkt)
@@ -1120,30 +1130,38 @@ proc runPing(ctx: MqttCtx) {.async.} =
     return
   ctx.pingRunned = true
   while true:
-    await sleepAsync(1000)
-
     if ctx.state notin {Connecting, Connected, Disconnecting}:
       break
     if ctx.state != Connected:
+      await sleepAsync(500)
       continue
 
     let now = getMonoTime()
     if ctx.awaitingPingResp:
       if now >= ctx.timeoutPingRespTime:
         await ctx.close("ping response timeout")
         break
+      await sleepAsync(500)
+
+    else:
+      let ms = (ctx.keepAlive - (now - ctx.lastSend)).inMilliseconds
+      if ms < 0:
+        discard await ctx.sendPingReq()
+        ctx.timeoutPingRespTime = now + ctx.pingRespTimeout
+        ctx.awaitingPingResp = true
 
-    elif (now - ctx.lastSend).inSeconds >= int64(ctx.keepAlive):
-      discard await ctx.sendPingReq()
-      ctx.timeoutPingRespTime = now + ctx.pingRespTimeout
-      ctx.awaitingPingResp = true
+      elif ms < 1000:
+        await sleepAsync(int(ms))
+
+      else:
+        await sleepAsync(1000)
 
   ctx.pingRunned = false
 
 proc connectBroker(ctx: MqttCtx) {.async.} =
   ## Connect to the broker.
-  if ctx.keepAlive == 0:
-    ctx.keepAlive = 60
+  if ctx.keepAlive == DurationZero:
+    ctx.keepAlive = initDuration(seconds = 60)
 
   if ctx.verbosity >= 1:
     ctx.dbg("Connecting to " & ctx.host & ":" & $ctx.port)
@@ -1206,12 +1224,12 @@ proc runConnect(ctx: MqttCtx) {.async.} =
 proc newMqttCtx*(clientId: string): MqttCtx =
   ## Initiate a new MQTT client.
   MqttCtx(clientId: clientId, state: Disconnected, maxInflightMessages: 20,
-          keepAlive: 60, pingRespTimeout: initDuration(seconds = 10))
+          keepAlive: initDuration(seconds = 60), pingRespTimeout: initDuration(seconds = 10))
 
 proc setPingInterval*(ctx: MqttCtx, txInterval: int = 60) =
   ## Set the clients ping interval in seconds. Default is 60 seconds.
   if txInterval > 0 and txInterval < 65535:
-    ctx.keepAlive = txInterval.uint16
+    ctx.keepAlive = initDuration(seconds = int64(txInterval))
 
 proc setHost*(ctx: MqttCtx, host: string, port: int = 1883, sslOn = false) =
   ## Set the MQTT host.

nmqtt.nimble

@@ -1,5 +1,5 @@
 # Package
-version       = "1.1.1"
+version       = "1.1.2"
 author        = "zevv & ThomasTJdev & python36"
 description   = "Native MQTT library and binaries for publishing, subscribing and broker"
 license       = "MIT"

nmqtt/nmqtt.nim

@@ -13,18 +13,21 @@
 include "../nmqtt.nim"
 
 proc keepAliveMonitor(ctx: MqttCtx) {.async.} =
-  ctx.lastAction = epochTime()
   # The keep alive time is one and a half times the Keep Alive periode [MQTT-3.1.2-24]
-  let keepAlive = toFloat(ctx.keepAlive.int) * 1.5 * 1000
-  while ctx.state in [Connecting, Connected]:
-    let saveLastAction = ctx.lastAction
-    await sleepAsync(keepAlive)
-    if ctx.lastAction <= saveLastAction:
+  if ctx.keepAliveMonitorRunned:
+    return
+  ctx.keepAliveMonitorRunned = true
+  while true:
+    await sleepAsync(1000)
+    if ctx.state notin {Connecting, Connected}:
+      break
+    if getMonoTime() - ctx.lastAction > ctx.keepAliveTimeout:
       ctx.state = Error
       ctx.s.close()
       if mqttbroker.verbosity >= 1:
         verbose("Connections >> " & ctx.clientid & " was disconnected. Keep alive time overdue.")
       break
+  ctx.keepAliveMonitorRunned = false
 
 proc processClient(s: AsyncSocket) {.async.} =
   ## Create new client

tests/ping.nim

@@ -64,3 +64,27 @@ suite "test suite for ping":
       check(pingResp == 0)
 
     waitFor conn()
+
+  test "keepalive timeout":
+    let
+      ctxMain = newCtx()
+
+    proc conn() {.async.} =
+      await sleepAsync(500)
+      await ctxMain.disconnect()
+      await sleepAsync(500)
+
+      ctxMain.setPingInterval(1)
+      await ctxMain.connect()
+
+      await sleepAsync(500)
+      ctxMain.setPingInterval(30)
+
+      var disconnected = not ctxMain.isConnected()
+      for i in 1 .. 15:
+        await sleepAsync(1000)
+        if not ctxMain.isConnected():
+          break
+      check(ctxMain.isConnected() == false)
+
+    waitFor conn()