Add support QOS 2

[ghost]

Any plans to support QOS level 2? I know there are probably some difficulties but it would be sweeeeet for our purposes. We have a certain messages that we really only want to send once and need to make sure that the device gets it.

Currently we have to implement a sort of handshake (where the devices responds with a confirmation message) to achieve this.

[mcollina]

The main issue with QoS 2 is that it cannot be implemented across all
supported brokers, so Mosca must detect if that is supported or not. As an
example, Redis and MongoDB does not support it.

Plus, I currently do not have time for this :/ and I think I should focus
on consolidating Mosca's current features.

Are you using any other backend/broker to scale out Mosca?

What are you doing that requires exactly once delivery?

[ghost]

I totally understand (not why Redis and MongoDB don't support it but the part about not having time ;)

Our current setup is Mosca with MongoDB for persistence. We have a device (running node!) connected with the server thats sending sensor data and has some peripheral outputs (can turn watering pump on/off for crop irrigation).

If sensor data hits a given threshold, we want to turn on the water pump for a given amount of time (passed in the message).

I guess when I think about it, QOS 1 might be enough and then I just need to code against additional messages so funky things don't happen if it gets received multiple times.

Any comments on this approach? How likely is it that QOS 1 will send the message more than once (how quickly is it repeated(configurable?) and what causes it to stop). I've read the MQTT spec but the QOS stuff can make my head spin.

[AaronM04]

Why not use QoS 1 and send the times to turn on and off the pump in your messages? This would require good time synchronization (e.g., using NTP). This way, it would not matter if the message arrives more than once. Would that fix your problem?

[ghost]

@AaronM04 Thats a solid idea but doesn't exactly fit into the use cases (its probably more confusing than I should get into). Either way, this has help me think though how I can embrace QOS1. My only concern would be if duplicates continue to show up well after the message was received.

If someone could give me a little more specifics on how QOS1 is implemented on mosca it might help.

I've read through this but i guess its still a bit ambiguous, just as far as specific times goes.
http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#qos-flows

Is the message pretty much repeatedly retried until a PUBACK is received? Is there a way to configure the timespan between each retry?

Also does the QOS refer to the connection between the publisher and broker, broker and subscribers or between the publisher and subscribers?

I know thats alot of stuff at once so whatever you guys can help clarify would be appreciated.

[mcollina]

The QoS flows have been clarified in MQTT 3.1.1:

The receiver of a QoS 1 PUBLISH Packet acknowledges receipt with a PUBACK Packet. If the Client reconnects and the Session is resumed, the sender MUST resend any in-flight QoS 1 messages setting their Dup flags to 1

So by the spec there is no required retransmission. This is not true in Mosca, but it's something I would like to fix/see fixed (if anyone wants to give it a try #116). However, it redeliver messages with the dup flag set to 1 (https://github.com/mcollina/mosca/blob/master/lib/client.js#L178). Please verify everything and in case open issues/pull requests.

What's the practical difference between QoS 1 and QoS 2?
The difference is that the publisher knows if a message have been received.

Why MongoDB and Redis does not support QoS 2

Because if you publish a messages to MongoDB or Redis you:

1. do not know how many subscribers there are
2. do not know if those subscribers have received and processed that message

At least once semantics in MQTT

In MQTT, you have at-most-once semantics, but you can deduplicate messages on client side by checking the dup flag and the message id. I'm not sure how this is handled in MQTT.js, but it's possible to add a de-duplication layer there.

[mcollina]

@andypiper what's your opinion on this matter?

[knolleary]

@mcollina - I disagree with your statement about the difference between qos1/2

The difference is that the publisher knows if a message have been received.

How do you come to that conclusion? In both QoS 1 and QoS 2 the first ACK (qos1:puback, qos2:pubrec) is sufficient to inform the publisher the message has been received.

The difference is that the receiver is not able to detect duplicates with QoS 1, whereas it can with QoS 2 and ensure they don't get forwarded on.

[ghost]

@knolleary

The difference is that the receiver is not able to detect duplicates with QoS 1, whereas it can with QoS 2 and ensure they don't get forwarded on.

I'm assuming when you say "reciever" you mean the broker

@mcollina
I'm looking at the Client "actualSend" function to try and understand how QoS1 is handled. There are a couple things that stand out...

https://github.com/mcollina/mosca/blob/master/lib/client.js#L162
If the client is disconnected the message stops trying to publish? Isn't there where we want to keep retrying? Or is this referring only to clean disconnects (where the client requests it) and not network drops/issues.

https://github.com/mcollina/mosca/blob/master/lib/client.js#L164-L182

From the rest of the code, with QoS1, it seems like the message will be sent 10 times regardless of a ACK? Essentially with this implementation QoS1 will send 10 messages, with an exponential timeout between each subsequent message? Or am I missing something?

[knolleary]

@illbilly As the qos flows are symmetrical, they can be initiated by either the client and broker. In this instance, as we're discussing a client acting as the publisher, then the receiver is, indeed, the broker.

We tightened up the language in 3.1.1 to use 'sender' and 'receiver' when talking about the qos flows.

[ghost]

@knolleary ahhh i see. I think that's been the biggest cause of my confusion is that the client/server paradigm is symmetrical and I need to consider things from both the client and subscribers point of views.

@mcollina
You can ignore the second part of my comment above. I see that you're destroying the timeout on "pubback".
https://github.com/mcollina/mosca/blob/master/lib/client.js#L338-L348

However, when you retry message, it seems you're overwriting the old timeout object but not clearing it?
https://github.com/mcollina/mosca/blob/master/lib/client.js#L186
Won't this result in the old timer continuing to run, and only the one that receives a PUBBACK will get cleared? Maybe something like this around line 184

 clearTimeout(this.inflight[packet.messageId].timer);
 this.inflight[packet.messageId] = {
        packet: packet,
        timer: timer
      };

Once again, I'm not a js expert so maybe I'm missing something?

[ghost]

Ahhhh nvm I see. The timeout has already fired...

[mcollina]

@mcollina https://github.com/mcollina - I disagree with your statement
about the difference between qos1/2

The difference is that the publisher knows if a message have been received.

How do you come to that conclusion? In both QoS 1 and QoS 2 the first ACK
(qos1:puback, qos2:pubrec) is sufficient to inform the publisher the
message has been received.

I re-read the spec, and you are right. I am not sure where I got this bad
interpretation. The newer spec is really much clearer. So, this is
implementable in Mosca, it's just a matter of the work to be done to add
the storage.

It seems I have plenty of stuff to be done in here :). Anyone willing to
help? :D

The difference is that the receiver is not able to detect duplicates with
QoS 1, whereas it can with QoS 2 and ensure they don't get forwarded on.

Why not? The messages comes with their own message identifiers across
redeliveries, plus the dup flag. So a subscriber might know if a message
has already been seen or not. Can you explain it further? I think there
might be issues when the client crashes and so on but surely you have much
more to say!

[andypiper]

@mcollina seems like you have the right folks helping out in this thread, I'm swamped ATM.

[knolleary]

For QoS1, there are two scenarios when a PUBLISH will be resent:

1. the receiver received the PUBLISH, processed it, but the PUBACK didn't make it back to the sender. As far as the receiver is concerned, this PUBLISH has been dealt with and it can forget about it.
2. the receiver never receives the PUBLISH.

In both cases, when the receiver gets the resent PUBLISH (with dup set), it has no record of having ever processed the message ID; in case 1 it has forgotten about it, in case 2 it never knew about it. This means all it can do is send on the message to the subscribers.

The reality of the situation is the dup flag is fairly meaningless in the protocol. It gives a hint, but that is all - you shouldn't rely on it as an indication you've definitely already processed the message.

In QoS 2, the second stage of the flow (PUBREL/PUMPCOMP) is there for both sides to agree that the message has been received and can be processed. This prevent duplication.

[mcollina]

thanks a million, it was really helpful.

So, QoS 2 is on the todo list. I'll ask you to revise the branch when it's
up.

[jteplitz]

Any update on this?

[mcollina]

It's done when it's done :).

I very little bandwidth for tackling this at the moment, but if you (or anybody else) wants it we can collaborate, the plan to get there is a little bit long. The main issue is that I did a very bad decision at the time, and unfortunately it requires lots of refactoring and API changes to fix that.

[jteplitz]

I'm down to give it a shot. Can you give me an outline of what needs to be
done?
On Aug 23, 2014 1:54 AM, "Matteo Collina" notifications@github.com wrote:

It's done when it's done :).

I very little bandwidth for tackling this at the moment, but if you (or
anybody else) wants it we can collaborate, the plan to get there is a
little bit long. The main issue is that I did a very bad decision at the
time, and unfortunately it requires lots of refactoring and API changes to
fix that.

—
Reply to this email directly or view it on GitHub
#115 (comment).

[mcollina]

Good!

So, here it is: Mosca is built around the idea that we are flowing PUBLISH messages in & out the persistence, and in & out the broker. However, QoS 2 mandates that you should store in the persistence PUBREL packets too.

When a Client reconnects with CleanSession set to 0, both the Client and Server MUST re-send any unacknowledged PUBLISH Packets (where QoS > 0) and PUBREL Packets using their original Packet Identifiers (http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/csprd02/mqtt-v3.1.1-csprd02.pdf)

Mosca's core delivery method client.forward() is built around the wrong assumption that it has a topic and a payload as arguments. The same olds true for server.publish().

This is getting more complicated by the fact that even Ascoltatori is built around that wrong topic-payload concept. There might be some hacky way around that, maybe look around in the API.

My steps for this are:

• Write some (failing) unit tests for the QoS 2 case.
• Refactor client.forward and server.publish() so that they objects and non-publish packets.
• Decouple the global message identifier (https://github.com/mcollina/mosca/blob/master/lib/server.js#L319) from the per-client message identifier (for which we have the ugly server. updateOfflinePacket().
• Refactor ascoltatori to accept objects instead of a topic-payload pair (if-needed).

[psorowka]

Is there any update on this topic?

I am currently working on a production MQTT server which is based on mosca. QoS 2 is not needed from today on, but I would like to see that working. I am of course also willing to contribute to the core code.

So, @mcollina and @jteplitz602 have you made any advances on QoS 2 since discussing in that thread?

@mcollina did I get it right, that your previous statement

The main issue with QoS 2 is that it cannot be implemented across all
supported brokers, so Mosca must detect if that is supported or not. As an
example, Redis and MongoDB does not support it.

is not true but originated in a misunderstanding of the QoS concept?

[mcollina]

Exactly, more specifically of the guarantees QoS 2 gives.

It is supportable, and I am working on it on my spare time. Unfortunately it requires a complete refactoring of Mosca and Ascoltatori. This refactoring will also gives something between 2x and 4x throughput, which is highly welcomed :). However, this is far from being production ready.

The hard part of the tasks defined above is:

Refactor ascoltatori to accept objects instead of a topic-payload pair (if-needed).

This is not possible, and in fact I am rewriting ascoltatori and splitting it up on multiple modules (http://npm.im/mqemitter, http://npm.im/mqemitter-redis, http://npm.im/mqemitter-mongodb).

[mcollina]

As I have some time, I will open source the current state of the new modules and ping you about it.

[psorowka]

Ok looking forward to that refactor; an increase in throughput is of course always welcome.

On the other hand, I am curious why you are planning to implement the QoS routines in Ascoltatore rather than in Mosca. In my understanding QoS is highly MQTT specific.

I imagine implementing QoS 2 in Mosca cannot be so hard. Let me sketch my thoughts:

• for easiest implementation, I would choose Method B from the QoS2 flow in MQTT 3.1.1 spec
• this would mean that we can pass the message to ascoltatore and let it deliver exactly as before
• after that, we would issue the PUBREC message to the sender
• in client.js we would store a list of inflight package identifiers (which could eventually be persisted in a 2nd step by the persistency classes)
• subsequent publish of a message with an identifier that can be found in this list would be blocked
• something like client.on('pubrel') would purge a confirmed identifier from the list

on the outgoing side (aka forward), I suppose all that would be needed would be a retransmission timeout and a handler for the pubrec message, which would erase the timeout and issue the pubcomp message.

am I missing something?

[mcollina]

The QoS features needs to live outside of Ascoltatori inside the persistence. Ascoltatori is just a simple pub/sub layer, and QoS is specific between the client and the broker, not client <-> client. The current architectures allows only PUBLISH commands to flow through the persistence, while in the new thing also other commands can transit, allowing QoS 2.

[psorowka]

In the current implementation, I see two persistency layers

• one from Ascoltatori, depending on its chosen backend
• one from Mosca, to enable offline package support

while I agree that it could be of interest to streamline this, I would argue that it could be possible to add QoS 2 already in the current structure.

When you say:

in the new thing also other commands can transit, allowing QoS 2.

do you describe the new things as some kind of outgoing queue?

[mcollina]

Ascoltatori has already too many responsibilities and too much code. No more features should get in there. All QoS is specific to MQTT, thus it should stay in Mosca.

It is possible to add QoS 2 in the current structure by slowing Mosca further down. Basically the issue is that doing it maintaining the current performance is quite a task, because most of the API follows this pattern (topic, payload, options), while it should just be (packet). Switching between these two slow things down. There are already a ton of hacks into Mosca (that is why I added an options argument) for this reason.

do you describe the new things as some kind of outgoing queue?

Yes exactly. Maintaining the same data format all along the way. Almost no special logic should be needed for QoS 2.

[ghost]

@mcollina @psorowka have there been any updates on this?

[mcollina]

@bmcustodio this will be included when I port Mosca on top Aedes, which already supports QoS 2. Some months and it will be a breaking change.

I welcome any help that might come!

[jimutt]

What's the status of this issue and Mosca in general? Seems like stuff has been slowing down a bit in general looking at the repo graphs. I really don't mean to criticize anyone at all, I'm just wondering what the current state of Mosca looks like @mcollina as I'm currently looking at a couple of different MQTT broker solutions for a future project.

[mcollina]

I'm preparing a new major release (rewrite), based on
http://github.com/mcollina/aedes. It would be 10x faster and more solid.
The new approach would be more modular, so there won't be many commits
here.There is a some more work to be done, I'm currently working on redis
support. Then mongo, some boilerplate and then I'll release.
Il giorno mar 29 dic 2015 alle 12:22 jimutt notifications@github.com ha
scritto:

What's the status of this issue and Mosca in general? Seems like stuff has
been slowing down a bit in general looking at the repo graphs. I really
don't mean to criticize anyone at all, I'm just wondering what the current
state of Mosca looks like @mcollina https://github.com/mcollina as I'm
currently looking at a couple of different MQTT broker solutions for a
future project.

—
Reply to this email directly or view it on GitHub
#115 (comment).

[behrad]

I'm eager to test in staging when redis support is ready @mcollina , When dya think it would be available? and will this OK to plan for production?

[jimutt]

@mcollina Tanks for answering, sounds great! Looking forward to the new release.

[mcollina]

@behrad here you are: https://gist.github.com/mcollina/a37f135136192d9c58a4

I haven't performance tested this (on my box it ticks at ~10k msg/s, but there are a couple of deopts and lots of anon functions), or put in production anyhow. Plus, it's still missing your delivered event :D.

[behrad]

Great @mcollina

it's still missing your delivered event

Sure, Should I create two PRs one for aedes and one redis implementation at aedes-redis?

[behrad]

Actually delivered will go only into core aedes, I meant for those features that need backend implementation.

[mcollina]

@behrad if you need one more feature for the persistences, https://github.com/mcollina/aedes-persistence and https://github.com/mcollina/aedes-persistence-redis (mongodb coming soon).

I have already added the ability to count offline clients and subscriptions, so you should be good.

(it is still missing all stats stuff)

[dushyantbangal]

Feels rude to ask (I hate when people ask George RR Martin to write books faster), but is QoS 2 support nearby or its gonna take time?

[mcollina]

QoS 2 support is in http://npm.im/aedes.
At some point, I will refactor this to use Aedes, once I am happy that it is stable enough.