Moved the wiring to the Server in the persistance itself.

moscajs · mcollina · Jun 26, 2013 · Jun 17, 2013 · Jun 17, 2013 · Jun 18, 2013
commit 718126ed679b2c8437ded582f467d7f2ad16ff78
diff --git a/lib/persistance/abstract.js b/lib/persistance/abstract.js
@@ -0,0 +1,32 @@
+
+function AbstractPersistence() {
+
+}
+
+AbstractPersistence.prototype.wire = function(server) {
+  var that = this;
+
+  server.on("published", function(packet) {
+    if (packet.retain) {
+      that.storeRetained(packet);
+    }
+  });
+
+  server.on("subscribed", function(pattern, client) {
+    that.lookupRetained(pattern, function(err, matches) {
+      if (err) {
+        client.emit("error", err);
+        return;
+      }
+      matches.forEach(function(match) {
+        client.forward(match.topic, match.payload, match, pattern);
+      });
+    });
+  });
+
+  server.on("close", function() {
+    that.close();
+  });
+};
+
+module.exports = AbstractPersistence;
diff --git a/lib/persistance/levelup.js b/lib/persistance/levelup.js
@@ -1,6 +1,8 @@
 
 var levelup = require("levelup");
 var sublevel = require("level-sublevel");
+var AbstractPersistence = require("./abstract");
+var util = require("util");
 
 function LevelUpPersistance(path, options) {
   options = options || {};
@@ -9,6 +11,8 @@ function LevelUpPersistance(path, options) {
   this._retained = this.db.sublevel("retained");
 }
 
+util.inherits(LevelUpPersistance, AbstractPersistence);
+
 LevelUpPersistance.prototype.storeRetained = function(packet, cb) {
   this._retained.put(packet.topic, packet, cb);
 };

diff --git a/lib/persistance/memory.js b/lib/persistance/memory.js
@@ -2,7 +2,7 @@
 var LevelUpPersistance = require("./levelup");
 var util = require("util");
 var MemDOWN = require("memdown");
-var factory = function (location) { return new MemDOWN(location) };
+var factory = function (location) { return new MemDOWN(location); };
 
 function MemoryPersistance() {
   LevelUpPersistance.call(this, "RAM", { db: factory });

diff --git a/lib/server.js b/lib/server.js
@@ -29,7 +29,7 @@ var persistance = require("./persistance");
  *    the client is passed as a parameter.
  *  - `published`, when a new message is published;
  *    the packet and the client are passed as parameters.
- *  - `subcribed`, when a new client is subscribed to a pattern;
+ *  - `subscribed`, when a new client is subscribed to a pattern;
  *    the pattern and the client are passed as parameters.
  *
  * @param {Object} opts The option object
@@ -71,8 +71,6 @@ function Server(opts, callback) {
   this.ascoltatore = ascoltatori.build(this.opts.backend);
   this.ascoltatore.on("error", this.emit.bind(this));
 
-  this.persistance = this.opts.persistance || new persistance.Memory();
-
   that.once("ready", callback);
 
   async.series([
@@ -96,24 +94,6 @@ function Server(opts, callback) {
   that.on("clientDisconnected", function(client) {
     delete that.clients[client.id];
   });
-
-  that.on("published", function(packet) {
-    if (packet.retain) {
-      that.persistance.storeRetained(packet);
-    }
-  });
-
-  that.on("subscribed", function(pattern, client) {
-    that.persistance.lookupRetained(pattern, function(err, matches) {
-      if (err) {
-        client.emit("error", err);
-        return;
-      }
-      matches.forEach(function(match) {
-        client.forward(match.topic, match.payload, match, pattern);
-      });
-    });
-  });
 }
 
 module.exports = Server;

diff --git a/test/persistance/abstract.js b/test/persistance/abstract.js
@@ -1,6 +1,7 @@
 "use strict";
 
 var async = require("async");
+var EventEmitter = require("events").EventEmitter;
 
 module.exports = function(create) {
 
@@ -93,4 +94,51 @@ module.exports = function(create) {
       }
     ], done);
   });
+
+  it("should wire itself up to the 'published' event of a Server", function(done) {
+    var em = new EventEmitter();
+    var packet1 = {
+      topic: "hello/1",
+      qos: 0,
+      payload: "world",
+      messageId: 42,
+      retain: true
+    };
+
+    instance.wire(em);
+
+    em.emit("published", packet1);
+
+    instance.lookupRetained(packet1.topic, function(err, results) {
+      expect(results).to.eql([packet1]);
+      done();
+    });
+  });
+
+  it("should wire itself up to the 'subscribed' event of a Server", function(done) {
+    var em = new EventEmitter();
+    var packet1 = {
+      topic: "hello/1",
+      qos: 0,
+      payload: "world",
+      messageId: 42,
+      retain: true
+    };
+
+    var client = { 
+      forward: function(topic, payload, options, pattern) {
+        expect(topic).to.eql(packet1.topic);
+        expect(payload).to.eql(packet1.payload);
+        expect(options).to.eql(packet1);
+        expect(pattern).to.eql("hello/#");
+        done();
+      }
+    };
+
+    instance.wire(em);
+
+    instance.storeRetained(packet1, function() {
+      em.emit("subscribed", "hello/#", client);
+    });
+  });
 };
diff --git a/test/persistance/levelup_spec.js b/test/persistance/levelup_spec.js
@@ -4,7 +4,7 @@ var abstract = require("./abstract");
 var LevelUp = require("../../").persistance.LevelUp;
 var tmp = require("tmp");
 
-describe("mosca.persitance.LevelUp", function() {
+describe("mosca.persistance.LevelUp", function() {
   abstract(function(cb) {
     tmp.dir(function (err, path) {
       if (err) {

diff --git a/test/persistance/memory_spec.js b/test/persistance/memory_spec.js
@@ -3,7 +3,7 @@
 var abstract = require("./abstract");
 var Memory = require("../../").persistance.Memory;
 
-describe("mosca.persitance.Memory", function() {
+describe("mosca.persistance.Memory", function() {
   abstract(function(cb) {
     cb(null, new Memory());
   });

diff --git a/test/server_spec.js b/test/server_spec.js
@@ -1035,6 +1035,9 @@ describe("mosca.Server", function() {
   });
 
   it("should support retained messages", function(done) {
+    var pers = new mosca.persistance.Memory();
+
+    pers.wire(instance);
 
     async.waterfall([