MQTT5 retained message

General
1

Hi, my question almost close to these topics
Understanding MQTT v5 retain options and Ignore retained MQTT messages at restart
but I'm still not able to get what I need without awful workarounds.
My problem is this:

  • MQTT5
  • MQTT server, publisher and subscriber on 3 different machines (debians)
  • messages QoS 2 and retained flag.
    I need to get and use this messages just once, not more. The problem is that doesn't matter the configuration I use regarding clean start and retained flag handled, at reboot I always get the last message with the retained flag set as true. I cannot simply ignore the first message because the subscriber could be off while the publisher send it, and in this case I have to process the message. I know I could use the timestamp to detect the last message, but I would like to use just the MQTT option, avoiding to add the timestamp to each single message
2

If you don't want the retained message, then make the topic not retained. Then you will only get a message at the time it is published.

3

No, I need the rained messge. As you see, otherwise if the subscriber is off, it wan't get the message pubblished in that time

4

I see what you mean now. I can't think of any option other than adding a timestamp (or other unique id) to the message, so you can ignore duplicates at the receiver end.

5

Yes, and save it in persistent context. By the way at the moment (since this saving must be immediatly in order to avoid to process two times the same message) I use this configuration in the settings,js

file: {
      module: 'localfilesystem',
      config: {
            cache: false
      },
}

and this in a function node:
flow.set("mqttID", msg.payload.timestamp), "file", function(err){if(err){node.error(err, msg)}});

Is that correct or there is a simpler way?

6

This sounds like you are hoping the broker will queue messages up until your subscriber comes back online.

My understanding is that it only retains the last message received on any topic, so if your subscriber is down it can still miss messages regardless of the Retain flag from the publisher.

8

Yes, i't ok missing the intermediate messages in this case

9

Personally I would use node context in the function rather than flow. Always keep the scope of data as local as possible.

Bear in mind, though, that this method is not infallible. Node red writes the context out to file at a configurable interval, or on shutdown. That means that a normal shutdown of node-red will guarantee that the most recent value is saved in the file, however if node-red crashes or there is a power fail then the latest value may not yet have been written to disc.

As a matter of interest, what operation is being performed at the receiving end that must not be repeated?

10

Also, obviously, before setting it in file storage you need to get it and compare it with the one in the current message, and discard the current message if they are the same.

11

Yesm you're rigth

I thought the configuration I posted, would have written immediately on file. It's not so?

it depends on the messages, but for example it could even reboot the system.

Oh yes, sure, I was more concerned about passing the idea than the all logic. You're right

12

No, as in the docs:
" it caches the values in memory and only writes them out to the file-system every 30 seconds."

That is to prevent it spending all its time saving stuff to disc if a context var is updated rapidly. Imagine what would happen if the context were set 1000 times a second if it wrote the file every time.

The time is configurable in settings.js.

13

Looking here Local Filesystem Context Store : Node-RED I understand that flushInterval works only if cache is set to true. Setting it to false, it writes directly on disk.
If it's not like this, how can I set it to write the changed context after 1 seconds (I means as soon as possible) on disk?

14

Oh you are right, I did not know you could do that. Be careful how often you write to any context vars using that store.

That is on the next line under the cache definition you pointed to :slight_smile:

15

Thanks. So just to be sure I'll recap it:

  • this writes immediatly on disk
file: {
      module: 'localfilesystem',
      config: {
            cache: false
      },
}

and in a function node:
context.set("mqttID", msg.payload.timestamp), "file", function(err){if(err){node.error(err, msg)}});

  • While this one writes on disk each second
file: {
      module: 'localfilesystem',
      config: {
            cache: true,
            flushInterval: '1'
      },
}

and in a function node, as above,
context.set("mqttID", msg.payload.timestamp), "file", function(err){if(err){node.error(err, msg)}});

16

I had a similar but slightly different requirement. As I already have a redis instance in my environment I used redis to store the message ts - all works well however the standard context storage to disk would be the default soon

17

I didn't know about redis, I'll give it a look even if I think that node-red context is ok for my use, Just I need to understand a bit better how to directly write on disk as soon as possible, since these messages are not so often

18

This might help you: Local Filesystem Context Store : Node-RED

19

Yes, I even posted the same link.
Is this my recap correct? MQTT5 retained message - #15 by Lupin_III

20

I think the flushInterval should to be a number and you have it set to a string:

file: {
      module: 'localfilesystem',
      config: {
            cache: true,
            flushInterval: '1'
      },
}

Try changing it to

file: {
      module: 'localfilesystem',
      config: {
            cache: true,
            flushInterval: 1
      },
}
21

Well you tell us. It is easy enough to check by injecting into a change node setting a global var while watching .node-red/context/global.