Get all retained messages not sent while subscriber is offline, per device

Abdozy

Hey, greetings,

Wondering, how to fetch retained messages per device which has not been sent to any subscribers ?
For example, a client subscriber is getting data from device topic, then subscriber disconnected from topic (goes offline), I need to get only the new recent messages (which has not been sent yet) from device once subscriber returns online.
How to reach this behavior using MQTT API ?

thanks

Abdozy

I checked mqtt api flespi/state/gw/devices/{device_id}/telemetry
it only get the the last recent message,
I need to get the whole unsent messages to a subscriber (e.g. as a patch chunked), using MQTT or REST API !

shal

Abdozy have you tried to use persistent MQTT sessions (clean=false and expire_interval set on connect)?

Abdozy

shal Thanks for your reply, yes I just tested with persisted mqtt session, and it works.

One more inquiry please, if I need to delete the delivered retained messages to subscriber, should I set the messages ttl to zero in Devices ?

Thanks

shal

Abdozy retained messages are different. Do not mix it with offline persistent sessions - no correlation between them.
To delete retained message from the topic publish message without payload.

I suggest you to read MQTT essentials guide by Hive MQ (or any other) to learn the basics and understand these numerous nuances about MQTT protocol especially it's powerful 5.0 version.

Abdozy

shal Really excellent resources.

I have an inquiry regaridng the following scenario related to persistent messages:

If the number of offline messages exceeds certain no of messages (e.g. 100)

Does Flespi support (or could be configured) to send offline messages in batches ?, i.e. each batch contains # messages (e.g. 50), then sending theses batches as intervals or immediately, once subscriber returns online

Imagine, if we have a network break, a lot of persisted offline messages needs to be in hold.
These messages would have priority over new incoming messages meaning there would be even more of a delay before the application was receiving up to date data, and not historical data.

Thanks

shal

And of course read all information on this page: https://flespi.com/mqtt-broker there are some special features of flespi MQTT Broker that can save you a lot of time and which are not available in any other broker.

Abdozy

shal Thank you shal for your answer, I will check these resources.

shal

Abdozy just google for "mqtt 5.0 inflight window" option on mqtt connect. 5.0 protocol should be used by the client for this.

Abdozy

shal
I checked the mqtt 5 protocol for in-flight
https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html
If both Client and Server set Receive Maximum to 1, they make sure that no more than one message is “in-flight” at any one time. In this case no QoS 1 message will be received after any later one even on re-connection. For example a subscriber might receive them in the order 1,2,3,3,4 but not 1,2,3,2,3,4. Refer to section 4.9 Flow Control for details of how the Receive Maximum is used.
I think it not related to "message batching technique",

I believe that its implementation is introduced by cloud vendor.
e.g. Google and Alibaba pub/sub services, introduce it as a feature to achieve a higher throughput of messages.

shal

Abdozy with inflight message count option on connect you control how many unacked messages broker will send to you. Set it to 50 and you will receive a batch of 50 messages. Once they are processed - send ack for each separate message to receive more new messages or for all 50 at once to request next "batch".

Abdozy

shal
Unfortunately, the MQTT.JS lib which support v5.0 , does not support max inflight message count option, I will try to find another js lib.

Thanks Shal I appreciate your support.

Abdozy

Update: I am not sure is it the "receiveMaximum" : representing the Receive Maximum value number, when tested still receiving offline messages one-by-one

shal

Abdozy this option is applied to QoS=1 publications and subscriptions only, just in case.

As far as I remember mqtt.js automatically acknowledges packets for you once processed by the handler: https://github.com/mqttjs/MQTT.js/blob/main/src/lib/handlers/publish.ts#L157
May be customHandleAcks callback may adjust this behavior.

In any way there is no need for artificial break of messages delivery and current mqttjs implementation will wait until you process the message in your handler and only after that will send "puback" confirmation to the broker.

Abdozy

shal Got it, my subscriber is on qos 1, but I am still not getting how to configure mqtt.js to set inflight message max count, to receive the offline messages in batches instead of one-by-one.

I subscribed to "flespi/state/mqtt/sessions/{id}/inflight_msgs_size" to check the value
and got zero.

Knowing that I am using mqtt.js as a client subscribing to "flespi/message/gw/devices/{device_id}" with client: false, qos:1 and sessionExpiryInterval: value, and the offline messages retrieved one-by-one.

Abdozy

Update: I've inquired about how to handle this option using mqtt.js github repo, got a reply that offline messages batching needs to be implemented from broker side, that is why I saw such a feature in other vendors like e.g. Google cloud, Alibaba ApsaraMQ.
I did try with Paho mqtt client in python setting inflight messages count, but got nothing.

shal

Abdozy maximum inflight messages is the single mechanism that backs "batching" on the broker side. This should be implemented by any MQTT Broker that fully conforms to 5.0 MQTT protocol standard. You may read here Flow Control. Other providers can label this feature as "batch" but it is backed by this single option. Which flespi MQTT Broker fully supports.

Abdozy

@shal So, from the client side, we just need to set this param so as to test the batching mechanism, I checked the python paho client lib, setting the inflight max count, but with no success to get it works. it might be something else needs to be configured in broker side, I am not sure.