6

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 generates SELECT query.
  • Wire to PostgreSQL node.
  • Parse output, use flow.get and generate INSERT 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. ?

  • Are you using the same key to store the value in the flow context? – hardillb Jun 24 '18 at 10:09
  • @hardillb, yes. – cogitoergosum Jun 24 '18 at 10:19
2

It's worth remembering that everything in the NodeJS world is asynchronous which means nothing blocks the event loop.

In this case it sounds like you get the first incoming message, which you store with a fixed key in the flow context, you then move on to the Select query. At this point the SQL node is going to end up doing some network IO which will block while it waits for a response from the database so while it waits it will give up the execution context.

If while it's waiting a new MQTT messages arrives it will be handled immediately and passed to the first function node which is going to overwrite what ever was stored in the flows context because the key is the same.

When the first Select statement returns the first message will move on to the second function node and when it retrieves the value from the flow context it will be the second message not the first.

The way to solve this is to not use the context to keep state but to move the information you want to keep from msg.payload to a different key on the msg object. Well behaved Node-RED nodes should always pass on the original msg object and by default only really change the msg.payload (there are exceptions, but they tend to document what they change and why).

Using the msg to hold state for a unit of work guarantees that it can only be changed in step with the message travelling through a flow.

|improve this answer|||||
  • Hmm...I thought the flows in Node Red abstracted the synchronisation. Will try the msg. option. – cogitoergosum Jun 24 '18 at 13:46
  • The msg. object store worked! – cogitoergosum Jun 26 '18 at 14:14
  • On your third para of response, when a new MQTT message comes, a new msg. object is created for that particular MQTT message, right? – cogitoergosum Jul 25 '18 at 9:11
  • 1
    Yes, each new incoming MQTT message gets its own new msg – hardillb Jul 25 '18 at 9:23
  • Suppose a message is expected to traverse through certain set of nodes in a deployed flow. Suppose the next message is also expected to traverse the same set of nodes. Then is it possible that a node might 'see' msg different from what was received at the input nose? – cogitoergosum Jul 25 '18 at 11:41

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.