diff --git a/README.md b/README.md index f07649c..ce2c490 100644 --- a/README.md +++ b/README.md @@ -264,7 +264,7 @@ ____ ### setHost* ```nim -proc setHost*(ctx: MqttCtx, host: string, port: int=1883, sslOn=false) = +proc setHost*(ctx: MqttCtx, host: string, port: int = 1883, sslOn = false) = ``` Set the MQTT host. @@ -286,7 +286,7 @@ ____ ### setWill* ```nim -proc setWill*(ctx: MqttCtx, topic, msg: string, qos=0, retain=false) = +proc setWill*(ctx: MqttCtx, topic, msg: string, qos = 0, retain = false) = ``` Set the clients will. @@ -345,7 +345,7 @@ ____ ### publish* ```nim -proc publish*(ctx: MqttCtx, topic: string, message: string, qos=0, retain=false) {.async.} = +proc publish*(ctx: MqttCtx, topic: string, message: string, qos = 0, retain = false) {.async.} = ``` Publish a message. @@ -376,7 +376,7 @@ ____ ### subscribe* ```nim -proc subscribe*(ctx: MqttCtx, topic: string, qos: int, callback: PubCallback): Future[void] = +proc subscribe*(ctx: MqttCtx, topic: string, qos: QoS, callback: PubCallback): Future[void] = ``` Subscribe to a topic. diff --git a/config/nmqtt.conf b/config/nmqtt.conf index 6f63b63..ce947b3 100644 --- a/config/nmqtt.conf +++ b/config/nmqtt.conf @@ -46,7 +46,7 @@ clientid_spaces = false # If empty client id's is allowed. If it is set to `true`, the client will be # assigned a random integer as the client id. # -clientid_empty = false +clientid_empty = true # # If set to true the publishers client id will be inserted in front of the diff --git a/nmqtt.nim b/nmqtt.nim index 8ef008d..d0ac56e 100644 --- a/nmqtt.nim +++ b/nmqtt.nim @@ -21,6 +21,14 @@ when defined(broker): from os import fileExists type + QoS* = range[0..2] + + PubCallback = object + cb: proc(topic: string, message: string) + qos: QoS + + TopicCb = tuple[templ: seq[string], cb: PubCallback] + MqttCtx* = ref object host: string port: Port @@ -37,13 +45,13 @@ type ssl: SslContext msgIdSeq: MsgId workQueue: OrderedTable[MsgId, Work] - pubCallbacks: Table[string, PubCallback] + pubCallbacks: Table[string, TopicCb] inWork: bool hasNewWorks: bool keepAlive: uint16 maxInflightMessages: int willFlag: bool - willQoS: uint8 + willQoS: QoS willRetain: bool willTopic: string willMsg: string @@ -52,8 +60,7 @@ type proto: string version: uint8 connFlags: string - retained: seq[string] - subscribed: Table[string, uint8] # Topic, Qos + subscribed: Table[string, QoS] # Topic, QoS lastAction: float # Check keepAlive #when defined(broker): @@ -70,8 +77,6 @@ type MsgId = uint16 - Qos = range[0..2] - PktType = enum Notype = 0 Connect = 1 @@ -110,15 +115,11 @@ type WorkState = enum WorkNew, WorkSent, WorkAcked - PubCallback = object - cb: proc(topic: string, message: string) - qos: int - Work = ref object state: WorkState msgId: MsgId topic: string - qos: Qos + qos: QoS typ: PktType flags: uint16 #when defined(broker) case wk: WorkKind @@ -128,11 +129,9 @@ type of SubWork: discard - when defined(broker): type - MqttSub* = ref object ## Managing the subscribers - subscribers: Table[string, seq[MqttCtx]] + TopicCtx = tuple[templ: seq[string], ctxs: seq[MqttCtx]] MqttBroker* = ref object host: string @@ -143,7 +142,7 @@ when defined(broker): verbosity: int connections: Table[string, MqttCtx] retained: Table[string, RetainedMsg] # Topic, RetaindMsg - subscribers: Table[string, seq[MqttCtx]] + subscribers: Table[string, TopicCtx] version: uint8 clientIdMaxLen: int clientKickOld: bool @@ -155,7 +154,7 @@ when defined(broker): RetainedMsg = object msg: string - qos: uint8 + qos: QoS time: float clientid: string @@ -164,28 +163,31 @@ when defined(broker): mqttbroker = MqttBroker() r = initRand(toInt(epochTime())) - # # Packet helpers # proc put(pkt: var Pkt, v: uint16) = - pkt.data.add (v.int /% 256).uint8 - pkt.data.add (v.int mod 256).uint8 + pkt.data.add((v.int /% 256).uint8) + pkt.data.add((v.int mod 256).uint8) proc put(pkt: var Pkt, v: uint8) = - pkt.data.add v + pkt.data.add(v) proc put(pkt: var Pkt, data: string, withLen: bool) = if withLen: - pkt.put data.len.uint16 + pkt.put(data.len.uint16) for c in data: - pkt.put c.uint8 + pkt.put(c.uint8) proc getu8(pkt: Pkt, offset: int): (uint8, int) = let val = pkt.data[offset] result = (val, offset+1) +proc getQoS(pkt: Pkt, offset: int): (QoS, int) = + let val = pkt.data[offset] + result = (QoS(val), offset+1) + proc getu16(pkt: Pkt, offset: int): (uint16, int) = let val = (pkt.data[offset].int*256 + pkt.data[offset+1].int).uint16 result = (val, offset+2) @@ -195,18 +197,18 @@ proc getstring(pkt: Pkt, offset: int, withLen: bool): (string, int) = if withLen: var (len, offset2) = pkt.getu16(offset) for i in 0..= 2: - stderr.write "\e[1;30m" & s & "\e[0m\n" + stderr.write("\e[1;30m" & s & "\e[0m\n") when defined(broker): if defined(dev) or mqttbroker.verbosity >= 2: - stderr.write "\e[1;30m" & s & "\e[0m\n" + stderr.write("\e[1;30m" & s & "\e[0m\n") when defined(test): let s = split(s, " ") testDmp.add(@[$(s[0] & " " & s[1]), $join(s[2..s.len-1], " ")]) proc dbg(ctx: MqttCtx, s: string) = - stderr.write "\e[37m" & s & "\e[0m\n" + stderr.write("\e[37m" & s & "\e[0m\n") proc verbose(s: string) = - stderr.write "\e[37m" & s & "\e[0m\n" + stderr.write("\e[37m" & s & "\e[0m\n") when defined(broker): proc verbose(e: string, s: Table) = @@ -252,8 +254,8 @@ when defined(broker): when c is RetainedMsg: output.add("{" & t & "}") else: - output.add("{" & t & ": " & $c.len & "}") - stderr.write "\e[37m" & e & " >> " & output & "\e[0m\n" + output.add("{" & t & ": " & $c.ctxs.len & "}") + stderr.write("\e[37m" & e & " >> " & output & "\e[0m\n") proc verbose(ctx: auto) = var output: string @@ -269,28 +271,66 @@ when defined(broker): output.add(" " & t & ": " & $c & "\n") when ctx is MqttBroker: - stderr.write "\e[37m" & "Broker >>\n" & output & "\e[0m\n" + stderr.write("\e[37m" & "Broker >>\n" & output & "\e[0m\n") when ctx is MqttCtx: - stderr.write "\e[37m" & "Client >> " & ctx.clientid & "\n" & output & "\e[0m\n" + stderr.write("\e[37m" & "Client >> " & ctx.clientid & "\n" & output & "\e[0m\n") proc wrn(ctx: MqttCtx, s: string) = - stderr.write "\e[1;31mWarning >> " & s & "\e[0m\n" + stderr.write("\e[1;31mWarning >> " & s & "\e[0m\n") proc wrn(s: string) = - stderr.write "\e[1;31mWarning >> " & s & "\e[0m\n" + stderr.write("\e[1;31mWarning >> " & s & "\e[0m\n") # # Subscribers # +proc matchTopic(templ: seq[string], topic: string): bool {.inline.} = + var i = 0 + for t in templ: + if t == "#": + return i == 0 or i >= topic.len or topic[i] == '/' + + elif i > topic.len: + return false + + else: + if i > 0: + while i < topic.len: + if topic[i] == '/': + break + inc i + + if i + t.len > topic.len or t != topic[i ..< i + t.len]: + return false + i += t.len + + if i == topic.len: + return true + +proc splitTopic(topic: string): seq[string] = + if topic == "#": + @["#"] + elif topic[^1] == '#': + topic[0..^3].split('+') & @["#"] + else: + topic.split('+') + +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) + when defined(broker): proc addSubscriber*(ctx: MqttCtx, topic: string) {.async.} = ## Adds a subscriber to MqttBroker try: if mqttbroker.subscribers.hasKey(topic): - mqttbroker.subscribers[topic].insert(ctx) + for c in mqttbroker.subscribers[topic].ctxs: + if c.clientId == ctx.clientId: + return + mqttbroker.subscribers[topic].ctxs.insert(ctx) else: - mqttbroker.subscribers[topic] = @[ctx] + mqttbroker.subscribers[topic] = (templ: splitTopic(topic), ctxs: @[ctx]) except: wrn("Crash when adding a new subcriber") @@ -299,7 +339,7 @@ when defined(broker): ## Removes a subscriber from specific topic try: if mqttbroker.subscribers.hasKey(topic): - mqttbroker.subscribers[topic] = filter(mqttbroker.subscribers[topic], proc(x: MqttCtx): bool = x != ctx) + mqttbroker.subscribers[topic].ctxs = filter(mqttbroker.subscribers[topic].ctxs, proc(x: MqttCtx): bool = x != ctx) except: wrn("Crash when removing subscriber with specific topic") @@ -307,18 +347,18 @@ when defined(broker): proc removeSubscriber*(ctx: MqttCtx) {.async.} = ## Removes a subscriber without knowing the topics var delTop: seq[string] - for t, c in mqttbroker.subscribers: - if ctx in c: - mqttbroker.subscribers[t] = filter(c, proc(x: MqttCtx): bool = x != ctx) + for t, c in mqttbroker.subscribers.mpairs(): + if ctx in c.ctxs: + c.ctxs = filter(c.ctxs, proc(x: MqttCtx): bool = x != ctx) - if mqttbroker.subscribers[t].len() == 0: + if c.ctxs.len() == 0: delTop.add(t) for t in delTop: mqttbroker.subscribers.del(t) when defined(broker): - proc qosAlign(qP, qS: uint8): uint8 = + proc qosAlign(qP, qS: QoS): QoS = ## Aligns the QOS for publisher and subscriber. if qP == qS: result = qP @@ -353,7 +393,7 @@ proc close(ctx: MqttCtx, reason: string) {.async.} = if ctx.state in {Connecting, Connected}: ctx.state = Disconnecting if ctx.verbosity >= 1: - ctx.dbg "Closing: " & reason + ctx.dbg("Closing: " & reason) discard await ctx.sendDisconnect() ctx.s.close() ctx.state = Disconnected @@ -372,11 +412,13 @@ proc send(ctx: MqttCtx, pkt: Pkt): Future[bool] {.async.} = len = len div 128 if len > 0: b = b or 128 - buf.add b.uint8 + buf.add(b.uint8) if len == 0: break - ctx.dmp "tx> " & $pkt + # TODO: + # only when test + ctx.dmp("tx> " & $pkt) if pkt.data.len > 0: let hdrlen = buf.len buf.setLen(hdrlen + pkt.data.len) @@ -428,7 +470,7 @@ proc recv(ctx: MqttCtx): Future[Pkt] {.async.} = break if len > 0: - pkt.data.setlen len + pkt.data.setLen(len) var offset = 0 while offset < len: @@ -445,7 +487,7 @@ proc recv(ctx: MqttCtx): Future[Pkt] {.async.} = await ctx.close("remote closed connection") return - ctx.dmp "rx> " & $pkt + ctx.dmp("rx> " & $pkt) return pkt proc sendConnect(ctx: MqttCtx): Future[bool] = @@ -466,21 +508,21 @@ proc sendConnect(ctx: MqttCtx): Future[bool] = flags = flags or PasswordFlag.uint8 var pkt = newPkt(Connect) - pkt.put "MQTT", true - pkt.put 4.uint8 - pkt.put flags - pkt.put ctx.keepAlive.uint16 - pkt.put ctx.clientId, true + pkt.put("MQTT", true) + pkt.put(4.uint8) + pkt.put(flags) + pkt.put(ctx.keepAlive.uint16) + pkt.put(ctx.clientId, true) if ctx.willFlag: - pkt.put (ctx.willTopic.len).uint16 - pkt.put ctx.willTopic, false - pkt.put (ctx.willMsg.len).uint16 - pkt.put ctx.willMsg, false + pkt.put((ctx.willTopic.len).uint16) + pkt.put(ctx.willTopic, false) + pkt.put((ctx.willMsg.len).uint16) + pkt.put(ctx.willMsg, false) if ctx.username != "": - pkt.put ctx.username, true + pkt.put(ctx.username, true) if ctx.password != "": - pkt.put ctx.password, true + pkt.put(ctx.password, true) ctx.state = Connecting result = ctx.send(pkt) @@ -488,48 +530,57 @@ proc sendDisconnect(ctx: MqttCtx): Future[bool] = let pkt = newPkt(Disconnect, 0) result = ctx.send(pkt) -proc sendSubscribe(ctx: MqttCtx, msgId: MsgId, topic: string, qos: Qos): Future[bool] = +proc sendSubscribe(ctx: MqttCtx, msgId: MsgId, topic: string, qos: QoS): Future[bool] = var pkt = newPkt(Subscribe, 0b0010) - pkt.put msgId.uint16 - pkt.put topic, true - pkt.put qos.uint8 + pkt.put(msgId.uint16) + pkt.put(topic, true) + pkt.put(qos.uint8) result = ctx.send(pkt) proc sendUnsubscribe(ctx: MqttCtx, msgId: MsgId, topic: string): Future[bool] = var pkt = newPkt(Unsubscribe, 0b0010) - pkt.put msgId.uint16 - pkt.put topic, true + pkt.put(msgId.uint16) + pkt.put(topic, true) result = ctx.send(pkt) -proc sendPublish(ctx: MqttCtx, msgId: MsgId, topic: string, message: string, qos: Qos, retain: bool): Future[bool] = - var flags = (qos shl 1).uint8 +proc sendPublish(ctx: MqttCtx, msgId: MsgId, topic: string, message: string, qos: QoS, retain: bool): Future[bool] = + var flags = (qos.uint8 shl 1).uint8 if retain: flags = flags or 1 var pkt = newPkt(Publish, flags) - pkt.put topic, true + pkt.put(topic, true) if qos > 0: - pkt.put msgId.uint16 - pkt.put message, false + pkt.put(msgId.uint16) + pkt.put(message, false) result = ctx.send(pkt) -proc sendPubAck(ctx: MqttCtx, msgId: MsgId): Future[bool] = - var pkt = newPkt(PubAck, 0b0010) - pkt.put msgId.uint16 +proc sendPubAck(ctx: MqttCtx, msgId: MsgId, qos: QoS): Future[bool] = + when defined(broker): + var pkt = newPkt(PubAck) + else: + var pkt = newPkt(PubAck, 2) + pkt.put(msgId.uint16) result = ctx.send(pkt) proc sendPubRec(ctx: MqttCtx, msgId: MsgId): Future[bool] = - var pkt = newPkt(PubRec, 0b0010) - pkt.put msgId.uint16 + when defined(broker): + var pkt = newPkt(PubRec) + else: + var pkt = newPkt(PubRec, 2) + pkt.put(msgId.uint16) result = ctx.send(pkt) proc sendPubRel(ctx: MqttCtx, msgId: MsgId): Future[bool] = var pkt = newPkt(PubRel, 0b0010) - pkt.put msgId.uint16 + pkt.put(msgId.uint16) result = ctx.send(pkt) proc sendPubComp(ctx: MqttCtx, msgId: MsgId): Future[bool] = - var pkt = newPkt(PubComp, 0b0010) - pkt.put msgId.uint16 + when defined(broker): + var pkt = newPkt(PubComp) + else: + var pkt = newPkt(PubComp, 2) + pkt.put(msgId.uint16) result = ctx.send(pkt) proc sendPingReq(ctx: MqttCtx): Future[bool] = @@ -539,19 +590,20 @@ proc sendPingReq(ctx: MqttCtx): Future[bool] = #when defined(broker): proc sendConnAck(ctx: MqttCtx, flags: uint16): Future[bool] = var pkt = newPkt(ConnAck) - pkt.put flags.uint16 + pkt.put(flags.uint16) result = ctx.send(pkt) #when defined(broker): -proc sendSubAck(ctx: MqttCtx, msgId: MsgId): Future[bool] = - var pkt = newPkt(SubAck, 0b0010) - pkt.put msgId.uint16 +proc sendSubAck(ctx: MqttCtx, msgId: MsgId, message: string): Future[bool] = + var pkt = newPkt(SubAck) + pkt.put(msgId.uint16) + pkt.put(message, false) result = ctx.send(pkt) #when defined(broker): proc sendUnsubAck(ctx: MqttCtx, msgId: MsgId): Future[bool] = - var pkt = newPkt(Unsuback, 0b0010) - pkt.put msgId.uint16 + var pkt = newPkt(Unsuback) + pkt.put(msgId.uint16) result = ctx.send(pkt) #when defined(broker): @@ -568,7 +620,7 @@ proc sendWork(ctx: MqttCtx, work: Work): Future[bool] = result = ctx.sendPubRel(work.msgId) of PubAck: # Subscribe qos=1 (activated from a Publish) - result = ctx.sendPubAck(work.msgId) + result = ctx.sendPubAck(work.msgId, work.qos) of PubRec: # Subscribe qos=2 (1/2) (activated from a Publish) result = ctx.sendPubRec(work.msgId) @@ -592,7 +644,7 @@ proc sendWork(ctx: MqttCtx, work: Work): Future[bool] = of SubAck: #when defined(broker): - result = ctx.sendSubAck(work.msgId) + result = ctx.sendSubAck(work.msgId, work.message) of Unsuback: #when defined(broker): @@ -616,14 +668,14 @@ proc work(ctx: MqttCtx) {.async.} = #when defined(broker): if work.typ in [ConnAck, SubAck, UnsubAck, PingResp]: if await ctx.sendWork(work): - ctx.workQueue.del msgId + ctx.workQueue.del(msgId) continue if work.wk == PubWork and work.state == WorkNew: if work.typ == Publish: if work.qos == 0: if await ctx.sendWork(work): - ctx.workQueue.del msgId + ctx.workQueue.del(msgId) elif hasInflightSlots(ctx): if await ctx.sendWork(work): @@ -631,11 +683,11 @@ proc work(ctx: MqttCtx) {.async.} = elif work.typ == PubAck and work.qos == 1: if await ctx.sendWork(work): - ctx.workQueue.del msgId + ctx.workQueue.del(msgId) elif work.typ == PubComp and work.qos == 2: if await ctx.sendWork(work): - ctx.workQueue.del msgId + ctx.workQueue.del(msgId) else: if await ctx.sendWork(work): @@ -649,7 +701,7 @@ proc work(ctx: MqttCtx) {.async.} = elif work.typ == Unsubscribe: if await ctx.sendWork(work): work.state = WorkSent - ctx.pubCallbacks.del work.topic + ctx.pubCallbacks.del(work.topic) ctx.inWork = false @@ -657,14 +709,16 @@ when defined(broker): proc sendWill(ctx: MqttCtx) {.async.} = ## Send the will if ctx.willTopic != "": - for c in mqttbroker.subscribers[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, message: ctx.willMsg, typ: Publish) + 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, topic, message: string, qos: uint8, retain: bool, senderId: string) {.async.} = + proc publishToSubscribers(seqctx: seq[MqttCtx], pkt: Pkt, matchedTopic, topic, message: string, qos: QoS, retain: bool, senderId: string) {.async.} = ## Publish async to clients for c in seqctx: if c.state != Connected: @@ -672,7 +726,7 @@ when defined(broker): continue let msgId = c.nextMsgId() - qosSub = qosAlign(qos, c.subscribed[topic]) + qosSub = qosAlign(qos, c.subscribed[matchedTopic]) if mqttbroker.passClientId: c.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: topic, qos: qosSub, retain: retain, message: senderId & ":" & message, typ: Publish) @@ -690,7 +744,7 @@ when defined(broker): #when defined(broker): proc onConnect(ctx: MqttCtx, pkt: Pkt) {.async.} = when not defined(broker): - ctx.wrn "Packet type only supported for broker: " & $pkt.typ + ctx.wrn("Packet type only supported for broker: " & $pkt.typ) else: var offset: int @@ -719,10 +773,10 @@ proc onConnect(ctx: MqttCtx, pkt: Pkt) {.async.} = # Will qos=2 if ctx.connFlags[3] == '1': - ctx.willQos = 2.uint8 + ctx.willQoS = 2 # Will qos=1 elif ctx.connFlags[4] == '1': - ctx.willQos = 1.uint8 + ctx.willQoS = 1 # Username if ctx.connFlags[0] == '1': @@ -806,14 +860,14 @@ proc onConnAck(ctx: MqttCtx, pkt: Pkt): Future[void] = if code == 0: ctx.beenConnected = true if ctx.verbosity >= 1: - ctx.dbg "Connection established" + ctx.dbg("Connection established") else: - ctx.wrn "Connect failed, code: " & $code + ctx.wrn("Connect failed, code: " & $code) result = ctx.work() proc onPublish(ctx: MqttCtx, pkt: Pkt) {.async.} = let - qos = (pkt.flags shr 1) and 0x03 + qos = QoS((pkt.flags shr 1) and 0x03) retain = if (pkt.flags and 0x01) == 1: true else: false # When subscribing and first message is a # retained message, this will be `1` @@ -833,53 +887,29 @@ proc onPublish(ctx: MqttCtx, pkt: Pkt) {.async.} = (message, offset) = pkt.getstring(offset, false) when defined(broker): - # Send message to all subscribers on "#" - if mqttbroker.subscribers.hasKey("#"): - await publishToSubscribers(mqttbroker.subscribers["#"], pkt, "#", message, qos, retain, ctx.clientid) - # Send message to all subscribers on _the topic_ - if mqttbroker.subscribers.hasKey(topic): - await publishToSubscribers(mqttbroker.subscribers[topic], pkt, topic, message, qos, retain, ctx.clientid) + for matchedTopic, topicCtx in mqttbroker.subscribers: + if matchTopic(topicCtx.templ, topic): + await publishToSubscribers(topicCtx.ctxs, pkt, matchedTopic, topic, message, qos, retain, ctx.clientid) if mqttbroker.verbosity >= 1: verbose("Client >> " & ctx.clientId & " has published a message") if retain: - if qos == 0 and message == "": + if message == "": mqttbroker.retained.del(topic) else: # Add or overwrite existing retained messages on this topic. - mqttbroker.retained[topic] = RetainedMsg(msg: message, qos: qos, time: epochTime(), clientid: ctx.clientid) - # Check if client already has published a retained messaged on this topic. In that - # case do not add it, since the QOS, msg and time is preserved in the MqttBroker.retained. - if topic notin ctx.retained: - ctx.retained.add(topic) + retainMsg(topic, message, qos, ctx.clientid) if mqttbroker.verbosity >= 1: verbose("Retained ", mqttbroker.retained) when not defined(broker): var callbacks: seq[PubCallback] - for top, cb in ctx.pubCallbacks: - if top == topic or top == "#": - callbacks.add(cb) - if top.endsWith("/#"): - # the multi-level wildcard can represent zero levels. - if topic == top[0 .. ^3]: - callbacks.add(cb) - continue - var topicw = top - topicw.removeSuffix("#") - if topic.contains(topicw): - callbacks.add(cb) - if top.contains("+"): - var topelem = split(top, '/') - if len(topelem) == count(topic, '/') + 1: - var i = 0 - for e in split(topic, '/'): - if topelem[i] != "+" and e != topelem[i]: break - i = i+1 - if i == len(topelem): - callbacks.add(cb) + for t in ctx.pubCallbacks.values(): + if matchTopic(t.templ, topic): + callbacks.add(t.cb) + for cb in callbacks: cb.cb(topic, message) @@ -896,7 +926,7 @@ proc onPubAck(ctx: MqttCtx, pkt: Pkt) {.async.} = assert ctx.workQueue[msgId].wk == PubWork assert ctx.workQueue[msgId].state == WorkSent assert ctx.workQueue[msgId].qos == 1 - ctx.workQueue.del msgId + ctx.workQueue.del(msgId) await ctx.work() proc onPubRec(ctx: MqttCtx, pkt: Pkt) {.async.} = @@ -923,65 +953,68 @@ proc onPubComp(ctx: MqttCtx, pkt: Pkt) {.async.} = assert ctx.workQueue[msgId].wk == PubWork assert ctx.workQueue[msgId].state == WorkSent assert ctx.workQueue[msgId].qos == 2 - ctx.workQueue.del msgId + ctx.workQueue.del(msgId) await ctx.work() #when defined(broker): proc onSubscribe(ctx: MqttCtx, pkt: Pkt) {.async.} = when not defined(broker): - ctx.wrn "Packet type only supported for broker: " & $pkt.typ + ctx.wrn("Packet type only supported for broker: " & $pkt.typ) else: var offset: int msgId: MsgId topic: string - qos: uint8 + qos: QoS nextLen: uint16 (msgId, offset) = pkt.getu16(0) ctx.msgIdSeq = msgId + var + newTopics: seq[string] + reply: string while offset < pkt.data.len: (nextLen, offset) = pkt.getu16(offset) (topic, offset) = pkt.getstring(offset, parseInt($nextLen)) - (qos, offset) = pkt.getu8(offset) + (qos, offset) = pkt.getQoS(offset) ctx.subscribed[topic] = qos await addSubscriber(ctx, topic) + newTopics.add(topic) + reply = reply & char(qos) - ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 0, typ: SubAck) - - # Send retained messaged for # - if topic == "#": - for top, ret in mqttbroker.retained: - let - msgId = ctx.nextMsgId() - qosRet = qosAlign(qos, ret.qos) - ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: top, qos: qosRet, message: ret.msg, typ: Publish) - # Send retained messaged for specific topic - elif mqttbroker.retained.hasKey(topic): - let - msgId = ctx.nextMsgId() - qosRet = qosAlign(qos, mqttbroker.retained[topic].qos) - ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: topic, qos: qosRet, message: mqttbroker.retained[topic].msg, typ: Publish) - - if mqttbroker.verbosity >= 1: - verbose("Client >> " & ctx.clientId & " has subscribed to a topic") - verbose("Subscribers", mqttbroker.subscribers) + if mqttbroker.verbosity >= 1: + verbose("Client >> " & ctx.clientId & " has subscribed to a topic") + verbose("Subscribers", mqttbroker.subscribers) + ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 0, message: reply, typ: SubAck) await ctx.work() + for topic in newTopics: + if topic notin mqttbroker.subscribers: + continue + let templ = mqttbroker.subscribers[topic].templ + for retTop, ret in mqttbroker.retained: + if matchTopic(templ, retTop): + let + msgId = ctx.nextMsgId() + qosRet = qosAlign(qos, ret.qos) + ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: retTop, qos: qosRet, message: ret.msg, typ: Publish) + + await ctx.work() + proc onSubAck(ctx: MqttCtx, pkt: Pkt) {.async.} = let (msgId, _) = pkt.getu16(0) assert msgId in ctx.workQueue assert ctx.workQueue[msgId].wk == SubWork assert ctx.workQueue[msgId].state == WorkSent - ctx.workQueue.del msgId + ctx.workQueue.del(msgId) #when defined(broker): proc onUnsubscribe(ctx: MqttCtx, pkt: Pkt) {.async.} = when not defined(broker): - ctx.wrn "Packet type only supported for broker: " & $pkt.typ + ctx.wrn("Packet type only supported for broker: " & $pkt.typ) else: var offset: int @@ -1011,12 +1044,12 @@ proc onUnsubAck(ctx: MqttCtx, pkt: Pkt) {.async.} = assert msgId in ctx.workQueue assert ctx.workQueue[msgId].wk == SubWork assert ctx.workQueue[msgId].state == WorkSent - ctx.workQueue.del msgId + ctx.workQueue.del(msgId) #when defined(broker): proc onDisconnect(ctx: MqttCtx, pkt: Pkt) {.async.} = when not defined(broker): - ctx.wrn "Packet type only supported for broker: " & $pkt.typ + ctx.wrn("Packet type only supported for broker: " & $pkt.typ) else: #await removeSubscriber(ctx) #await sendWill(ctx) @@ -1027,7 +1060,7 @@ proc onDisconnect(ctx: MqttCtx, pkt: Pkt) {.async.} = #when defined(broker): proc onPingReq(ctx: MqttCtx, pkt: Pkt) {.async.} = when not defined(broker): - ctx.wrn "Packet type only supported for broker: " & $pkt.typ + ctx.wrn("Packet type only supported for broker: " & $pkt.typ) else: var msgId = ctx.nextMsgId() + 1000 while ctx.workQueue.hasKey(msgId): @@ -1058,7 +1091,7 @@ proc handle(ctx: MqttCtx, pkt: Pkt) {.async.} = of Unsubscribe: await ctx.onUnsubscribe(pkt) of Disconnect: await ctx.onDisconnect(pkt) of PingReq: await ctx.onPingReq(pkt) - else: ctx.wrn "Unknown pkt type " & $pkt.typ + else: ctx.wrn("Unknown pkt type " & $pkt.typ) # # Async work functions @@ -1073,7 +1106,7 @@ proc runRx(ctx: MqttCtx) {.async.} = await ctx.handle(pkt) except OsError: if ctx.verbosity >= 2: - ctx.wrn "Boom, socket is closed" + ctx.wrn("Boom, socket is closed") proc runPing(ctx: MqttCtx) {.async.} = while true: @@ -1089,7 +1122,7 @@ proc connectBroker(ctx: MqttCtx) {.async.} = ctx.keepAlive = 60 if ctx.verbosity >= 1: - ctx.dbg "Connecting to " & ctx.host & ":" & $ctx.port + ctx.dbg("Connecting to " & ctx.host & ":" & $ctx.port) ctx.state = Error # set to Connecting by sendConnect @@ -1099,7 +1132,7 @@ proc connectBroker(ctx: MqttCtx) {.async.} = 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" + ctx.wrn("Requested SSL session but ssl is not enabled") await ctx.close("SSL not enabled") let ok = await ctx.sendConnect() @@ -1119,7 +1152,7 @@ proc runConnect(ctx: MqttCtx) {.async.} = await ctx.connectBroker() except OSError as e: if ctx.verbosity >= 1 or not ctx.beenConnected: - ctx.dbg "Error connecting to " & ctx.host + ctx.dbg("Error connecting to " & ctx.host) if ctx.verbosity >= 2: echo e.msg ctx.state = Error @@ -1132,9 +1165,10 @@ proc runConnect(ctx: MqttCtx) {.async.} = # work() checks that `state=Connected`. Therefor our re-Subscribe # will be inserted first in the queue. if ctx.workQueue.len() == 0: - for topic, cb in ctx.pubCallbacks: + for topic, t in ctx.pubCallbacks: let msgId = ctx.nextMsgId() - ctx.workQueue[msgId] = Work(wk: SubWork, msgId: msgId, topic: topic, qos: cb.qos, typ: Subscribe) + ctx.workQueue[msgId] = Work(wk: SubWork, msgId: msgId, topic: topic, qos: t.cb.qos, typ: Subscribe) + await sleepAsync(1000) # @@ -1150,7 +1184,7 @@ proc setPingInterval*(ctx: MqttCtx, txInterval: int = 60) = if txInterval > 0 and txInterval < 65535: ctx.keepAlive = txInterval.uint16 -proc setHost*(ctx: MqttCtx, host: string, port: int=1883, sslOn=false) = +proc setHost*(ctx: MqttCtx, host: string, port: int = 1883, sslOn = false) = ## Set the MQTT host. ctx.host = host ctx.port = Port(port) @@ -1167,12 +1201,12 @@ proc setAuth*(ctx: MqttCtx, username: string, password: string) = ctx.username = username ctx.password = password -proc setWill*(ctx: MqttCtx, topic, msg: string, qos=0, retain=false) = +proc setWill*(ctx: MqttCtx, topic, msg: string, qos: QoS = 0, retain = false) = ## Set the clients will. - ctx.willFlag = true + ctx.willFlag = topic != "" ctx.willTopic = topic ctx.willMsg = msg - ctx.willQoS = qos.uint8 + ctx.willQoS = qos ctx.willRetain = retain proc setMaxInflightMessages*(ctx: MqttCtx, maxInflightMessages: int) = @@ -1200,7 +1234,7 @@ proc disconnect*(ctx: MqttCtx) {.async.} = await ctx.close("disconnect") ctx.state = Disabled -proc publish*(ctx: MqttCtx, topic: string, message: string, qos=0, retain=false) {.async.} = +proc publish*(ctx: MqttCtx, topic: string, message: string, qos: QoS = 0, retain = false) {.async.} = ## Publish a message. ## ## **Required:** @@ -1227,7 +1261,7 @@ proc publish*(ctx: MqttCtx, topic: string, message: string, qos=0, retain=false) ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: topic, qos: qos, message: message, retain: retain, typ: Publish) await ctx.work() -proc subscribe*(ctx: MqttCtx, topic: string, qos: int, callback: PubCallback.cb): Future[void] = +proc subscribe*(ctx: MqttCtx, topic: string, qos: QoS | int, callback: PubCallback.cb): Future[void] = ## Subscribe to a topic. ## ## Access the callback with: @@ -1235,8 +1269,8 @@ proc subscribe*(ctx: MqttCtx, topic: string, qos: int, callback: PubCallback.cb) ## proc callbackName(topic: string, message: string) = ## echo "Topic: ", topic, ": ", message let msgId = ctx.nextMsgId() - ctx.workQueue[msgId] = Work(wk: SubWork, msgId: msgId, topic: topic, qos: qos, typ: Subscribe) - ctx.pubCallbacks[topic] = PubCallback(cb: callback, qos: qos) + ctx.workQueue[msgId] = Work(wk: SubWork, msgId: msgId, topic: topic, qos: QoS(qos), typ: Subscribe) + ctx.pubCallbacks[topic] = (templ: splitTopic(topic), cb: PubCallback(cb: callback, qos: QoS(qos))) result = ctx.work() proc unsubscribe*(ctx: MqttCtx, topic: string): Future[void] = diff --git a/nmqtt/nmqtt.nim b/nmqtt/nmqtt.nim index 94ac176..f20b2cf 100644 --- a/nmqtt/nmqtt.nim +++ b/nmqtt/nmqtt.nim @@ -26,7 +26,6 @@ proc keepAliveMonitor(ctx: MqttCtx) {.async.} = verbose("Connections >> " & ctx.clientid & " was disconnected. Keep alive time overdue.") break - proc processClient(s: AsyncSocket) {.async.} = ## Create new client let ctx = MqttCtx() @@ -75,10 +74,10 @@ proc processClient(s: AsyncSocket) {.async.} = if mqttbroker.connections.hasKey(ctx.clientid): mqttbroker.connections.del(ctx.clientid) - # Cleanup retained messages from client. - for top in ctx.retained: - if mqttbroker.retained[top].clientid == ctx.clientid: - mqttbroker.retained.del(top) + # # Cleanup retained messages from client. + # for top in ctx.retained: + # if mqttbroker.retained[top].clientid == ctx.clientid: + # mqttbroker.retained.del(top) if not ctx.s.isClosed() and ctx.beenConnected: ctx.s.close() @@ -87,12 +86,15 @@ proc processClient(s: AsyncSocket) {.async.} = if mqttbroker.verbosity >= 3: verbose(ctx) - proc serve(host: string, port: int) {.async.} = var broker = newAsyncSocket() - broker.setSockOpt(OptReuseAddr, true) - broker.bindAddr(Port(port), host) - broker.listen() + try: + broker.setSockOpt(OptReuseAddr, true) + broker.bindAddr(Port(port), host) + broker.listen() + except CatchableError as e: + verbose("ERROR: " & e.msg) + return if mqttbroker.sslOn: if not fileExists(mqttbroker.sslCert) or not fileExists(mqttbroker.sslKey): @@ -110,7 +112,6 @@ proc serve(host: string, port: int) {.async.} = let client = await broker.accept() asyncCheck processClient(client) - proc showConf(mb: MqttBroker, configfile: string) = ## Show the config details @@ -147,7 +148,6 @@ CONFIG: """.format(configfile) - proc loadPasswords(passwordFile: string) = ## Loads the usernames and passwords if passwordFile == "": @@ -161,7 +161,6 @@ proc loadPasswords(passwordFile: string) = let pass = split(line, ":", maxsplit=1) mqttbroker.passwords[pass[0]] = pass[1] - proc loadConf(mb: MqttBroker, config: string) = ## Parses the config file @@ -194,7 +193,6 @@ proc loadConf(mb: MqttBroker, config: string) = let passwordFile = dict.getSectionValue("","password_file") loadPasswords(passwordFile) - proc handler() {.noconv.} = ## Catch ctrl+c from user echo " " @@ -202,7 +200,6 @@ proc handler() {.noconv.} = verbose(mqttbroker) quit() - proc nmqttBroker(config="", host="127.0.0.1", port=1883, verbosity=0, max_conn=0, clientid_maxlen=60, clientid_spaces=false, clientid_empty=false, client_kickold=false, clientid_pass=false, password_file="", @@ -240,8 +237,6 @@ proc nmqttBroker(config="", host="127.0.0.1", port=1883, verbosity=0, max_conn=0 runForever() - - when isMainModule: let topLvlUse = """nmqtt version """ & nmqttVersion & """ diff --git a/nmqtt/nmqtt_sub.nim b/nmqtt/nmqtt_sub.nim index 2464f77..2dde1a1 100644 --- a/nmqtt/nmqtt_sub.nim +++ b/nmqtt/nmqtt_sub.nim @@ -59,7 +59,7 @@ proc nmqttSub(host="127.0.0.1", port=1883, ssl=false, clientid="", username="", # Subscribe to topic await ctx.subscribe(t, qos, onData) if ctx.verbosity >= 1: - ctx.dbg "Subscribing to: " & t + ctx.dbg("Subscribing to: " & t) # Control CTRL+c hook setControlCHook(handler) diff --git a/tests/publish_retained.nim b/tests/publish_retained.nim index f1af265..110639d 100644 --- a/tests/publish_retained.nim +++ b/tests/publish_retained.nim @@ -5,22 +5,129 @@ suite "test suite for publish retained": let ctxMain = newCtx() ctxListen = newCtx() + ctxNotEmpty = newCtx() + ctxEmpty = newCtx() (tpc, msg) = tdata("publish retain msg") proc conn() {.async.} = - var msgFound: bool + var + msgFound: bool + msgFound2: bool + msgFound3: bool - waitFor ctxMain.publish(tpc, msg, qos=1, retain=true) + waitFor ctxMain.publish(tpc, msg, retain=true) waitFor sleepAsync(500) proc onDataRetain(topic: string, message: string) = - if topic == tpc: - check(message == msg) + if topic == tpc and message == msg: + check(msgFound == false) msgFound = true + proc onDataRetain2(topic: string, message: string) = + if topic == tpc and message == msg: + check(msgFound2 == false) + msgFound2 = true + + proc onDataRetain3(topic: string, message: string) = + if topic == tpc and message == msg: + check(msgFound3 == false) + msgFound3 = true + await ctxListen.subscribe(tpc, 2, onDataRetain) await sleepAsync(500) + waitFor ctxMain.publish(tpc, "") + await sleepAsync(500) + await ctxNotEmpty.subscribe(tpc, 2, onDataRetain2) + await sleepAsync(500) + + waitFor ctxMain.publish(tpc, "", retain=true) + await sleepAsync(500) + await ctxEmpty.subscribe(tpc, 2, onDataRetain3) + await sleepAsync(500) + check(msgFound == true) + check(msgFound2 == true) + check(msgFound3 == false) + + waitFor conn() + + test "subscribe retain msg": + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("publish retain msg") + + tpc1 = tpc & "/test/" + tpc2 = tpc & "/test" + tpc3 = tpc & "//data" + tpc4 = tpc & "/test/data" + tpc5 = tpc & "/test/random" + tpc6 = tpc & "/test/random/1/2/3" + tpc7 = tpc & "/test/data/" + tpc8 = tpc & "test/" + + proc conn() {.async.} = + var + msgFound1: int + msgFound2: int + msgFound3: int + msgFound4: int + + waitFor ctxMain.publish(tpc1, msg, retain=true) + waitFor ctxMain.publish(tpc2, msg, retain=true) + waitFor ctxMain.publish(tpc3, msg, retain=true) + waitFor ctxMain.publish(tpc4, msg, retain=true) + waitFor ctxMain.publish(tpc5, msg, retain=true) + waitFor ctxMain.publish(tpc6, msg, retain=true) + waitFor ctxMain.publish(tpc7, msg, retain=true) + waitFor ctxMain.publish(tpc8, msg, retain=true) + + waitFor sleepAsync(500) + + proc onDataRetain1(topic: string, message: string) = + if topic == tpc1 and message == msg: + inc msgFound1 + + proc onDataRetain2(topic: string, message: string) = + if message == msg: + inc msgFound2 + + proc onDataRetain3(topic: string, message: string) = + if topic == tpc3 and message == msg: + inc msgFound3 + + proc onDataRetain4(topic: string, message: string) = + if topic == tpc6 and message == msg: + inc msgFound4 + + await ctxListen.subscribe(tpc1, 0, onDataRetain1) + await sleepAsync(500) + await ctxListen.subscribe(tpc & "/+/data", 0, onDataRetain3) + await sleepAsync(500) + await ctxListen.subscribe(tpc & "/+/random/1/#", 0, onDataRetain4) + await sleepAsync(500) + await ctxListen.subscribe(tpc & "/#", 0, onDataRetain2) + await sleepAsync(500) + + # Expected count is 2 due to MQTT Retained message behavior: + # Each message is received twice because it is triggered first by its + # own specific subscription, and then a second time by the wildcard `/#` + # subscription, which forces the broker to resend all retained data. + check(msgFound1 == 2) + check(msgFound3 == 2) + check(msgFound4 == 2) + check(msgFound2 == 7) + + waitFor ctxMain.publish(tpc1, "", retain=true) + waitFor ctxMain.publish(tpc2, "", retain=true) + waitFor ctxMain.publish(tpc3, "", retain=true) + waitFor ctxMain.publish(tpc4, "", retain=true) + waitFor ctxMain.publish(tpc5, "", retain=true) + waitFor ctxMain.publish(tpc6, "", retain=true) + waitFor ctxMain.publish(tpc7, "", retain=true) + waitFor ctxMain.publish(tpc8, "", retain=true) + + await sleepAsync(500) waitFor conn() diff --git a/tests/subscribe.nim b/tests/subscribe.nim index 34fa2ca..58b95e2 100644 --- a/tests/subscribe.nim +++ b/tests/subscribe.nim @@ -8,23 +8,26 @@ suite "test suite for subscribe": (tpc, msg) = tdata("subscribe to topic qos=0") proc conn() {.async.} = + var receivedMsg: bool + proc onDataSubQoS0(topic: string, message: string) = - if topic == tpc: - check(message == msg) - return + if topic == tpc and message == msg: + receivedMsg = true + await ctxListen.subscribe(tpc, 0, onDataSubQoS0) await sleepAsync(500) await ctxMain.publish(tpc, msg, 0) await sleepAsync(500) - await ctxListen.unsubscribe(tpc) - await sleepAsync(500) + check(receivedMsg == true) check(hasAllInDmp(@["tx> Subscribe(02):", "rx> SubAck(00):", "tx> Publish(00):", "rx> Publish(00):"])) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "subscribe to topic qos=1": let @@ -33,10 +36,11 @@ suite "test suite for subscribe": (tpc, msg) = tdata("subscribe to topic qos=1") proc conn() {.async.} = + var receivedMsg: bool + proc onDataSubQoS1(topic: string, message: string) = - if topic == tpc: - check(message == msg) - return + if topic == tpc and message == msg: + receivedMsg = true await ctxListen.subscribe(tpc, 1, onDataSubQoS1) await sleepAsync(500) @@ -45,6 +49,7 @@ suite "test suite for subscribe": await ctxListen.unsubscribe(tpc) await sleepAsync(500) + check(receivedMsg == true) check(hasAllInDmp(@["tx> Subscribe(02):", "rx> SubAck(00):", "tx> Publish(02):", @@ -53,6 +58,8 @@ suite "test suite for subscribe": "tx> PubAck(02):"])) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "subscribe to topic qos=2": let @@ -61,10 +68,11 @@ suite "test suite for subscribe": (tpc, msg) = tdata("subscribe to topic qos=2") proc conn() {.async.} = + var receivedMsg: bool + proc onDataSubQoS2(topic: string, message: string) = - if topic == tpc: - check(message == msg) - return + if topic == tpc and message == msg: + receivedMsg = true await ctxListen.subscribe(tpc, 2, onDataSubQoS2) await sleepAsync(500) @@ -73,6 +81,7 @@ suite "test suite for subscribe": await ctxListen.unsubscribe(tpc) await sleepAsync(500) + check(receivedMsg == true) check(hasAllInDmp(@["tx> Subscribe(02):", "rx> SubAck(00):", "tx> Publish(04):", @@ -85,39 +94,45 @@ suite "test suite for subscribe": "tx> PubComp(02):"])) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "subscribe to multiple topics": let ctxMain = newCtx() ctxListen = newCtx() (tpc, msg) = tdata("subscribe to multiple topics") + tpc1 = tpc & "-1" + tpc2 = tpc & "-2" + msg1 = msg & "-mul1" + msg2 = msg & "-mul2" proc conn() {.async.} = - var topic1, topic2: bool + var + receivedMsg1: bool + receivedMsg2: bool proc onDataSubMul1(topic: string, message: string) = - check(message == msg & "-mul1") - check(not topic1) - topic1 = true + if topic == tpc1 and message == msg1: + receivedMsg1 = true proc onDataSubMul2(topic: string, message: string) = - check(message == msg & "-mul2") - check(not topic2) - topic2 = true + if topic == tpc2 and message == msg2: + receivedMsg2 = true - await ctxListen.subscribe(tpc & "-1", 0, onDataSubMul1) - await ctxListen.subscribe(tpc & "-2", 0, onDataSubMul2) + await ctxListen.subscribe(tpc1, 0, onDataSubMul1) + await ctxListen.subscribe(tpc2, 0, onDataSubMul2) await sleepAsync(500) - await ctxMain.publish(tpc & "-1", msg & "-mul1", 0) - await ctxMain.publish(tpc & "-2", msg & "-mul2", 0) + await ctxMain.publish(tpc1, msg1, 0) + await ctxMain.publish(tpc2, msg2, 0) await sleepAsync(500) - await ctxListen.unsubscribe(tpc & "-1") - await ctxListen.unsubscribe(tpc & "-2") - check(topic1) - check(topic2) + check(receivedMsg1 == true) + check(receivedMsg2 == true) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "subscribe to multiple with identical topic": let @@ -129,15 +144,15 @@ suite "test suite for subscribe": var sub1, sub2, sub3: int proc onDataSubMul1(topic: string, message: string) = - if topic == tpc: + if topic == tpc and message == msg: sub1 += 1 proc onDataSubMul2(topic: string, message: string) = - if topic == tpc: + if topic == tpc and message == msg: sub2 += 1 proc onDataSubMul3(topic: string, message: string) = - if topic == tpc: + if topic == tpc and message == msg: sub3 += 1 check(ctxListen.pubCallbacks.len() == 0) @@ -171,105 +186,296 @@ suite "test suite for subscribe": check(sub3 == 3) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "subscribe to #": let ctxMain = newCtx() ctxListen = newCtx() (tpc, msg) = tdata("subscribe to #") + tpc1 = tpc & "/random1" + tpc2 = tpc & "/random2/1" + tpc3 = tpc & "/random3/2/1/0" proc conn() {.async.} = - var msgCount: int + var + receivedTpc1: bool + receivedTpc2: bool + receivedTpc3: bool proc onDataSubAll(topic: string, message: string) = - msgCount += 1 + if topic == tpc1 and message == msg: + receivedTpc1 = true + elif topic == tpc2 and message == msg: + receivedTpc2 = true + elif topic == tpc3 and message == msg: + receivedTpc3 = true - await ctxListen.subscribe(tpc & "/#", 0, onDataSubAll) + await ctxListen.subscribe("#", 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) + check(receivedTpc1 == false) + check(receivedTpc2 == false) + check(receivedTpc3 == false) + + await ctxMain.publish(tpc1, msg, 0) + await ctxMain.publish(tpc2, msg, 0) + await ctxMain.publish(tpc3, msg, 0) await sleepAsync(500) - await ctxListen.unsubscribe(tpc & "/#") - check(msgCount == 3) + check(receivedTpc1 == true) + check(receivedTpc2 == true) + check(receivedTpc3 == true) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "subscribe to test/#": let ctxMain = newCtx() ctxListen = newCtx() + ctxOther = newCtx() (tpc, msg) = tdata("subscribe to test/#") + tpc1 = tpc & "/test/random1" + tpc2 = tpc & "/test/" + tpc3 = tpc & "/test" + tpc4 = tpc & "/test/random3/2" proc conn() {.async.} = - var msgCount: int + var + receivedTpc1: bool + receivedTpc2: bool + receivedTpc3: bool + receivedTpc4: bool proc onDataSubWild(topic: string, message: string) = - msgCount += 1 + check(message == msg) + if topic == tpc1: + receivedTpc1 = true + elif topic == tpc2: + receivedTpc2 = true + elif topic == tpc3: + receivedTpc3 = true + elif topic == tpc4: + receivedTpc4 = true + + proc empty(topic: string, message: string) = + discard await ctxListen.subscribe(tpc & "/test/#", 0, onDataSubWild) + await ctxListen.subscribe(tpc & "/second/#", 0, empty) + await ctxOther.subscribe(tpc & "/second/#", 0, empty) await sleepAsync(500) - await ctxMain.publish(tpc & "/test/random1", msg, 0) + check(receivedTpc1 == false) + check(receivedTpc2 == false) + check(receivedTpc3 == false) + check(receivedTpc4 == false) + + await ctxMain.publish(tpc1, 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 ctxMain.publish(tpc2, msg, 0) + await ctxMain.publish(tpc3, msg, 0) + await ctxMain.publish(tpc4, msg, 0) await sleepAsync(500) - await ctxListen.unsubscribe(tpc & "/test/#") - check(msgCount == 3) + check(receivedTpc1 == true) + check(receivedTpc2 == true) + check(receivedTpc3 == true) + check(receivedTpc4 == true) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() + waitFor ctxOther.disconnect() test "subscribe to test/+": let ctxMain = newCtx() ctxListen = newCtx() (tpc, msg) = tdata("subscribe to test/+") + tpc1 = tpc & "/test/random1" + tpc2 = tpc & "/test/" + tpc3 = tpc & "/test/3" proc conn() {.async.} = - var msgCount: int + var + receivedTpc1: bool + receivedTpc2: bool + receivedTpc3: bool proc onDataSubWild(topic: string, message: string) = - msgCount += 1 + check(message == msg) + if topic == tpc1: + receivedTpc1 = true + elif topic == tpc2: + receivedTpc2 = true + elif topic == tpc3: + receivedTpc3 = true await ctxListen.subscribe(tpc & "/test/+", 0, onDataSubWild) await sleepAsync(500) - await ctxMain.publish(tpc & "/test/random1", msg, 0) + check(receivedTpc1 == false) + check(receivedTpc2 == false) + check(receivedTpc3 == false) + + await ctxMain.publish(tpc1, 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(tpc2, msg, 0) + await ctxMain.publish(tpc3, msg, 0) await ctxMain.publish(tpc & "/test/random3/2", msg, 0) await sleepAsync(500) - await ctxListen.unsubscribe(tpc & "/test/+") - check(msgCount == 2) + + check(receivedTpc1 == true) + check(receivedTpc2 == true) + check(receivedTpc3 == true) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() - test "subscribe to test/+/test": + test "subscribe to test/+/data": let ctxMain = newCtx() ctxListen = newCtx() - (tpc, msg) = tdata("subscribe to test/+/test") + (tpc, msg) = tdata("subscribe to test/+/data") + tpc1 = tpc & "test/1/data" + tpc2 = tpc & "test/random4/data" + tpc3 = tpc & "test//data" proc conn() {.async.} = - var msgCount: int + var + receivedTpc1: bool + receivedTpc2: bool + receivedTpc3: bool proc onDataSubWild(topic: string, message: string) = - msgCount += 1 + check(message == msg) + if topic == tpc1: + receivedTpc1 = true + elif topic == tpc2: + receivedTpc2 = true + elif topic == tpc3: + receivedTpc3 = true await ctxListen.subscribe(tpc & "test/+/data", 0, onDataSubWild) await sleepAsync(500) + check(receivedTpc1 == false) + check(receivedTpc2 == false) + check(receivedTpc3 == false) + + await ctxMain.publish(tpc1, msg, 0) + await ctxMain.publish(tpc & "second/random2/data", msg, 0) + await ctxMain.publish(tpc & "test/random3", msg, 0) + await ctxMain.publish(tpc2, msg, 0) + await ctxMain.publish(tpc3, msg, 0) + await ctxMain.publish(tpc & "test/random5/data/random6", msg, 0) + await sleepAsync(500) + + check(receivedTpc1 == true) + check(receivedTpc2 == true) + check(receivedTpc3 == true) + + waitFor conn() + + test "subscribe to test/+/+/data": + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to test/+/+/data") + tpc1 = tpc & "test/random5/random6/data" + tpc2 = tpc & "test///data" + tpc3 = tpc & "test/0/1/data" + + proc conn() {.async.} = + var + receivedTpc1: bool + receivedTpc2: bool + receivedTpc3: bool + + proc onDataSubWild(topic: string, message: string) = + check(message == msg) + if topic == tpc1: + receivedTpc1 = true + elif topic == tpc2: + receivedTpc2 = true + elif topic == tpc3: + receivedTpc3 = true + + await ctxListen.subscribe(tpc & "test/+/+/data", 0, onDataSubWild) + await sleepAsync(500) + check(receivedTpc1 == false) + check(receivedTpc2 == false) + check(receivedTpc3 == false) + 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 ctxMain.publish(tpc & "test/random5/random6/data", msg, 0) + await ctxMain.publish(tpc & "test///data", msg, 0) + await ctxMain.publish(tpc & "test/random5/random6/random7/data", msg, 0) + await ctxMain.publish(tpc & "test/random5/random6/data/random8", msg, 0) + await ctxMain.publish(tpc & "test/0/1/data", msg, 0) await sleepAsync(500) - await ctxListen.unsubscribe(tpc & "test/+/data") - check(msgCount == 2) + + check(receivedTpc1 == true) + check(receivedTpc2 == true) + check(receivedTpc3 == true) + + waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() + + test "subscribe to test/+/data/#": + let + ctxMain = newCtx() + ctxListen = newCtx() + (tpc, msg) = tdata("subscribe to test/+/data/#") + tpc1 = tpc & "test//data" + tpc2 = tpc & "test/random5/data/" + tpc3 = tpc & "test/random5/data/random6/random7/random8" + + proc conn() {.async.} = + var + receivedTpc1: bool + receivedTpc2: bool + receivedTpc3: bool + + proc onDataSubWild(topic: string, message: string) = + check(message == msg) + if topic == tpc1: + receivedTpc1 = true + elif topic == tpc2: + receivedTpc2 = true + elif topic == tpc3: + receivedTpc3 = true + + await ctxListen.subscribe(tpc & "test/+/data/#", 0, onDataSubWild) + await sleepAsync(500) + check(receivedTpc1 == false) + check(receivedTpc2 == false) + check(receivedTpc3 == false) + + await ctxMain.publish(tpc & "test/random0/random1/data", msg, 0) + await ctxMain.publish(tpc & "second/random2/data", msg, 0) + await ctxMain.publish(tpc & "second/random2/data/test", msg, 0) + await ctxMain.publish(tpc & "test/random3", msg, 0) + await ctxMain.publish(tpc1, msg, 0) + await ctxMain.publish(tpc2, msg, 0) + await ctxMain.publish(tpc & "test/random5//data/random6/random7", msg, 0) + await ctxMain.publish(tpc3, msg, 0) + await ctxMain.publish(tpc & "random1/test/random5/data/random6/random7", msg, 0) + await sleepAsync(500) + + check(receivedTpc1 == true) + check(receivedTpc2 == true) + check(receivedTpc3 == true) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "stay subscribed after disconnect with reconnect": let @@ -281,7 +487,8 @@ suite "test suite for subscribe": var msgCount: int proc onDataSubKeep(topic: string, message: string) = - msgCount += 1 + if topic == tpc and message == msg: + msgCount += 1 await ctxListen.subscribe(tpc, 0, onDataSubKeep) await sleepAsync(500) @@ -322,6 +529,8 @@ suite "test suite for subscribe": await ctxListen.disconnect() waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "stay subscribed after disconnect with reconnect with same qos=2": let @@ -333,7 +542,8 @@ suite "test suite for subscribe": var msgCount: int proc onDataSubKeep(topic: string, message: string) = - msgCount += 1 + if topic == tpc and message == msg: + msgCount += 1 await ctxListen.subscribe(tpc, 2, onDataSubKeep) await sleepAsync(500) @@ -375,6 +585,8 @@ suite "test suite for subscribe": "rx> Unsuback(00):"])) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "stay subscribed after multipe (2) disconnect with reconnect": let @@ -386,7 +598,8 @@ suite "test suite for subscribe": var msgCount: int proc onDataSubKeepMultiple(topic: string, message: string) = - msgCount += 1 + if topic == tpc and message == msg: + msgCount += 1 await ctxListen.subscribe(tpc, 0, onDataSubKeepMultiple) await sleepAsync(500) @@ -437,6 +650,8 @@ suite "test suite for subscribe": "rx> Unsuback(00):"])) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxListen.disconnect() test "stay subscribed after long disconnect with reconnect": ## This test currently needs manual actions - you need to close/disconnect @@ -454,7 +669,8 @@ suite "test suite for subscribe": var msgCount: int proc onDataSubKeepLong(topic: string, message: string) = - msgCount += 1 + if topic == tpc and message == msg: + msgCount += 1 await ctxSlave.subscribe(tpc, 0, onDataSubKeepLong) await sleepAsync(500) @@ -509,3 +725,5 @@ suite "test suite for subscribe": "rx> Unsuback(00):"])) waitFor conn() + waitFor ctxMain.disconnect() + waitFor ctxSlave.disconnect() diff --git a/tests/tester.nim b/tests/tester.nim index 879fd76..8a4454c 100644 --- a/tests/tester.nim +++ b/tests/tester.nim @@ -10,11 +10,11 @@ include "../nmqtt.nim" randomize() -proc newCtx(): MqttCtx = +proc newCtx(autoStart: bool = true): MqttCtx = result = newMqttCtx("nmqttTest-" & $genOid()) result.setHost("127.0.0.1", 1883) - result.setPingInterval(1200) - waitFor result.start() + if autoStart: + waitFor result.start() proc tout(t, m, s: string) = ## Print test data during test. diff --git a/tests/willmsg.nim b/tests/willmsg.nim index 254843b..0679a70 100644 --- a/tests/willmsg.nim +++ b/tests/willmsg.nim @@ -3,67 +3,95 @@ suite "test suite for will messages": test "send will msg with default values": let - ctxMain = newCtx() + ctxMain = newCtx(autoStart = false) ctxListen = newCtx() (tpc, _) = tdata("send will msg with default values") proc conn() {.async.} = const willMsg = "willmsg_qos0_retain=false" - var willMsgCheck: bool + var willMsgCheck: int proc onDataWill(topic: string, message: string) = if topic == tpc and message == willMsg: - willMsgCheck = true + check(willMsgCheck == 0) + inc willMsgCheck await ctxListen.subscribe(tpc, 2, onDataWill) + await sleepAsync(500) + check(willMsgCheck == 0) ctxMain.setWill(tpc, willMsg) - await ctxMain.connect() - await sleepAsync(500) # Wait for full connection + + await ctxMain.start() + await sleepAsync(500) + await ctxMain.disconnect() + await sleepAsync(500) + + check(willMsgCheck == 0) + + await ctxMain.start() + await sleepAsync(500) ctxMain.s.close() - await sleepAsync(500) # Wait for willMsg to be sent + await sleepAsync(500) - check(willMsgCheck == true) + check(willMsgCheck == 1) waitFor conn() - test "send will msg retained = true": + test "send will msg retained = true,false": let - ctxMain = newCtx() + ctxRetain = newCtx(autoStart = false) + ctxNotRetain = newCtx(autoStart = false) ctxListen = newCtx() ctxDestroy = newCtx() - (tpc, _) = tdata("send will msg retained = true") + (tpc1, _) = tdata("send will msg retained = true") + (tpc2, _) = tdata("send will msg retained = false") proc conn() {.async.} = - - const willMsg = "willmsg_qos0_retain=true" + const willMsgRetain = "willmsg_qos0_retain=true" + const willMsgNotRetain = "willmsg_qos0_retain=false" var - willMsgCheck: bool - willMsgRetain: bool + willMsgRetainReceived: bool + willMsgNotRetainReceived: bool + willMsgRetained: bool + willMsgNotRetained: bool proc onDataWill(topic: string, message: string) = - if topic == tpc and message == willMsg: - willMsgCheck = true + if topic == tpc1 and message == willMsgRetain: + willMsgRetainReceived = true + elif topic == tpc2 and message == willMsgNotRetain: + willMsgNotRetainReceived = true - await ctxListen.subscribe(tpc, 2, onDataWill) + await ctxListen.subscribe(tpc1, 2, onDataWill) + await ctxListen.subscribe(tpc2, 2, onDataWill) # Set will and send - 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 + ctxRetain.setWill(tpc1, willMsgRetain, retain=true) + ctxNotRetain.setWill(tpc2, willMsgNotRetain, retain=false) + await ctxRetain.start() + await ctxNotRetain.start() + await sleepAsync(500) + check(willMsgRetainReceived == false) + check(willMsgNotRetainReceived == false) + ctxRetain.s.close() + ctxNotRetain.s.close() + await sleepAsync(500) - check(willMsgCheck == true) + check(willMsgRetainReceived == true) + check(willMsgNotRetainReceived == true) proc onDataWillRetain(topic: string, message: string) = - if topic == tpc and message == willMsg: - willMsgRetain = true + if topic == tpc1 and message == willMsgRetain: + willMsgRetained = true + elif topic == tpc2 and message == willMsgNotRetain: + willMsgNotRetained = true - await ctxDestroy.subscribe(tpc, 2, onDataWillRetain) + await ctxDestroy.subscribe(tpc1, 2, onDataWillRetain) + await ctxDestroy.subscribe(tpc2, 2, onDataWillRetain) await sleepAsync(500) - check(willMsgRetain == true) + check(willMsgRetained == true) + check(willMsgNotRetained == false) waitFor conn()