diff --git a/nmqtt.nim b/nmqtt.nim index fe462c5..4e0d88f 100644 --- a/nmqtt.nim +++ b/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,11 +1130,10 @@ 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() @@ -1132,18 +1141,27 @@ proc runPing(ctx: MqttCtx) {.async.} = 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. diff --git a/nmqtt.nimble b/nmqtt.nimble index f9837fa..eb7ab05 100644 --- a/nmqtt.nimble +++ b/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" diff --git a/nmqtt/nmqtt.nim b/nmqtt/nmqtt.nim index f20b2cf..8c9e486 100644 --- a/nmqtt/nmqtt.nim +++ b/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 diff --git a/tests/ping.nim b/tests/ping.nim index 309b3de..4517c16 100644 --- a/tests/ping.nim +++ b/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()