Merge pull request #77 from mcollina/v0.14.3-wip

V0.14.3 wip

lib/client.js

@@ -512,11 +512,20 @@ Client.prototype.close = function(callback) {
     }
   };
 
+  that.connection.stream.on('end' , function(){
+    that._streamClosedRequiresCleanup = true;
+  });
+
   that._closing = true;
+
   async.parallel(Object.keys(that.subscriptions).map(that.unsubscribeMapTo.bind(that)), function() {
     that.server.persistClient(that);
-    that.connection.stream.on('end', cleanup);
-    that.connection.stream.end();
+    if(that._streamClosedRequiresCleanup){
+      cleanup();
+    } else {
+      that.connection.stream.on('end', cleanup);
+      that.connection.stream.end();
+    }
   });
 };
 

lib/persistence/mongo.js

@@ -155,14 +155,24 @@ MongoPersistence.prototype.lookupSubscriptions = function(client, done) {
 };
 
 MongoPersistence.prototype.storeRetained = function(packet, cb) {
-  var that = this;
-  this._retained.remove({ topic: packet.topic }, function(err) {
-    that._retained.insert(packet, function(err) {
-      if (cb) {
+
+  this._retained.findAndModify(
+    { topic: packet.topic },
+    [],
+    packet,
+    {
+      upsert: true,
+      new: true
+    },
+    function(err, result){
+      if(!err) {
+        packet._id = result._id;
+      }
+      if(cb) {
         return cb(err);
       }
     });
-  });
+
 };
 
 MongoPersistence.prototype.lookupRetained = function(pattern, cb) {

package.json

@@ -46,15 +46,16 @@
     "chai": "~1.8.0",
     "sinon": "~1.7.3",
     "sinon-chai": "~2.1.2",
-    "async_bench": "0.2.1",
     "underscore": "~1.4.4",
     "dox-foundation": "~0.5.4",
     "jshint": "~2.3.0",
+    "microtime": "~0.5.0",
     "js-beautify": "~0.4.2",
     "tmp": "0.0.22",
     "supertest": "~0.8.1",
     "coveralls": "~2.5.0",
-    "istanbul": "~0.1.45"
+    "istanbul": "~0.1.45",
+    "async_bench": "~0.3.0"
   },
   "dependencies": {
     "mqtt": "~0.3.7",
@@ -85,6 +86,6 @@
     "zmq": "~2.4.0",
     "amqp": "~0.1.8",
     "redis": "~0.8.2",
-    "mongodb": "~1.3.20"
+    "mongodb": "~1.3.19"
   }
 }

test/abstract_server.js

@@ -1271,10 +1271,9 @@ module.exports = function(moscaSettings, createConnection) {
 
           client.on('connack', function(packet) {
             var subscriptions = [{
-                topic: "hello",
-                qos: 0
-              }
-            ];
+              topic: "hello",
+              qos: 0
+            }];
 
             client.subscribe({
               subscriptions: subscriptions,
@@ -1293,6 +1292,109 @@ module.exports = function(moscaSettings, createConnection) {
     ], done);
   });
 
+  it("should return only a single retained message", function(done) {
+
+    var pers = new mosca.persistence.Memory();
+
+    pers.wire(instance);
+
+    async.waterfall([
+
+      function(cb) {
+        var client = createConnection(settings.port, settings.host);
+        client.name = "Phase 1";
+        var defaultMessage = {
+          topic: "hello",
+          qos: 0,
+          payload: null,
+          messageId: null,
+          retain: true
+        };
+
+        client.on("connected", function() {
+          var opts = buildOpts();
+          opts.clean = true;
+
+          var totalMessages = 10;
+          var publishCount = 0;
+
+          client.connect(opts);
+
+          client.on('publish', function(packet){
+            publishCount++;
+            if(publishCount == totalMessages) {
+              client.stream.end();
+              cb();
+            }
+
+          });
+
+          client.on('connack', function(packet) {
+
+            var subscriptions = [{
+              topic: "hello",
+              qos: 0
+            }];
+
+            client.subscribe({
+              subscriptions: subscriptions,
+              messageId: 20
+            });
+
+            for(var c = 1 ; c <= 10 ; c++)
+            {
+              defaultMessage.payload = (c == totalMessages) ? new Buffer("Final Message") : new Buffer("Message " + c);
+              defaultMessage.messageId = 40 + c;
+              client.publish(defaultMessage);
+            }
+
+          });
+        });
+      },
+
+      function(cb) {
+        var client = createConnection(settings.port, settings.host);
+
+        var retainedReceivedCount = 0;
+
+        client.on("connected", function() {
+          var opts = buildOpts();
+          opts.clean = true;
+
+          client.connect(opts);
+
+          client.on('connack', function(packet) {
+            var subscriptions = [{
+              topic: "hello",
+              qos: 0
+            }
+            ];
+
+            client.subscribe({
+              subscriptions: subscriptions,
+              messageId: 20
+            });
+          });
+
+          var handleTimeout = function() {
+            expect(retainedReceivedCount).to.be.equal(1);
+            client.stream.end();
+            cb();
+          };
+
+          var timeout;
+
+          client.on("publish", function(packet) {
+            clearInterval(timeout);
+            timeout = setTimeout(handleTimeout, 100);
+            retainedReceivedCount ++;
+          });
+
+        });
+      }
+    ], done);
+  });
+
   it("should support unclean clients", function(done) {
     var pers = new mosca.persistence.Memory();
     var opts = buildOpts();

test/cli_spec.js

@@ -2,6 +2,7 @@ var async = require("async");
 var tmp = require('tmp');
 var fs = require("fs");
 var mqtt = require("mqtt");
+var os = require("os");
 
 var SECURE_KEY = __dirname + '/secure/tls-key.pem';
 var SECURE_CERT = __dirname + '/secure/tls-cert.pem';
@@ -318,12 +319,15 @@ describe("mosca.cli", function() {
     });
   });
 
-  it("should reload the config using if killed with SIGHUP", function(done) {
+  it("should reload the current config if killed with SIGHUP on a Linux-based OS", function(done) {
+
+    if(os.platform() === "win32") return done();
+
     args.push("adduser");
     args.push("myuser");
     args.push("mypass");
     args.push("--credentials");
-    
+
     var cloned = null;
 
     async.waterfall([

test/persistence/abstract.js

@@ -84,6 +84,39 @@ module.exports = function(create, buildOpts) {
       ], done);
     });
 
+    it("should match and load a single retained message", function(done) {
+
+      var packetMessageId = 0;
+
+      var getPacket = function(){
+
+        packetMessageId++;
+
+        return {
+          topic: "hello",
+          qos: 0,
+          payload: new Buffer("world"),
+          messageId: packetMessageId,
+          retain: true
+        };
+      };
+
+      var instance = this.instance;
+
+      async.parallel([
+        function(cb) { instance.storeRetained(getPacket(), cb); },
+        function(cb) { instance.storeRetained(getPacket(), cb); },
+        function(cb) { instance.storeRetained(getPacket(), cb); },
+        function(cb) { instance.storeRetained(getPacket(), cb); },
+        function(cb) { instance.storeRetained(getPacket(), cb); }
+      ], function(err, results) {
+          instance.lookupRetained("hello", function(err, results) {
+            expect(results.length).to.be.eql(1);
+            done();
+          });
+      });
+    });
+
     it("should overwrite a retained message", function(done) {
       var packet = {
         topic: "hello",

test/persistence/mongo_spec.js

@@ -2,7 +2,6 @@
 
 var abstract = require("./abstract");
 var Mongo = require("../../").persistence.Mongo;
-var redis = require("redis");
 var MongoClient = require('mongodb').MongoClient;
 var async = require("async");
 

test/static/test

@@ -1 +1 @@
-42
+42

test/websocket_secure_spec.js

@@ -53,7 +53,7 @@ describe("mosca.Server - Secure Websocket", function() {
     var curPort = nextPort() - 1;
     var req = request("https://localhost:" + curPort);
 
-    req.get('/test').expect(200, "42\n").end(done);
+    req.get('/test').expect(200, "42").end(done);
   });
 
   it("should serve a browserify bundle", function(done) {

test/websocket_spec.js

@@ -34,7 +34,7 @@ describe("mosca.Server - Websocket", function() {
     var curPort = nextPort() - 1;
     var req = request("http://localhost:" + curPort);
 
-    req.get('/test').expect(200, "42\n").end(done);
+    req.get('/test').expect(200, "42").end(done);
   });
 
   it("should serve a browserify bundle", function(done) {