Merge pull request #174 from eiabea/master

Added retain option to mqtt

lib/mqtt_ascoltatore.js

@@ -134,7 +134,8 @@ MQTTAscoltatore.prototype.publish = function publish(topic, message, options, do
   this._raiseIfClosed();
 
   this._client.publish(this._pubTopic(topic), message, {
-    qos: (options && (options.qos !== undefined)) ? options.qos : 1
+    qos: (options && (options.qos !== undefined)) ? options.qos : 1,
+    retain: (options && (options.retain !== undefined)) ? options.retain : false
   }, function() {
     debug("new message published to " + topic);
     wrap(done)();

lib/zeromq_ascoltatore.js

@@ -122,7 +122,7 @@ ZeromqAscoltatore.prototype._startControl = function() {
       }, 250);
 
       that._control_conn.on("message", function(data) {
-        var dest = String(data)
+        var dest = String(data);
         debug("received connect response from " + dest);
         if (that._sub_conns.findIndex(function (conn) {
           return conn.port === dest;

test/mqtt_ascoltatore_spec.js

@@ -36,9 +36,10 @@ describeAscoltatore("MQTT", function() {
     var that = this;
     mqttServer.once('published', function(packet) {
       expect(packet.qos).to.eql(0);
+      expect(packet.retain).to.eql(true);
       done();
     });
-    that.instance.publish("hello/123", "42", { qos: 0 });
+    that.instance.publish("hello/123", "42", { qos: 0, retain: true });
   });
 
 });

test/zeromq_ascoltatore_spec.js

@@ -93,21 +93,21 @@ describeAscoltatore("zeromq", function() {
       if (--count === 0) {
         instance._control_conn.removeListener("message", donner);
 
-        var subs = {}
+        var subs = {};
         instance._sub_conns.forEach(function(c) {
           subs[c.port] = true;
         });
         expect(instance._sub_conns.length).to.equal(3);
         expect(Object.keys(subs).length).to.equal(3);
 
-        subs = {}
+        subs = {};
         other._sub_conns.forEach(function(c) {
           subs[c.port] = true;
         });
         expect(other._sub_conns.length).to.equal(3);
         expect(Object.keys(subs).length).to.equal(3);
 
-        subs = {}
+        subs = {};
         other2._sub_conns.forEach(function(c) {
           subs[c.port] = true;
         });
@@ -135,7 +135,7 @@ describeAscoltatore("zeromq", function() {
         instance.connect(other._opts.controlPort);
         instance.connect(other2._opts.controlPort);
         instance._control_conn.on("message", donner);
-        cb()
+        cb();
       }
     ]);
   });