diff --git a/.gitignore b/.gitignore index e0ac4d5..ab98e39 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,5 @@ tests/tester -bin \ No newline at end of file +bin +nimble.develop +nimble.paths +nimbledeps diff --git a/README.md b/README.md index 5c2226f..63f75a3 100644 --- a/README.md +++ b/README.md @@ -176,17 +176,17 @@ a MQTT-broker and for subscribing to a topic on a MQTT-broker. The library suppo import nmqtt, asyncdispatch let ctx = newMqttCtx("nmqttClient") -ctx.set_host("test.mosquitto.org", 1883) -#ctx.set_auth("username", "password") -#ctx.set_ping_interval(30) -#ctx.set_ssl_certificates("cert.crt", "private.key") +ctx.setHost("test.mosquitto.org", 1883) +#ctx.setAuth("username", "password") +#ctx.setPingInterval(30) +#ctx.setSSLCertificates("cert.crt", "private.key") proc mqttSub() {.async.} = await ctx.start() - proc on_data(topic: string, message: string) = + proc onData(topic: string, message: string) = echo "got ", topic, ": ", message - await ctx.subscribe("nmqtt", 2, on_data) + await ctx.subscribe("nmqtt", 2, onData) asyncCheck mqttSub() runForever() @@ -209,11 +209,11 @@ proc mqttSubPub() {.async.} = await ctx.start() # Callback when receiving on the topic - proc on_data(topic: string, message: string) = + proc onData(topic: string, message: string) = echo "got ", topic, ": ", message # Subscribe to topic the topic `nmqtt` - await ctx.subscribe("nmqtt", 2, on_data) + await ctx.subscribe("nmqtt", 2, onData) await sleepAsync 500 # Publish a message to the topic `nmqtt` @@ -236,57 +236,57 @@ waitFor mqttSubPub() proc newMqttCtx*(clientId: string): MqttCtx = ``` -Initiate a new MQTT client +Initiate a new MQTT client. ____ -### set_ping_interval* +### setPingInterval* ```nim -proc set_ping_interval*(ctx: MqttCtx, txInterval: int) = +proc setPingInterval*(ctx: MqttCtx, txInterval: int) = ``` Set the clients ping interval in seconds. Default is 60 seconds. ____ -### set_ssl_certificates* +### setSSLCertificates* ```nim -proc set_ssl_certificates*(ctx: MqttCtx, sslCert: string, sslKey: string) = +proc setSSLCertificates*(ctx: MqttCtx, sslCert: string, sslKey: string) = ``` -Sets the SSL Certificate and Key files to use Mutual TLS authentication +Sets the SSL Certificate and Key files to use Mutual TLS authentication. ____ -### set_host* +### setHost* ```nim -proc set_host*(ctx: MqttCtx, host: string, port: int=1883, sslOn=false) = +proc setHost*(ctx: MqttCtx, host: string, port: int=1883, sslOn=false) = ``` -Set the MQTT host +Set the MQTT host. ____ -### set_auth* +### setAuth* ```nim -proc set_auth*(ctx: MqttCtx, username: string, password: string) = +proc setAuth*(ctx: MqttCtx, username: string, password: string) = ``` -Set the authentication for the host +Set the authentication for the host. ____ -### set_will* +### setWill* ```nim -proc set_will*(ctx: MqttCtx, topic, msg: string, qos=0, retain=false) = +proc setWill*(ctx: MqttCtx, topic, msg: string, qos=0, retain=false) = ``` Set the clients will. @@ -294,6 +294,18 @@ Set the clients will. ____ +### setMaxInflightMessages* + +```nim +proc setMaxInflightMessages*(ctx: MqttCtx, maxInflightMessages: int) = +``` + +Sets the maximum number of unacknowledged MQTT messages (QoS 1 and QoS 2). Default = 20. + + +____ + + ### connect* ```nim @@ -367,7 +379,7 @@ ____ proc subscribe*(ctx: MqttCtx, topic: string, qos: int, callback: PubCallback): Future[void] = ``` -Subscribe to a topic +Subscribe to a topic. Access the callback with: ```nim diff --git a/config.nims b/config.nims new file mode 100644 index 0000000..8ee48d2 --- /dev/null +++ b/config.nims @@ -0,0 +1,4 @@ +# begin Nimble config (version 2) +when withDir(thisDir(), system.fileExists("nimble.paths")): + include "nimble.paths" +# end Nimble config diff --git a/nmqtt.nim b/nmqtt.nim index 547ed30..b67fcb6 100644 --- a/nmqtt.nim +++ b/nmqtt.nim @@ -1,6 +1,6 @@ ## Native Nim MQTT client library and binaries ## -## zevv (https://github.com/zevv) & ThomasTJdev (https://github.com/ThomasTJdev) +## zevv (https://github.com/zevv) & ThomasTJdev (https://github.com/ThomasTJdev) & python36 (https://github.com/python36) import strutils, @@ -41,6 +41,7 @@ type inWork: bool hasNewWorks: bool keepAlive: uint16 + maxInflightMessages: int willFlag: bool willQoS: uint8 willRetain: bool @@ -337,10 +338,17 @@ proc nextMsgId(ctx: MqttCtx): MsgId = inc ctx.msgIdSeq return ctx.msgIdSeq +proc hasInflightSlots(ctx: MqttCtx): bool = + var cnt = 0 + for w in ctx.workQueue.values(): + if w.qos in {1, 2} and (w.typ != Publish or w.state == WorkSent): + inc cnt + if cnt == ctx.maxInflightMessages: + return false + true proc sendDisconnect(ctx: MqttCtx): Future[bool] {.async.} - proc close(ctx: MqttCtx, reason: string) {.async.} = if ctx.state in {Connecting, Connected}: ctx.state = Disconnecting @@ -350,7 +358,6 @@ proc close(ctx: MqttCtx, reason: string) {.async.} = ctx.s.close() ctx.state = Disconnected - proc send(ctx: MqttCtx, pkt: Pkt): Future[bool] {.async.} = ## Send the packet if ctx.state notin {Connecting, Connected, Disconnecting}: @@ -378,7 +385,6 @@ proc send(ctx: MqttCtx, pkt: Pkt): Future[bool] {.async.} = return true - proc recv(ctx: MqttCtx): Future[Pkt] {.async.} = ## Receive and parse the packet if ctx.state notin {Connecting,Connected}: @@ -442,7 +448,6 @@ proc recv(ctx: MqttCtx): Future[Pkt] {.async.} = ctx.dmp "rx> " & $pkt return pkt - proc sendConnect(ctx: MqttCtx): Future[bool] = var flags: uint8 flags = flags or CleanSession.uint8 @@ -479,7 +484,6 @@ proc sendConnect(ctx: MqttCtx): Future[bool] = ctx.state = Connecting result = ctx.send(pkt) - proc sendDisconnect(ctx: MqttCtx): Future[bool] = let pkt = newPkt(Disconnect, 0) result = ctx.send(pkt) @@ -616,17 +620,26 @@ proc work(ctx: MqttCtx) {.async.} = continue if work.wk == PubWork and work.state == WorkNew: - if work.typ == Publish and work.qos == 0: - if await ctx.sendWork(work): ctx.workQueue.del msgId + if work.typ == Publish: + if work.qos == 0: + if await ctx.sendWork(work): + ctx.workQueue.del msgId + + elif hasInflightSlots(ctx): + if await ctx.sendWork(work): + work.state = WorkSent elif work.typ == PubAck and work.qos == 1: - if await ctx.sendWork(work): ctx.workQueue.del msgId + if await ctx.sendWork(work): + ctx.workQueue.del msgId elif work.typ == PubComp and work.qos == 2: - if await ctx.sendWork(work): ctx.workQueue.del msgId + if await ctx.sendWork(work): + ctx.workQueue.del msgId else: - if await ctx.sendWork(work): work.state = WorkSent + if await ctx.sendWork(work): + work.state = WorkSent #when not defined(broker): elif work.wk == SubWork and work.state == WorkNew: @@ -884,6 +897,7 @@ proc onPubAck(ctx: MqttCtx, pkt: Pkt) {.async.} = assert ctx.workQueue[msgId].state == WorkSent assert ctx.workQueue[msgId].qos == 1 ctx.workQueue.del msgId + await ctx.work() proc onPubRec(ctx: MqttCtx, pkt: Pkt) {.async.} = let (msgId, _) = pkt.getu16(0) @@ -910,6 +924,7 @@ proc onPubComp(ctx: MqttCtx, pkt: Pkt) {.async.} = assert ctx.workQueue[msgId].state == WorkSent assert ctx.workQueue[msgId].qos == 2 ctx.workQueue.del msgId + await ctx.work() #when defined(broker): proc onSubscribe(ctx: MqttCtx, pkt: Pkt) {.async.} = @@ -1069,7 +1084,7 @@ proc runPing(ctx: MqttCtx) {.async.} = await ctx.work() proc connectBroker(ctx: MqttCtx) {.async.} = - ## Connect to the broker + ## Connect to the broker. if ctx.keepAlive == 0: ctx.keepAlive = 60 @@ -1094,7 +1109,7 @@ proc connectBroker(ctx: MqttCtx) {.async.} = proc runConnect(ctx: MqttCtx) {.async.} = - ## Auto-connect and reconnect to broker + ## Auto-connect and reconnect to broker. while true: if ctx.state == Disabled: @@ -1127,32 +1142,32 @@ proc runConnect(ctx: MqttCtx) {.async.} = # proc newMqttCtx*(clientId: string): MqttCtx = - ## Initiate a new MQTT client - MqttCtx(clientId: clientId, state: Disconnected) + ## Initiate a new MQTT client. + MqttCtx(clientId: clientId, state: Disconnected, maxInflightMessages: 20) -proc set_ping_interval*(ctx: MqttCtx, txInterval: int = 60) = +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 -proc set_host*(ctx: MqttCtx, host: string, port: int=1883, sslOn=false) = - ## Set the MQTT host +proc setHost*(ctx: MqttCtx, host: string, port: int=1883, sslOn=false) = + ## Set the MQTT host. ctx.host = host ctx.port = Port(port) ctx.sslOn = sslOn -proc set_ssl_certificates*(ctx: MqttCtx, sslCert: string, sslKey: string) = +proc setSSLCertificates*(ctx: MqttCtx, sslCert: string, sslKey: string) = # Sets the SSL Certificate and Key to use when connecting to the remote broker # for mutal TLS authentication ctx.sslCert = sslCert ctx.sslKey = sslKey -proc set_auth*(ctx: MqttCtx, username: string, password: string) = +proc setAuth*(ctx: MqttCtx, username: string, password: string) = ## Set the authentication for the host. ctx.username = username ctx.password = password -proc set_will*(ctx: MqttCtx, topic, msg: string, qos=0, retain=false) = +proc setWill*(ctx: MqttCtx, topic, msg: string, qos=0, retain=false) = ## Set the clients will. ctx.willFlag = true ctx.willTopic = topic @@ -1160,7 +1175,11 @@ proc set_will*(ctx: MqttCtx, topic, msg: string, qos=0, retain=false) = ctx.willQoS = qos.uint8 ctx.willRetain = retain -proc set_verbosity*(ctx: MqttCtx, verbosity: int) = +proc setMaxInflightMessages*(ctx: MqttCtx, maxInflightMessages: int) = + ## Sets the maximum number of unacknowledged MQTT messages (QoS 1 and QoS 2). Default = 20. + ctx.maxInflightMessages = maxInflightMessages + +proc setVerbosity*(ctx: MqttCtx, verbosity: int) = ## Set the verbosity. ctx.verbosity = verbosity diff --git a/nmqtt.nimble b/nmqtt.nimble index 2dd3d8f..b09a8d8 100644 --- a/nmqtt.nimble +++ b/nmqtt.nimble @@ -1,5 +1,5 @@ # Package -version = "1.0.8" +version = "1.0.9" author = "zevv & ThomasTJdev & python36" description = "Native MQTT library and binaries for publishing, subscribing and broker" license = "MIT" @@ -20,7 +20,7 @@ from strutils import format task test, "Runs the test suite.": - exec "nimble c -y -r tests/tester" + exec "nim c -r tests/tester" task setup, "Generate default nmqtt configuration file": diff --git a/nmqtt/nmqtt_pub.nim b/nmqtt/nmqtt_pub.nim index 16f0a65..61d6c20 100644 --- a/nmqtt/nmqtt_pub.nim +++ b/nmqtt/nmqtt_pub.nim @@ -19,20 +19,20 @@ proc nmqttPub(host="127.0.0.1", port=1883, ssl=false, clientid="", username="", echo "Running nmqtt_pub v" & nmqttVersion let ctx = newMqttCtx(if clientid != "": clientid else: "nmqttpub-" & $getCurrentProcessId()) - ctx.set_host(host, port, ssl) + ctx.setHost(host, port, ssl) if username != "" or password != "": - ctx.set_auth(username, password) + ctx.setAuth(username, password) # Set the will message if willretain and (willtopic == "" or willmsg == ""): echo "Error: Will-retain giving, but no topic given" quit() elif willtopic != "" and willmsg != "": - ctx.set_will(willtopic, willmsg, willqos, willretain) + ctx.setWill(willtopic, willmsg, willqos, willretain) # Set the verbosity - ctx.set_verbosity(verbosity) + ctx.setVerbosity(verbosity) # Control CTRL+c hook setControlCHook(handler) diff --git a/nmqtt/nmqtt_sub.nim b/nmqtt/nmqtt_sub.nim index a529df6..2464f77 100644 --- a/nmqtt/nmqtt_sub.nim +++ b/nmqtt/nmqtt_sub.nim @@ -25,23 +25,23 @@ proc nmqttSub(host="127.0.0.1", port=1883, ssl=false, clientid="", username="", if clientid != "": ctx.clientid = clientid - ctx.set_host(host, port, ssl) + ctx.setHost(host, port, ssl) if username != "" or password != "": - ctx.set_auth(username, password) + ctx.setAuth(username, password) # Set the ping interval/keep alive - ctx.set_ping_interval(keepalive) + ctx.setPingInterval(keepalive) # Set the will message if willretain and (willtopic == "" or willmsg == ""): echo "Error: Will-retain giving, but no topic given" quit(0) elif willtopic != "" and willmsg != "": - ctx.set_will(willtopic, willmsg, willqos, willretain) + ctx.setWill(willtopic, willmsg, willqos, willretain) # Set the verbosity - ctx.set_verbosity(verbosity) + ctx.setVerbosity(verbosity) # Connec to broker await ctx.start() @@ -53,11 +53,11 @@ proc nmqttSub(host="127.0.0.1", port=1883, ssl=false, clientid="", username="", waitFor ctx.publish(t, "", 0, true) # Callback for subscribe - proc on_data(t, msg: string) = + proc onData(t, msg: string) = echo t, ": ", msg # Subscribe to topic - await ctx.subscribe(t, qos, on_data) + await ctx.subscribe(t, qos, onData) if ctx.verbosity >= 1: ctx.dbg "Subscribing to: " & t