Fixed missing message in RedisAscoltatore.

Sending a message on topic 'aa' was not forwarded
with subscribers of 'aa/#'. This was due to some missing topic-rewriting
business which have been added.

lib/behave_like_an_ascoltatore.js

@@ -92,6 +92,13 @@ module.exports = function() {
     });
   });
 
+  it("should support multi-level wildcard at the end of a topic with no separator", function(done) {
+    var that = this;
+    that.instance.sub("hello/*", wrap(done), function() {
+      that.instance.pub("hello");
+    });
+  });
+
   it("should support single-level wildcard at start of topic", function(done) {
     var that = this;
     that.instance.sub("+/hello", wrap(done), function() {

lib/redis_ascoltatore.js

@@ -130,9 +130,13 @@ RedisAscoltatore.prototype._startSub = function() {
         // we need to skip out this callback, so we do not
         // break the client when an exception occurs
         var ascoltatore = that._ascoltatores[sub];
+        topic = topic.replace(new RegExp('/$'), '');
+        topic = that._recvTopic(topic);
 
         if (ascoltatore) {
-          ascoltatore.publish(that._recvTopic(topic), message);
+          ascoltatore.publish(topic, message);
+        } else {
+          console.log("no ascoltatore for subscription: " + sub);
         }
       });
     };
@@ -170,12 +174,17 @@ RedisAscoltatore.prototype.subscribe = function subscribe(topic, callback, done)
 
   var subTopic = this._subTopic(topic);
 
+  if (!subTopic.match(new RegExp('[/*]$'))) {
+    subTopic += '/';
+  }
+
   if (this._containsWildcard(topic)) {
     this._sub.psubscribe(subTopic, newDone);
   } else {
     this._sub.subscribe(subTopic, newDone);
   }
 
+  debug("redis subscription topic is " + subTopic);
   this._subs_counter.add(subTopic);
 
   var ascoltatore = this._ascoltatores[subTopic];
@@ -198,7 +207,13 @@ RedisAscoltatore.prototype.publish = function publish(topic, message, done) {
     message = false; // so we can convert it to JSON
   }
 
-  this._client.publish(this._pubTopic(topic), message, function() {
+  topic = this._pubTopic(topic);
+
+  if (!topic.match(new RegExp('/$'))) {
+    topic += '/';
+  }
+
+  this._client.publish(topic, message, function() {
     debug("new message published to " + topic);
     util.defer(done);
   });
@@ -210,6 +225,10 @@ RedisAscoltatore.prototype.unsubscribe = function unsubscribe(topic, callback, d
   var isWildcard = this._containsWildcard(topic),
       subTopic = this._subTopic(topic);
 
+  if (!subTopic.match(new RegExp('[/*]$'))) {
+    subTopic += '/';
+  }
+
   this._subs_counter.remove(subTopic);
 
   var ascoltatore = this._ascoltatores[subTopic];