Repository: moscajs/aedes (js)

Issue:
[bug]subscribe Event will triggered while topic is negated.
**System Information**
- Aedes: 0.41.0
- NodeJS: 10.15.3
- OS: Centos 7

**Describe the bug**
Subscribe Event will triggered while topic is negated.

**To Reproduce**
```
//server.js
const aedes = require('aedes')();

aedes.authorizeSubscribe = function (client, sub, callback) {
    if (sub.topic === 'presence') {
        sub = null
    }
    callback(null, sub)
}

aedes.on("subscribe", (subs, client)=>{
    console.log(`subscribe event`, subs) // triggered
})


const server = require('net').createServer(aedes.handle)
const port = 1883

server.listen(port, function () {
    console.log('server started and listening on port ', port)
})
```

```
//client.js
const mqtt = require('mqtt')
let client  = mqtt.connect({
    servers: [{
        host: 'localhost',
        port: 1883,
        protocol: 'tcp'
    }]
});

client.on('connect', function () {
    client.subscribe('presence', function (err, granted) {
        if (!err) {
            console.log(granted) //[ { topic: 'presence', qos: 128 } ]
        }else
            console.log(err)
    })
});
```


Pull request description:
Return granted qos in broker subscribe event and better subAck docs
Fixes #467

Patch:
diff --git a/docs/Aedes.md b/docs/Aedes.md
index 5926fcd..18ffc9f 100644
--- a/docs/Aedes.md
+++ b/docs/Aedes.md
@@ -138,7 +138,7 @@ Emitted when `client` sends a `PINGREQ`.
 
 Emitted when `client` successfully subscribe the `subscriptions` in server.
 
-`subscriptions` are an array of `{ topic: topic, qos: qos }`
+`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)
 
 ## Event: unsubscribe
 
@@ -360,7 +360,7 @@ aedes.authorizeSubscribe = function (client, sub, callback) {
 }
 ```
 
-To negate a subscription, set the subscription to `null`:
+To negate a subscription, set the subscription to `null`. Aedes ignores the negated subscription and the `qos` in `SubAck` is set to `128` based on [MQTT 3.11 spec](https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html#_Toc385349323):
 
 ```js
 aedes.authorizeSubscribe = function (client, sub, callback) {
diff --git a/lib/handlers/subscribe.js b/lib/handlers/subscribe.js
index 886eab1..3131eba 100644
--- a/lib/handlers/subscribe.js
+++ b/lib/handlers/subscribe.js
@@ -19,6 +19,7 @@ const restoreTopicActions = fastfall([
 function SubAck (packet, granted) {
   this.cmd = 'suback'
   this.messageId = packet.messageId
+  // the qos granted
   this.granted = granted
 }
 
@@ -27,7 +28,7 @@ function Subscription (qos, func) {
   this.func = func
 }
 
-function SubscribeState (client, packet, restore, finish, granted) {
+function SubscribeState (client, packet, restore, finish) {
   this.client = client
   this.packet = packet
   this.actions = restore ? restoreTopicActions : subscribeTopicActions
@@ -56,11 +57,11 @@ function _dedupe (subs) {
 }
 
 function handleSubscribe (client, packet, restore, done) {
-  const subs = packet.subscriptions
+  packet.subscriptions = packet.subscriptions.length === 1 ? packet.subscriptions : _dedupe(packet.subscriptions)
   client.broker._parallel(
     new SubscribeState(client, packet, restore, done),
     doSubscribe,
-    subs.length === 1 ? subs : _dedupe(subs),
+    packet.subscriptions,
     restore ? done : completeSubscribe)
 }
 
@@ -92,6 +93,7 @@ function blockDollarSignTopics (func) {
 
 function storeSubscriptions (sub, done) {
   if (!sub || typeof sub !== 'object') {
+    // means failure: MQTT 3.1.1 specs > 3.9.3 Payload
     this.granted = 128
     return done(null, null)
   }
@@ -161,11 +163,19 @@ function completeSubscribe (err) {
   } else {
     done()
   }
-  this.subState = []
 
   const broker = client.broker
   const subs = packet.subscriptions
 
+  var topics = []
+
+  for (var i = 0; i < subs.length; i++) {
+    topics.push(subs[i].topic)
+    subs.qos = this.subState[i].granted
+  }
+
+  this.subState = []
+
   broker.emit('subscribe', subs, client)
 
   // Conform to MQTT 3.1.1 section 3.1.2.4
@@ -173,10 +183,6 @@ function completeSubscribe (err) {
   // Retained message should be only fetched from SUBSCRIBE.
 
   const persistence = broker.persistence
-  var topics = []
-  for (var i = 0; i < subs.length; i++) {
-    topics.push(subs[i].topic)
-  }
   const stream = persistence.createRetainedStreamCombi(topics)
   stream.pipe(through.obj(function sendRetained (packet, enc, cb) {
     packet = new Packet({


Test patch:
diff --git a/test/client-pub-sub.js b/test/client-pub-sub.js
index e0a5d60..92b9147 100644
--- a/test/client-pub-sub.js
+++ b/test/client-pub-sub.js
@@ -714,7 +714,7 @@ test('get message when client disconnects', function (t) {
 })
 
 test('should not receive a message on negated subscription', function (t) {
-  t.plan(2)
+  t.plan(4)
 
   const broker = aedes()
   t.tearDown(broker.close.bind(broker))
@@ -731,15 +731,24 @@ test('should not receive a message on negated subscription', function (t) {
       retain: true
     }, function (err) {
       t.error(err, 'no error')
-      client.subscribe({
+      client.subscribe([{
         topic: 'hello',
         qos: 0
-      }, function (err) {
+      },
+      {
+        topic: 'hello',
+        qos: 0
+      }], function (err) {
         t.error(err, 'no error')
       })
     })
   })
 
+  broker.on('subscribe', function (subs) {
+    t.pass(subs.length, 1, 'Should dedupe subs')
+    t.pass(subs[0].qos, 128, 'Qos should be 128 (Fail)')
+  })
+
   const s = connect(setup(broker))
   s.outStream.once('data', function (packet) {
     t.fail('Packet should not be received')