Packet in Broker ack event

[gnought]

In https://github.com/mcollina/aedes/blob/master/lib/handlers/puback.js#L6

Broker receives puback or pubcomp packet will go into puback.js

In non-clean QoS 1, the origPacket will be a PUBLISH packet
In non-clean QoS 2, the origPacket will be a PUBREL packet.

ack: when a packet published to a client is delivered successfully with QoS 1 or QoS 2

A little bit confusion, should the packet in ack event always be the PUBLISH one?

In a clean QoS 1 and QoS 2, the origPacket will be undefined. That means we will not get any packet info in ack event, is it by design?

[mcollina]

Can you add a code example to explain? I wrote this a long ago, and I do not remember / or I have time to figure this out.

[gnought]

TAP version 13
# subscribe & publish QoS 1 with clean=true
ok 1 should be equal
ok 2 should be equivalent
ok 3 should be equal
QoS 1 [clean=true] ack undefined

# subscribe & publish QoS 2 with clean=true
ok 4 should be equal
ok 5 should be equivalent
ok 6 should be equal
QoS 2 [clean=true] ack undefined

# subscribe & publish QoS 1 with clean=false
ok 7 should be equal
ok 8 should be equivalent
ok 9 should be equal
QoS 1 [clean=false] ack Packet {
  cmd: 'publish',
  brokerId: 'Os5WarRx4_',
  brokerCounter: 3,
  topic: 'hello',
  payload: <Buffer 77 6f 72 6c 64>,
  qos: 1,
  retain: false,
  messageId: 64566
}

# subscribe & publish QoS 2 with clean=false
ok 10 should be equal
ok 11 should be equivalent
ok 12 should be equal
QoS 2 [clean=false] ack PubRel { cmd: 'pubrel', messageId: 13867 }

1..12
# tests 12
# pass  12

# ok

The code

'use strict'

var Packet = require('aedes-packet')
var Buffer = require('safe-buffer').Buffer
var test = require('tape').test
var helper = require('./helper')
var aedes = require('../')
var setup = helper.setup
var connect = helper.connect
var subscribe = helper.subscribe

function publish (t, s, packet, done) {
  var msgId = packet.messageId

  s.inStream.write(packet)

  s.outStream.once('data', function (packet) {
    s.inStream.write({
      cmd: 'pubrel',
      messageId: msgId
    })
    s.outStream.once('data', function (packet) {
      if (done) {
        done()
      }
    })
  })
}

function receive (t, subscriber, expected, done) {
  subscriber.outStream.once('data', function (packet) {
    var msgId = packet.messageId

    subscriber.inStream.write({
      cmd: 'pubrec',
      messageId: msgId
    })

    subscriber.outStream.once('data', function (packet) {
      subscriber.inStream.write({
        cmd: 'pubcomp',
        messageId: msgId
      })
      done()
    })
  })
}
test('subscribe & publish QoS 1 with clean=true', function (t) {
  var broker = aedes()
  var publisher = connect(setup(broker), { clean: true })
  var subscriber = connect(setup(broker), { clean: true })
  var toPublish = {
    cmd: 'publish',
    topic: 'hello',
    payload: Buffer.from('world'),
    qos: 1,
    messageId: 42,
    dup: false,
    length: 14,
    retain: false
  }

  broker.on('ack', function (packet, client) {
    console.log('QoS 1 [clean=true] ack', packet)
  })

  subscribe(t, subscriber, 'hello', 1, function () {
    subscriber.outStream.once('data', function (packet) {
      subscriber.inStream.write({
        cmd: 'puback',
        messageId: packet.messageId
      })
      t.end()
    })

    publisher.inStream.write(toPublish)
  })
})

test('subscribe & publish QoS 2 with clean=true', function (t) {
  var broker = aedes()
  var publisher = connect(setup(broker), { clean: true })
  var subscriber = connect(setup(broker), { clean: true })
  var toPublish = {
    cmd: 'publish',
    topic: 'hello',
    payload: Buffer.from('world'),
    qos: 2,
    messageId: 42,
    dup: false,
    length: 14,
    retain: false
  }

  broker.on('ack', function (packet, client) {
    console.log('QoS 2 [clean=true] ack', packet)
  })

  subscribe(t, subscriber, 'hello', 2, function () {
    publish(t, publisher, toPublish)

    receive(t, subscriber, toPublish, t.end.bind(t))
  })
})

test('subscribe & publish QoS 1 with clean=false', function (t) {
  var broker = aedes()
  var publisher = connect(setup(broker), { clean: false })
  var subscriber = connect(setup(broker), { clean: false })
  var toPublish = {
    cmd: 'publish',
    topic: 'hello',
    payload: Buffer.from('world'),
    qos: 1,
    messageId: 42,
    dup: false,
    length: 14,
    retain: false
  }

  broker.on('ack', function (packet, client) {
    console.log('QoS 1 [clean=false] ack', packet)
  })

  subscribe(t, subscriber, 'hello', 1, function () {
    subscriber.outStream.once('data', function (packet) {
      subscriber.inStream.write({
        cmd: 'puback',
        messageId: packet.messageId
      })
      t.end()
    })

    publisher.inStream.write(toPublish)
  })
})

test('subscribe & publish QoS 2 with clean=false', function (t) {
  var broker = aedes()
  var publisher = connect(setup(broker), { clean: false })
  var subscriber = connect(setup(broker), { clean: false })
  var toPublish = {
    cmd: 'publish',
    topic: 'hello',
    payload: Buffer.from('world'),
    qos: 2,
    messageId: 42,
    dup: false,
    length: 14,
    retain: false
  }

  broker.on('ack', function (packet, client) {
    console.log('QoS 2 [clean=false] ack', packet)
  })

  subscribe(t, subscriber, 'hello', 2, function () {
    publish(t, publisher, toPublish)

    receive(t, subscriber, toPublish, t.end.bind(t))
  })
})

[mcollina]

The ack event includes the last packet sent with the given message id. It can be PUBLISH or PUBREL. Given how QoS2 is implemented.. I don’t think it’s feasible to change this.

We should probably document all of this better. Would you like to send a PR?

[gnought]

PUBREL is a control packet, I think it is more logical to be a PUBLISH packet in ack parameter from user point of view.
It is feasible to change the logic, but it requires more code changes and some fix should be applied.

[mcollina]

I agree. However that is tied with how the persistence interface is defined: we don’t have the PUBLISH packet anymore.
As a result, we should just document this for now, as the refactoring might take some time to settle.