support shared subscriptions

nmqtt.nim

@@ -1311,13 +1311,27 @@ proc publish*(ctx: MqttCtx, topic: string, message: string, qos: QoS = 0, retain
 proc subscribe*(ctx: MqttCtx, topic: string, qos: QoS | int, callback: PubCallback.cb): Future[void] =
   ## Subscribe to a topic.
   ##
+  ## Shared subscription topic filter must follow the format: $share/{ShareName}/{filter}
+  ## 
   ## Access the callback with:
   ## .. code-block::nim
   ##    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(qos), typ: Subscribe)
-  ctx.pubCallbacks[topic] = (templ: splitTopic(topic), cb: PubCallback(cb: callback, qos: QoS(qos)))
+  let topicCb = if topic.startswith("$share"):
+    let topicItems = topic.split('/', 3)
+    if topicItems.len < 3:
+      raise newException(ValueError, "Invalid shared topic. Expected '$share/{ShareName}/{filter}', but got fewer than 3 segments.")
+    elif topicItems[1].len == 0 or topicItems[2].len == 0:
+      raise newException(ValueError, "Invalid shared topic. 'ShareName' and 'filter' segments cannot be empty.")
+    elif {'+', '#'} in topicItems[1]: 
+      raise newException(ValueError, "Invalid shared topic. The shared subscription group name must not contain MQTT wildcards ('+' or '#').") 
+    topicItems[2]
+  else:
+    topic
+
+  ctx.pubCallbacks[topic] = (templ: splitTopic(topicCb), cb: PubCallback(cb: callback, qos: QoS(qos)))
   result = ctx.work()
 
 proc unsubscribe*(ctx: MqttCtx, topic: string): Future[void] =

tests/subscribe.nim

@@ -727,3 +727,55 @@ suite "test suite for subscribe":
     waitFor conn()
     waitFor ctxMain.disconnect()
     waitFor ctxSlave.disconnect()
+
+  test "shared subscribe":
+    # Need MQTT broker with support 5.0 protocol. nmqtt broker does not support it yet.
+
+    let
+      ctxMain = newCtx()
+      ctxSlave1 = newCtx()
+      ctxSlave2 = newCtx()
+      (tpc, msg) = tdata("shared subscribtion")
+
+    proc conn() {.async.} =
+      var
+        msgCount1: int
+        msgCount2: int
+
+      proc onData1(topic: string, message: string) =
+        check(topic == tpc)
+        check(message == msg)
+        msgCount1 += 1
+
+      proc onData2(topic: string, message: string) =
+        check(topic == tpc)
+        check(message == msg)
+        msgCount2 += 1
+
+      await ctxSlave1.subscribe("$share/ShareName" & tpc & "/" & tpc, 0, onData1)
+      await ctxSlave2.subscribe("$share/ShareName" & tpc & "/" & tpc, 0, onData2)
+      await sleepAsync(500)
+      await ctxMain.publish(tpc, msg, 0)
+      await ctxMain.publish(tpc, msg, 0)
+      await sleepAsync(500)
+
+      check(msgCount1 == 1)
+      check(msgCount2 == 1)
+
+      expect ValueError:
+        await ctxSlave1.subscribe("$share/ShareName/", 0, onData2)
+      expect ValueError:
+        await ctxSlave1.subscribe("$share//filter", 0, onData2)
+      expect ValueError:
+        await ctxSlave1.subscribe("$share/+/wildcard", 0, onData2)
+      expect ValueError:
+        await ctxSlave1.subscribe("$share/#/wildcard", 0, onData2)
+      expect ValueError:
+        await ctxSlave1.subscribe("$share/#", 0, onData2)
+      expect ValueError:
+        await ctxSlave1.subscribe("$share", 0, onData2)
+
+    waitFor conn()
+    waitFor ctxMain.disconnect()
+    waitFor ctxSlave1.disconnect()
+    waitFor ctxSlave2.disconnect()