event 'published' not enough for QoS 1

lib/client.js

@@ -432,14 +432,16 @@ Client.prototype.handleAuthorizePublish = function(err, success, packet) {
     function() {
       that.logger.info({ packet: packet }, "published packet");
 
-      if (packet.qos === 1 && !(that._closed || that._closing)) {
-        that.connection.puback({
-          messageId: packet.messageId
-        });
-      }
-
-      that.server.storePacket(packet);
-      that.server.emit("published", packet, that);
+      that.server.published(packet, that, function() {
+        if (packet.qos === 1 && !(that._closed || that._closing)) {
+          that.connection.puback({
+            messageId: packet.messageId
+          });
+        }
+
+        that.server.storePacket(packet);
+        that.server.emit("published", packet, that);
+      });
     }
   );
 };

lib/server.js

@@ -235,6 +235,21 @@ Server.prototype.authenticate = function(client, username, password, callback) {
   callback(null, true);
 };
 
+/**
+ * The function that is called before after receiving a publish message but before
+ * answering with puback for QoS 1 packet.
+ * This default implementation does nothing
+ * Override at will
+ *
+ * @api public
+ * @param {Object} packet The MQTT packet
+ * @param {Object} client The MQTTConnection that is a client
+ * @param {Function} callback The callback to send the puback
+ */
+Server.prototype.published = function(packet, client, callback) {
+  callback(null);
+};
+
 /**
  * The function that will be used to authorize clients to publish to topics.
  * This default implementation authorize everybody.

package.json

@@ -65,7 +65,8 @@
     "connect": "~2.8.8",
     "express": "~3.3.8",
     "browserify-middleware": "~1.19.0",
-    "json-buffer": "~2.0.7"
+    "json-buffer": "~2.0.7",
+    "brfs": "0.0.8"
   },
   "optionalDependencies": {
     "leveldown": "~0.9.0",

test/abstract_server.js

@@ -27,7 +27,6 @@ module.exports = function(moscaSettings, createConnection) {
     });
   });
 
-
   function buildClient(done, callback) {
     var client = createConnection(settings.port, settings.host);
 
@@ -516,7 +515,6 @@ module.exports = function(moscaSettings, createConnection) {
     });
   });
 
-
   it("should emit an event on every newly published packet", function(done) {
     buildAndConnect(done, function(client) {
 
@@ -534,6 +532,34 @@ module.exports = function(moscaSettings, createConnection) {
     });
   });
 
+  it("should call onPublished on every newly published packet", function(done) {
+    var onPublishedCalled = false;
+
+    instance.published = function(packet, serverClient, callback) {
+      onPublishedCalled = true;
+
+      expect(packet.topic).to.be.equal("hello");
+      expect(packet.payload.toString()).to.be.equal("some data");
+      expect(serverClient).not.to.be.equal(undefined);
+
+      callback();
+    };
+
+    buildAndConnect(done, function(client) {
+      client.publish({
+        messageId: 42,
+        topic: "hello",
+        payload: "some data",
+        qos: 1
+      });
+
+      client.on("puback", function() {
+        expect(onPublishedCalled).to.eql(true);
+        client.disconnect();
+      });
+    });
+  });
+
   it("should emit an event when a new client is connected", function(done) {
     buildClient(done, function(client) {