Merge pull request #34 from lmn/master

Wildcard Support

Author: ThomasTJdev

nmqtt.nim

@@ -836,7 +836,13 @@ proc onPublish(ctx: MqttCtx, pkt: Pkt) {.async.} =
 
   when not defined(broker):
     for top, cb in ctx.pubCallbacks:
-      if top == topic or top == "#": cb.cb(topic, message)
+      if top == topic or top == "#":
+        cb.cb(topic, message)
+      if top.endsWith("/#"):
+        var topicw = top
+        topicw.removeSuffix("#")
+        if topic.contains(topicw):
+          cb.cb(topic, message)
 
   if qos == 1:
     ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 1, typ: PubAck)

tests/subscribe.nim

@@ -169,14 +169,13 @@ suite "test suite for subscribe":
 
     waitFor conn()
 
-
   test "subscribe to #":
     let (_, msg) = tdata("subscribe to #")
 
     proc conn() {.async.} =
       var msgCount: int
       proc on_data_sub_all(topic: string, message: string) =
-        msgCount += 1
+         msgCount += 1
 
       await ctxListen.subscribe("#", 0, on_data_sub_all)
       await sleepAsync 500
@@ -190,6 +189,25 @@ suite "test suite for subscribe":
 
     waitFor conn()
 
+  test "subscribe to test/#":
+    let (_, msg) = tdata("subscribe to test/#")
+
+    proc conn() {.async.} =
+      var msgCount: int
+      proc on_data_sub_wild(topic: string, message: string) =
+        msgCount += 1
+
+      await ctxListen.subscribe("test/#", 0, on_data_sub_wild)
+      await sleepAsync 500
+      await ctxMain.publish("test/random1", msg, 0)
+      await ctxMain.publish("second/random2", msg, 0)
+      await ctxMain.publish("test/random3", msg, 0)
+      await sleepAsync 500
+      await ctxListen.unsubscribe("test/#")
+      await sleepAsync 500
+      check(msgCount == 2)
+
+    waitFor conn()
 
   test "stay subscribed after disconnect with reconnect":
     let (tpc, msg) = tdata("stay subscribed after disconnect with reconnect")