Queueing input messages

Developing Nodes
1

In a node I am creating I want to service messages sequentially rather than in parallel (it runs async tasks). Is there a way I can tell node to hold the messages for me (so on('input') will not get called till I re-enable it) or do I need to queue them internally?

2

You will need to queue them internally.

If your node's message handling is entirely synchronous then it's not a problem.

But it is an interesting idea for a node to be able to tell the runtime it can only be invoked one message at a time - relying on the node.done function to trigger the release of the next message. Something to consider for the future.

1 Like
3

OK, thanks. I imagine this is not an uncommon requirement.
This is actually a follow on from my previous thread about cleaning up where there are mulitple processes to kill caused by handling messages in parallel. I realised that though the node functions perfectly well in this mode, since there are progress messages passed on from spawned tasks those would be all mixed up and it would be a pain for the user to split them up. There isn't actually any significant benefit to running them in parallel so I have decided to serialise them instead.

4

In case anyone finds this and wants to know a solution, it was easier than I anticipated.

        let msgQueue = [];

        node.on('input', function(msg, send, done) {
          // push this one onto the queue
          msgQueue.push( {thisMsg: msg, thisSend: send, thisDone: done} )
          // if there is now only one in the queue then nothing going on so handle it immediately
          // otherwise already dealing with one so nothing more to do at the moment
          if (msgQueue.length == 1) {
            handleMessage(msgQueue[0].thisMsg, msgQueue[0].thisSend, msgQueue[0].thisDone)
            // leave it in the queue, it will be shifted out when done
          }
        });

Then in handleMessage() after calling done()

              // shift this message off the front of the queue
              msgQueue.shift()
              // see if any more to do
              if (msgQueue.length > 0) {
                // yes, start the next one
                handleMessage(msgQueue[0].thisMsg, msgQueue[0].thisSend, msgQueue[0].thisDone)
              }
5

This topic was automatically closed 60 days after the last reply. New replies are no longer allowed.