Merge pull request #545 from happybin92/master

Fixbug (subscribes with same topic, may lead to the ascoltatore added multi cbs, and can't unsubscribe)

lib/client.js

@@ -443,19 +443,25 @@ Client.prototype.handleAuthorizeSubscribe = function(err, success, s, cb) {
     that.forward(topic, payload, options, s.topic, s.qos);
   };
 
-  this.server.ascoltatore.subscribe(
-    s.topic,
-    handler,
-    function(err) {
-      if (err) {
-        cb(err);
-        return;
+  if (this.subscriptions[s.topic] === undefined) {
+    this.subscriptions[s.topic] = { qos: s.qos, handler: handler };
+    this.server.ascoltatore.subscribe(
+      s.topic,
+      handler,
+      function(err) {
+        if (err) {
+          delete that.subscriptions[s.topic];
+          cb(err);
+          return;
+        }
+        that.logger.info({ topic: s.topic, qos: s.qos }, "subscribed to topic");
+        //that.subscriptions[s.topic] = { qos: s.qos, handler: handler };
+        cb(null, true);
       }
-      that.logger.info({ topic: s.topic, qos: s.qos }, "subscribed to topic");
-      that.subscriptions[s.topic] = { qos: s.qos, handler: handler };
-      cb(null, true);
-    }
-  );
+    );
+  } else {
+    cb(null, true);
+  }
 };
 
 function handleEachSub (s, cb) {

test/server.js

@@ -169,6 +169,61 @@ describe("mosca.Server", function() {
     });
   });
 
+  it("should not receive the publish after unsubscription, while multi subscriptions with the same topic", function(done) {
+
+    // Simulate a situation that it takes same time to do authorizeSubscribe.
+    this.instance.authorizeSubscribe = function(client, topic, callback) {
+      setTimeout(function(){
+        callback(null, true)
+      }, 300);
+    };
+
+    buildAndConnect(function(){}, this.instance, function(client) {
+        function subAction(){
+          var messageId = Math.floor(65535 * Math.random());
+          client.subscribe({
+            subscriptions: [{topic: "hello", qos: 1 }],
+            messageId: messageId
+          });
+        }
+
+        var subCount = 3;  // subscribe the same topic for 3 times
+        for (var i = 0; i < subCount; ++i)
+          subAction();
+
+        var subackCount = 0;
+        client.on("suback", function() { // unsubscribe after subscriptions
+          subackCount++;
+          if (subackCount == subCount) { 
+            var messageId = Math.floor(65535 * Math.random());
+            client.unsubscribe({
+              unsubscriptions: ["hello"],
+              messageId: messageId
+            });
+          }
+        });
+
+        client.on("unsuback", function() { // publish message after unsubscription
+          var messageId = Math.floor(65535 * Math.random());
+          client.publish({
+            topic: "hello",
+            payload: "some data",
+            messageId: messageId,
+            qos: 1
+          });
+        });
+
+        client.on("publish", function(packet) { // should not receive the publish
+          done(new Error("unexpected publish"));
+        });
+
+        client.on("puback", function(packet) { // close client when puback
+          client.disconnect();
+          done();
+        });
+    });
+  });
+
   it("should fail if persistence can not connect", function (done) {
     var newSettings = moscaSettings();