Fix RedisAscoltatore and default options to {}

.travis.yml

@@ -11,7 +11,7 @@ before_install:
  - sudo /sbin/ldconfig
  - cd ..
 before_script:
- - (cd node_modules/mosca/node_modules && rm -rf ascoltatori && ln -s ../../.. ascoltatori)
+ - (cd node_modules/mosca/node_modules && rm -rf ascoltatori && mkdir ascoltatori && tar -c --exclude node_modules ../../.. | tar -x -C ascoltatori)
 script:
  - make publish-coverage
 language: node_js

lib/abstract_ascoltatore.js

@@ -114,7 +114,9 @@ AbstractAscoltatore.prototype._setPublish = function() {
     if ((typeof options === 'function') ||
         (done && (typeof done !== 'function'))) {
       done = options;
-      options = undefined;
+      options = {};
+    } else if ((options === undefined) || (options === null)) {
+      options = {};
     }
 
     return publish.call(this, topic, payload, options, done);

lib/behave_like_an_ascoltatore.js

@@ -447,4 +447,45 @@ module.exports = function() {
       }
     ]);
   });
+
+  it("should not deliver message twice for multiple subscriptions using wildcards", function(done) {
+    var that = this,
+      count = 3,
+      sub = null,
+      d = null;
+
+    d = function() {
+      count = count - 1;
+      if (count === 0) {
+        done();
+      }
+    };
+
+    sub = function(topic, cb) {
+      var called = false;
+      that.instance.sub(topic, function() {
+        expect(called).to.equal(false);
+        called = true;
+        d();
+      }, cb);
+    };
+
+    async.series([
+      function (cb) {
+        sub("a/+", cb);
+      },
+
+      function (cb) {
+        sub("+/b", cb);
+      },
+
+      function (cb) {
+        sub("a/b", cb);
+      },
+
+      function(cb) {
+        that.instance.pub("a/b", "ahha", cb);
+      }
+    ]);
+  });
 };

lib/redis_ascoltatore.js

@@ -36,7 +36,20 @@ function RedisAscoltatore(opts) {
   this._opts = opts || {};
   this._opts.redis = this._opts.redis || require("redis");
 
-  this._ascoltatore = new TrieAscoltatore();
+  this._ascoltatores = {};
+
+  this._domain = null;
+
+  var that = this;
+  this._ascoltatore = {
+    registerDomain: function (domain) {
+      that._domain = domain;
+      for (var subTopic in that._ascoltatores) {
+        that._ascoltatores[subTopic].registerDomain(domain);
+      }
+    }
+  };
+
   this._startSub();
   this._startPub();
 
@@ -107,18 +120,24 @@ RedisAscoltatore.prototype._startSub = function() {
       that._updateReady("_ready_sub");
     });
 
-    handler = function(topic, message) {
+    handler = function(sub, topic, message) {
       debug("new message received for topic " + topic);
       util.defer(function() {
         // we need to skip out this callback, so we do not
         // break the client when an exception occurs
-        that._ascoltatore.publish(topic, message);
+        var ascoltatore = that._ascoltatores[sub];
+
+        if (ascoltatore) {
+            ascoltatore.publish(topic, message);
+        }
       });
     };
 
-    this._sub.on("message", handler);
+    this._sub.on("message", function (topic, message) {
+      handler(topic, topic, message);
+    });
     this._sub.on("pmessage", function(sub, topic, message) {
-      handler(topic, message);
+      handler(sub, topic, message);
     });
   }
 
@@ -155,7 +174,17 @@ RedisAscoltatore.prototype.subscribe = function subscribe(topic, callback, done)
 
   this._subs_counter.add(subTopic);
 
-  this._ascoltatore.subscribe(topic, callback);
+  var ascoltatore = this._ascoltatores[subTopic];
+
+  if (!ascoltatore) {
+    ascoltatore = this._ascoltatores[subTopic] = new TrieAscoltatore();
+
+    if (this._domain) {
+      ascoltatore.registerDomain(this._domain);
+    }
+  }
+
+  ascoltatore.subscribe(topic, callback);
 };
 
 RedisAscoltatore.prototype.publish = function publish(topic, message, done) {
@@ -182,7 +211,12 @@ RedisAscoltatore.prototype.unsubscribe = function unsubscribe(topic, callback, d
   }
 
   this._subs_counter.remove(subTopic);
-  this._ascoltatore.unsubscribe(topic, callback);
+
+  var ascoltatore = this._ascoltatores[subTopic];
+
+  if (ascoltatore) {
+    ascoltatore.unsubscribe(topic, callback);
+  }
 
   var newDone = function() {
     debug("deregistered subscriber for topic " + topic);
@@ -194,6 +228,11 @@ RedisAscoltatore.prototype.unsubscribe = function unsubscribe(topic, callback, d
     return this;
   }
 
+  if (ascoltatore) {
+    ascoltatore.close();
+    delete this._ascoltatores[subTopic];
+  }
+
   if (isWildcard) {
     this._sub.punsubscribe(subTopic, newDone);
   } else {
@@ -233,7 +272,12 @@ RedisAscoltatore.prototype.close = function close(done) {
       closes = closes - 1;
     }
   });
-  this._ascoltatore.close();
+
+  for (var subTopic in this._ascoltatores) {
+    this._ascoltatores[subTopic].close();
+  }
+  this._ascoltatores = {};
+
   this.emit("closed");
 };