Merge pull request #170 from mocheng/fix_124_169

send suback 0x80 if subscribe authorization fails

lib/client.js

@@ -387,7 +387,7 @@ Client.prototype.handleAuthorizeSubscribe = function(err, success, s, cb) {
 
   if (!success) {
     this.logger.info({ topic: s.topic }, "subscribe not authorized");
-    cb("not authorized");
+    cb(null, false);
     return;
   }
 
@@ -407,7 +407,7 @@ Client.prototype.handleAuthorizeSubscribe = function(err, success, s, cb) {
       }
       that.logger.info({ topic: s.topic, qos: s.qos }, "subscribed to topic");
       that.subscriptions[s.topic] = { qos: s.qos, handler: handler };
-      cb();
+      cb(null, true);
     }
   );
 };
@@ -423,25 +423,29 @@ Client.prototype.handleSubscribe = function(packet) {
   logger.debug({ packet: packet }, "subscribe received");
 
   var granted = calculateGranted(this, packet);
-  var subs = packet.subscriptions.filter(function(s) {
-    return that.subscriptions[s.topic] === undefined;
-  });
 
-  async.parallel(subs.map(function(s) {
-    return function(cb) {
+  async.map(packet.subscriptions, function(s, cb) {
+    if (that.subscriptions[s.topic] === undefined) {
       server.authorizeSubscribe(that, s.topic, function(err, success) {
         that.handleAuthorizeSubscribe(err, success, s, cb);
       });
-    };
-  }), function(err) {
+    } else {
+      cb(null, true);
+    }
+  }, function(err, authorized) {
+
     if (err) {
       that.close();
       return;
     }
 
-    packet.subscriptions.forEach(function(sub) {
-      that.server.forwardRetained(sub.topic, that);
-      that.server.emit("subscribed", sub.topic, that);
+    packet.subscriptions.forEach(function(sub, index) {
+      if (authorized[index]) {
+        that.server.forwardRetained(sub.topic, that);
+        that.server.emit("subscribed", sub.topic, that);
+      } else {
+        granted[index] = 0x80;
+      }
     });
 
     if(!that._closed) {

test/abstract_server.js

@@ -1204,20 +1204,27 @@ module.exports = function(moscaSettings, createConnection) {
   });
 
   it("should support subscribe authorization (failure)", function(done) {
+    var d = donner(2, done);
+
     instance.authorizeSubscribe = function(client, topic, callback) {
       expect(topic).to.be.eql("hello");
       callback(null, false);
     };
 
-    buildAndConnect(done, function(client) {
+    buildAndConnect(d, function(client) {
 
       var subscriptions = [{
           topic: "hello",
           qos: 0
         }
       ];
 
-      // it exists no negation of auth, it just disconnect the client
+      client.on("suback", function(packet) {
+        expect(packet.granted).to.be.eql([0x80]);
+        client.disconnect();
+        d();
+      });
+
       client.subscribe({
         subscriptions: subscriptions,
         messageId: 42