I have a MQTT Input node that kicks of database operations (SELECT and INSERT) for PostgreSQL database. The database operations are done with node-contrib-postgres-multi.Since these operations are separated by function nodes, I am saving portion of the message with flow.set
and retrieving with flow.get
later. For example,
- Function node saves with
flow.set
and generatesSELECT
query. - Wire to PostgreSQL node.
- Parse output, use
flow.get
and generateINSERT
statement. - Wire to PostgreSQL node.
I can't help but imagine that the flow.set
and flow.get
are not in sync.
Currently, I am simulating around 20 devices to publish data every second with timestamp increased by 1 second for every publication. There is absolutely no reason why the generated messages should ever get duplicated. However, the database insert nodes fail because of unique index violation as seen in Node-RED log (.pm2/logs/red-out-0.log
).
If the function nodes and database processing take say, 2 seconds and the MQTT messages (QoS=0) are received every second, would MQTT or Node-RED buffer them? Therefore, every received message is treated as a Unit of Work until it errors out or 'leaves' the flow into a database, HTTP request, MQTT publish, etc. ?