setMaxInflightMessages

Author: python36

.gitignore

@@ -1,2 +1,5 @@
 tests/tester
-bin
+bin
+nimble.develop
+nimble.paths
+nimbledeps

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,64 +236,76 @@ 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.
 
 
 ____
 
+### 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

config.nims

@@ -0,0 +1,4 @@
+# begin Nimble config (version 2)
+when withDir(thisDir(), system.fileExists("nimble.paths")):
+  include "nimble.paths"
+# end Nimble config

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,40 +1142,44 @@ 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
   ctx.willMsg    = msg
   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
 

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":

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)