I will do my best to post the solution that worked for me; I cannot post the entire flow due to security, and the client is well-known. I have extracted the parts needed to manage the queue.
[{"id":"aa8e7807a8849bbc","type":"inject","z":"47d7ef94526cd759","name":"trigger","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"control","payload":"trigger","payloadType":"str","x":570,"y":340,"wires":[["3e9c374eb33ba559"]]},{"id":"7aa62bb8ab0ac80f","type":"inject","z":"47d7ef94526cd759","name":"reset","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"control","payload":"reset","payloadType":"str","x":570,"y":380,"wires":[["3e9c374eb33ba559"]]},{"id":"091bf18354a04712","type":"inject","z":"47d7ef94526cd759","name":"status","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"900","crontab":"","once":true,"onceDelay":"30","topic":"control","payload":"status","payloadType":"str","x":570,"y":420,"wires":[["3e9c374eb33ba559"]]},{"id":"711e68676bb0a856","type":"inject","z":"47d7ef94526cd759","name":"flush","props":[{"p":"payload","v":"flush","vt":"str"},{"p":"topic","v":"control","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"control","payload":"flush","payloadType":"str","x":570,"y":460,"wires":[["3e9c374eb33ba559"]]},{"id":"3e9c374eb33ba559","type":"link out","z":"47d7ef94526cd759","name":"link out 237","mode":"link","links":["31496907c99ba1c8"],"x":685,"y":400,"wires":[]},{"id":"22743809eddd8960","type":"comment","z":"47d7ef94526cd759","name":"Trigger Queue Keeps process moving","info":"","x":910,"y":360,"wires":[]},{"id":"4a59eb448e088159","type":"link in","z":"47d7ef94526cd759","name":"link in 166","links":["d8d12b83fb280526","6f46639094c98f98","41529d04f1ed4773","aa11340f9759f539"],"x":785,"y":400,"wires":[["1d7e05980213ccb1"]]},{"id":"1d7e05980213ccb1","type":"change","z":"47d7ef94526cd759","name":"trigger Queue","rules":[{"t":"set","p":"payload","pt":"msg","to":"trigger","tot":"str"},{"t":"set","p":"topic","pt":"msg","to":"control","tot":"str"}],"action":"","property":"","from":"","to":"","reg":false,"x":900,"y":400,"wires":[["7fa1252bed51b08a"]]},{"id":"7fa1252bed51b08a","type":"link out","z":"47d7ef94526cd759","name":"link out 238","mode":"link","links":["31496907c99ba1c8"],"x":1015,"y":400,"wires":[]},{"id":"d8d12b83fb280526","type":"link out","z":"47d7ef94526cd759","name":"link out 239","mode":"link","links":["4a59eb448e088159"],"x":1035,"y":500,"wires":[]},{"id":"cb55ac05d0879011","type":"q-gate","z":"47d7ef94526cd759","name":"","controlTopic":"control","defaultState":"queueing","openCmd":"open","closeCmd":"close","toggleCmd":"toggle","queueCmd":"queue","defaultCmd":"default","triggerCmd":"trigger","flushCmd":"flush","resetCmd":"reset","peekCmd":"peek","dropCmd":"drop","statusCmd":"status","maxQueueLength":"5000","keepNewest":false,"qToggle":false,"persist":false,"storeName":"default","x":1070,"y":540,"wires":[["34d01a1d10a8a5f1"]]},{"id":"31496907c99ba1c8","type":"link in","z":"47d7ef94526cd759","name":"link in 167","links":["3e9c374eb33ba559","7fa1252bed51b08a"],"x":965,"y":580,"wires":[["cb55ac05d0879011"]]},{"id":"156621175270c8ae","type":"function","z":"47d7ef94526cd759","name":"Reset Queue Refresh","func":"const index = msg.parts.index\n\n/** \n * Queue should be above zero, allow more request\n * when queue get back to zero\n*/\nif (index == 2) flow.set(\"queueRefresh\",false);\n\n/**\n * We have 60 items in the queue, I want to realease\n * (Trigger) 40 concurrent processes\n */\nif (index == 60) {\n for(let i = 0; i < 40; i++){\n node.send([msg,null])\n } \n}\nreturn [null,msg];","outputs":2,"timeout":0,"noerr":0,"initialize":"","finalize":"","libs":[],"x":880,"y":540,"wires":[["d8d12b83fb280526"],["cb55ac05d0879011"]]},{"id":"34d01a1d10a8a5f1","type":"delay","z":"47d7ef94526cd759","name":"","pauseType":"rate","timeout":"1","timeoutUnits":"seconds","rate":"4","nbRateUnits":"1","rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"allowrate":false,"outputs":1,"x":1230,"y":540,"wires":[[]]},{"id":"a4ee8f5cdcfe323e","type":"comment","z":"47d7ef94526cd759","name":"Load records from database","info":"","x":620,"y":540,"wires":[]},{"id":"cd69fd0b9905248c","type":"status","z":"47d7ef94526cd759","name":"","scope":["bd997fbca132985f"],"x":320,"y":660,"wires":[["21ef9bd4745cd976"]]},{"id":"21ef9bd4745cd976","type":"switch","z":"47d7ef94526cd759","name":"","property":"status.text","propertyType":"msg","rules":[{"t":"regex","v":"queuing|open","vt":"str","case":true}],"checkall":"true","repair":false,"outputs":1,"x":450,"y":660,"wires":[["e38502019ed09a13"]]},{"id":"e38502019ed09a13","type":"function","z":"47d7ef94526cd759","name":"Check if queue is Zero","func":"const queueCount = parseInt(msg.status.text.split(':')[1].trim());\n/**\n * \n * Status node watches queue changes\n * If queue gets to zero request more assets\n * \n * Because the queue will be zero while the\n * assets are loading, set flow context\n * that a refersh as been request and we\n * don't keep calling for assets\n */\nif (queueCount == 0 ) {\n //find out if we have already call a refresh\n let queueRefresh = flow.get(\"queueRefresh\") || false;\n //Refresh already in progress just exit\n if (queueRefresh) return;\n //no refresh called yet, set flow that we are calling\n //a refresh now\n flow.set(\"queueRefresh\",true);\n node.warn(\"queue refresh called\");\n return msg;\n\n}\n\nreturn;","outputs":1,"timeout":0,"noerr":0,"initialize":"","finalize":"","libs":[],"x":620,"y":660,"wires":[["ad5970a9f00ca977"]]},{"id":"ad5970a9f00ca977","type":"link out","z":"47d7ef94526cd759","name":"link out 240","mode":"link","links":["23791724d62c0bce"],"x":755,"y":660,"wires":[]},{"id":"2c5f7fca7bd14f16","type":"comment","z":"47d7ef94526cd759","name":"Check Queue Status and if zero, request more assets","info":"","x":480,"y":620,"wires":[]},{"id":"6bb7e5db011e0dcd","type":"comment","z":"47d7ef94526cd759","name":"Process Flow","info":"","x":1410,"y":540,"wires":[]},{"id":"aa11340f9759f539","type":"link out","z":"47d7ef94526cd759","name":"Trigger item from queue","mode":"link","links":["9975aef084889f75","4a59eb448e088159"],"x":1565,"y":540,"wires":[]},{"id":"f9bd7a999fb4af0b","type":"comment","z":"47d7ef94526cd759","name":"Trigger item from queue","info":"","x":1690,"y":540,"wires":[]}]
The queue gate mode is set to queue; incoming messages are, well.. queued.
The key is in the status routine, which fires when the gate module changes.
If the queue is zero, it will request more records from the database. We use a flow context to ensure we only request the records once per zero; we request 5,000 records each time.
When a new set of records is received, the queue is populated, and we look at the parts index (I know they can be out of sequence, but it's close enough).
Once there are at least two items in the queue, reset the queueRefresh context so we can refresh it when it reaches zero again.
When we have 60 items in the queue, trigger the queue, releasing an item 40 times, I want 40 concurrent processes running.
const index = msg.parts.index
/**
* Queue should be above zero, allow more request
* when queue get back to zero
*/
if (index == 2) flow.set("queueRefresh",false);
/**
* We have 60 items in the queue, I want to realease
* (Trigger) 40 concurrent processes
*/
if (index == 60) {
for(let i = 0; i < 40; i++){
node.send([msg,null])
}
}
return [null,msg];
At the end of the flow, when an item has been processed, trigger the queue again.
These are just some queue checks and balances. The key one is status, which is called every 5 minutes. This fires the status and checks if we have zero items in the queue.
Trigger: Manually releases an item from the queue.
Reset: Deletes all items in the queue.
Flush: Releases all the items in the queue, hence the delay, as extra protection.
I'm not concerned with the queue's volatile state, as the records are stored in a database and marked complete when done.
