Repository: moscajs/aedes (js)

Issue:
[feat] Could it support client subscribe/unsubscribe notification to other client?
I used mosca.js before, which provide the topic "$SYS/+/new/subscribes"  "$SYS/+/new/unsubscribes"   to let my service client knows that the app client had subscribed the particular topic, but in aedes  it did not publish this message with similar topic. 


Pull request:
feat: add $SYS/<brokerId>/new /subscribes and $SYS/<brokerId>/new /unsubscribes #540
Fixes #540 

Patch:
diff --git a/docs/Aedes.md b/docs/Aedes.md
index 9a169d3..832ef37 100644
--- a/docs/Aedes.md
+++ b/docs/Aedes.md
@@ -141,6 +141,8 @@ Emitted when `client` successfully subscribe the `subscriptions` in server.
 
 `subscriptions` is an array of `{ topic: topic, qos: qos }`. The array excludes duplicated topics and includes negated subscriptions where `qos` equals to `128`. See more on [authorizeSubscribe](#handler-authorizesubscribe-client-subscription-callback)
 
+Server publishes a SYS topic `$SYS/<aedes.id>/new/subscribers` to inform a client successfully subscribed to one or more topics. The payload is a JSON that has `clientId` and `subs` props, `subs` equals to `subscriptions` array.
+
 ## Event: unsubscribe
 
 - `unsubscriptions` `Array<string>`
@@ -150,6 +152,8 @@ Emitted when `client` successfully unsubscribe the `subscriptions` in server.
 
 `unsubscriptions` are an array of unsubscribed topics.
 
+Server publishes a SYS topic `$SYS/<aedes.id>/new/unsubscribers` to inform a client successfully unsubscribed to one or more topics. The payload is a JSON that has `clientId` and `subs` props, `subs` equals to `unsubscriptions` array.
+
 ## Event: connackSent
 
 - `packet` `<object>` [`CONNACK`][CONNACK]
diff --git a/lib/handlers/subscribe.js b/lib/handlers/subscribe.js
index 7163fa6..7db31fd 100644
--- a/lib/handlers/subscribe.js
+++ b/lib/handlers/subscribe.js
@@ -177,6 +177,13 @@ function completeSubscribe (err) {
   this.subState = []
 
   broker.emit('subscribe', subs, client)
+  broker.publish({
+    topic: '$SYS/' + broker.id + '/new/subscribes',
+    payload: Buffer.from(JSON.stringify({
+      clientId: client.id,
+      subs: subs
+    }), 'utf8')
+  }, noop)
 
   // Conform to MQTT 3.1.1 section 3.1.2.4
   // Restored sessions should not contain any retained message.
@@ -198,4 +205,6 @@ function completeSubscribe (err) {
   }))
 }
 
+function noop () {}
+
 module.exports = handleSubscribe
diff --git a/lib/handlers/unsubscribe.js b/lib/handlers/unsubscribe.js
index c616fc4..a1c093c 100644
--- a/lib/handlers/unsubscribe.js
+++ b/lib/handlers/unsubscribe.js
@@ -89,7 +89,16 @@ function completeUnsubscribe (err) {
 
   if ((!client.closed || client.clean === true) && packet.unsubscriptions.length > 0) {
     client.broker.emit('unsubscribe', packet.unsubscriptions, client)
+    client.broker.publish({
+      topic: '$SYS/' + client.broker.id + '/new/unsubscribes',
+      payload: Buffer.from(JSON.stringify({
+        clientId: client.id,
+        subs: packet.unsubscriptions
+      }), 'utf8')
+    }, noop)
   }
 }
 
+function noop () {}
+
 module.exports = handleUnsubscribe


Test patch:
diff --git a/test/basic.js b/test/basic.js
index 893ffc0..985e3f0 100644
--- a/test/basic.js
+++ b/test/basic.js
@@ -550,6 +550,43 @@ test('publish to $SYS/broker/new/clients', function (t) {
   })
 })
 
+test('publish to $SYS/broker/new/subsribers and $SYS/broker/new/unsubsribers', function (t) {
+  t.plan(7)
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const sub = {
+    topic: 'hello',
+    qos: 0
+  }
+
+  broker.mq.on('$SYS/' + broker.id + '/new/subscribes', function (packet, done) {
+    var payload = JSON.parse(packet.payload.toString())
+    t.equal(payload.clientId, 'abcde', 'clientId matches')
+    t.deepEqual(payload.subs, [sub], 'subscriptions matches')
+    done()
+  })
+
+  broker.mq.on('$SYS/' + broker.id + '/new/unsubscribes', function (packet, done) {
+    var payload = JSON.parse(packet.payload.toString())
+    t.equal(payload.clientId, 'abcde', 'clientId matches')
+    t.deepEqual(payload.subs, [sub.topic], 'unsubscriptions matches')
+    done()
+  })
+
+  var subscriber = connect(setup(broker), {
+    clean: false, clientId: 'abcde'
+  }, function () {
+    subscribe(t, subscriber, sub.topic, sub.qos, function () {
+      subscriber.inStream.write({
+        cmd: 'unsubscribe',
+        messageId: 43,
+        unsubscriptions: ['hello']
+      })
+    })
+  })
+})
+
 test('restore QoS 0 subscriptions not clean', function (t) {
   t.plan(5)
 
diff --git a/test/retain.js b/test/retain.js
index 134ed9d..7a723de 100644
--- a/test/retain.js
+++ b/test/retain.js
@@ -208,7 +208,7 @@ test('new QoS 0 subscribers receive QoS 0 retained messages when clean', functio
   })
 
   clock.setTimeout(() => {
-    t.equal(broker.counter, 6)
+    t.equal(broker.counter, 8)
   }, 200)
 })
 
@@ -248,7 +248,7 @@ test('new QoS 0 subscribers receive downgraded QoS 1 retained messages when clea
     })
   })
   broker.on('closed', function () {
-    t.equal(broker.counter, 6)
+    t.equal(broker.counter, 7)
   })
 })