Mqtt.js: handleMessage support for QoS 2

[mcollina]

as titled.
Discovered on a chat with @pzfreo.

I've also noticed there are no unit tests for this feature.

Anybody wants to fix this? It should be an easy fix.

[BrandonSmith]

@mcollina Reading through code and issues, I believe lack of handleMessage in QoS 2 is the cause of a undesired behavior in my application. Namely, I need to execute logic for each message sent from the broker serially before the next message is acknowledged. For instance, one of my message types is to reset the application with new configuration information, and this includes disconnecting from MQTT. By the time that message is processed by application code, several other messages have also been "delivered" from an MQTT perspective, but not processed by the application. Thus, there are "missed" when the MQTT connection is re-established.

I am evaluating whether I have enough context to contribute this handleMessag feature for QoS 2. However, I'm trying to read through the spec to understand which stage of QoS 2 processing is it appropriate to execute a handleMessage callback. It appears the callback should execute when the PUBLISH is received.

Another question, it seems from code that the incomingStore never deletes any entries? Is this intentional?

[mcollina]

@BrandonSmith yes I agree with you.

Regarding IncomingStore - it depends if they are acked or not. Maybe there is a bug in our APIs when you are using them in the above way. I have never worked in such a case.

Would you like to assemble a PR?

[mcollina]

Fixed in #697

[oprince]

@BrandonSmith (or anybody else) do you have a client code sample of how to use handleMessage method with QoS 2 ?
I have googled long time with no results...
Thanks

[BrandonSmith]

@oprince Override the handleMessage function with your own. The important thing is that you must always invoke the callback argument. QoS 2 support landed in v2.13.0 with improvements in v2.14.0.

const client = mqtt.connect()
client.handleMessage = (packet, callback) => {
  somePromiseReturningFn(packet) // do something asynchronous with packet
    .then(() => callback())
    .finally(() => callback())
}

const client = mqtt.connect()
client.handleMessage = (packet, callback) => {
  someNodeStyleCallback(packet, (err, result) => {
    if (err) {
      callback()
    } else {
      callback()
    }
  })
}

[ChristiaanWillemsen]

I still see some problems here. Imagine the following scenario: You have multiple subscribers on a topic using shared subscriptions (so load balancing). Now imagine that the handling of a single message takes multiple minutes, but because of the amount of subscribers, you are able to effectively handle the traffic.

In this case, the library will fail: after some timeout because you don't call the callback in time. It says you must call the callback, but it does not say when.

Would be good to have some kind of control over this.

Other scenario: What if you want to be able to handle multiple inflight messages at ones? Would you then start multiple clients for this?

[mcollina]

MQTT Acks are to indicate message reception, not handling of the message itself. It's a transfer of ownership. If you would to improve with multiple in-flight message support during reception, feel free to send a PR!

[mcmullinboy15]

@oprince Override the handleMessage function with your own. The important thing is that you must always invoke the callback argument. QoS 2 support landed in v2.13.0 with improvements in v2.14.0.

const client = mqtt.connect()
client.handleMessage = (packet, callback) => {
  somePromiseReturningFn(packet) // do something asynchronous with packet
    .then(() => callback())
    .finally(() => callback())
}

const client = mqtt.connect()
client.handleMessage = (packet, callback) => {
  someNodeStyleCallback(packet, (err, result) => {
    if (err) {
      callback()
    } else {
      callback()
    }
  })
}

@BrandonSmith How do I get the message out of the packet?