Merge pull request #39 from pkinney/delivered-event

Add 'delivered' event on PUBACK handler

README.md

@@ -91,6 +91,9 @@ Events:
   1. `packet`
   2. `client`, it will be null if the message is published using
      [`publish`](#publish).
+* `deliver`: when a packet published to a client is delivered successfully, arguments:
+  1. `packet`
+  2. `client`
 * `subscribe`: when a client sends a SUBSCRIBE, arguments:
   1. `subscriptions`, as defined in the `subscriptions` property of the
      [SUBSCRIBE](https://github.com/mqttjs/mqtt-packet#subscribe)

lib/handlers/puback.js

@@ -3,6 +3,7 @@
 function handlePuback (client, packet, done) {
   var persistence = client.broker.persistence
   persistence.outgoingClearMessageId(client, packet, done)
+  client.broker.emit('deliver', packet, client)
 }
 
 module.exports = handlePuback

test/client-pub-sub.js

@@ -73,6 +73,77 @@ test('publish direct to a single client QoS 1', function (t) {
   })
 })
 
+test('emit a `deliver` event on PUBACK for QoS 1', function (t) {
+  t.plan(3)
+
+  var broker = aedes()
+  var messageId
+
+  broker.on('client', function (client) {
+    client.publish({
+      topic: 'hello',
+      payload: new Buffer('world'),
+      qos: 1
+    }, function (err) {
+      t.error(err, 'no error')
+    })
+  })
+
+  broker.once('deliver', function (packet, client) {
+    t.equal(packet.messageId, messageId)
+    t.pass('got the deliver event')
+  })
+
+  var s = connect(setup(broker))
+
+  s.outStream.once('data', function (packet) {
+    messageId = packet.messageId
+    s.inStream.write({
+      cmd: 'puback',
+      messageId: packet.messageId
+    })
+  })
+})
+
+test('emit a `deliver` event on PUBCOMP for QoS 2', function (t) {
+  t.plan(3)
+
+  var broker = aedes()
+  var messageId
+
+  broker.on('client', function (client) {
+    client.publish({
+      topic: 'hello',
+      payload: new Buffer('world'),
+      qos: 2
+    }, function (err) {
+      t.error(err, 'no error')
+    })
+  })
+
+  broker.once('deliver', function (packet, client) {
+    t.equal(packet.messageId, messageId)
+    t.pass('got the deliver event')
+  })
+
+  var s = connect(setup(broker))
+
+  s.outStream.on('data', function (packet) {
+    if (packet.cmd === 'publish') {
+      s.inStream.write({
+        cmd: 'pubrec',
+        messageId: packet.messageId
+      })
+    } else {
+      messageId = packet.messageId
+      s.inStream.write({
+        cmd: 'pubcomp',
+        messageId: packet.messageId
+      })
+    }
+  })
+})
+
 test('offline message support for direct publish', function (t) {
   t.plan(2)