Added "topic based fair queue" option to delay node.

[Dave C-J]

HI all

just to let you know I have added what I call a "topic based fair queue" option to the delay node... What this should do is...

When a message arrives on a topic 

    if it's on a new topic it gets added to the back of the queue

    If it's on an existing topic it replaces the existing topic value at it's position in the queue

Every delay time period the message at the front of the queue gets sent

This means that each topic gets a go at being sent (= fair), but it's always the most recent value for that topic that gets sent... (so yes it drops "old" values)

A use case could be for example...  say we have 10 devices reporting values every second - but can only feed back a total of one per second on the next hop to the server, then we need to drop 9 out of 10 messages - but want to make sure everything gets it's latest position reported....

comments and bug reports welcome as usual...

[Franklin Dattein]

Very useful.

I got a bit confused to when the first 'tick' happens. It looks like it is at the end of the period.

I was trying to use it to limit the number of emails sent per day, for a topic. However, it would only send the first email at the end of the day.

It would be nice to have an option to be able to not limit the first message..

Thanks

[Dave C-J]

Hi,

yes currently at each timer expiry (which is independent of any message arrival times) it sends the message for the next topic in the queue. So I'm not sure it would do what you want either way ? - I think you are asking for (say) one message per topic per day... ie the first one goes out but no more on that topic until the next day ?   

Currently if you set it to tick at one day interval you would only get one message in total per day (and you would get one for the next topic the next day etc).

Am I mis-reading your use case ? just to clarify.

[dave....@gmail.com]

HI, I try to use Delay node with topic based fair queue but I don't understand why it is working the way it does ...

I send 5 message topics to the delay node I see under the node the counter 5 at the end of the period (2/msg per minute  like 1 msg each 30 sec) I expected to get 1 msg sent from the 5 queued messages but it sends all 5 at once , am I misunderstanding ?

[Dave C-J]

any chance you can attach the complete flow so we can see the internal settings you have. thanks

[Dave C-J]

The topic based fair queue should only release 1 message at each "tick" - so in your case every 30 seconds. if you "stack up" 5 messages, 1 from each of your topics then I would expect topic 1 at 30 secs , topic 2 at 60 secs etc...  - ie each topic gets released in turn. If the payload on that topic happens to change (new arrival) then the new payload gets released.

The timed release queue releases one of every topic at each tick  - so if you set that to 30 secs (2 per min) then if all 5 topics had arrived then it would release all 5...  again if during that 30 secs other payload values on each topic arrive , then they replace those already in the queue.

[dave....@gmail.com]

The topic based fair queue should only release 1 message at each "tick" - so in your case every 30 seconds. if you "stack up" 5 messages, 1 from each of your topics then I would expect topic 1 at 30 secs , topic 2 at 60 secs etc...  - ie each topic gets released in turn. If the payload on that topic happens to change (new arrival) then the new payload gets released.

That's what I expected but it doesn't appear to be working that way example flow below (although this is not the flow I am using this simple one gives the same result !) all 5 topics fire every 30 seconds 

 

[{"id":"94cdb16c.2e9fe","type":"delay","z":"e324760.b29cc88","name":"","pauseType":"queue","timeout":"2","timeoutUnits":"seconds","rate":"2","rateUnits":"minute","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":true,"x":774.7299652099609,"y":1714.9442176818848,"wires":[["3fbd821e.ff242e"]]},{"id":"3fbd821e.ff242e","type":"debug","z":"e324760.b29cc88","name":"","active":true,"console":"false","complete":"true","x":974.2855262756348,"y":1714.6106777191162,"wires":[]},{"id":"d3f0ab01.b2e1d8","type":"inject","z":"e324760.b29cc88","name":"","topic":"1","payload":"1","payloadType":"str","repeat":"10","crontab":"","once":false,"x":498.88885498046875,"y":1641.1111345291138,"wires":[["94cdb16c.2e9fe"]]},{"id":"591f6402.1fa6bc","type":"inject","z":"e324760.b29cc88","name":"","topic":"2","payload":"2","payloadType":"str","repeat":"10","crontab":"","once":false,"x":496.6666259765625,"y":1681.1111631393433,"wires":[["94cdb16c.2e9fe"]]},{"id":"da1270d7.8e8fa","type":"inject","z":"e324760.b29cc88","name":"","topic":"3","payload":"3","payloadType":"str","repeat":"10","crontab":"","once":false,"x":494.44439697265625,"y":1721.1111249923706,"wires":[["94cdb16c.2e9fe"]]},{"id":"c703b493.d7bb38","type":"inject","z":"e324760.b29cc88","name":"","topic":"4","payload":"4","payloadType":"str","repeat":"5","crontab":"","once":false,"x":495.5555191040039,"y":1757.7776708602905,"wires":[["94cdb16c.2e9fe"]]},{"id":"f8185df9.f1cae","type":"inject","z":"e324760.b29cc88","name":"","topic":"5","payload":"5","payloadType":"str","repeat":"5","crontab":"","once":false,"x":495.5555114746094,"y":1796.666666984558,"wires":[["94cdb16c.2e9fe"]]}]

[dave....@gmail.com]

I did a bit more test ... the result if I change from "topic based fair queue" to "timed release" is identical,  but I can confirm in both cases If the payload on that topic happens to change (new arrival) then the new payload gets released. 

[Dave C-J]

Hi... hmm that flow works fine for me... so I guess ... which node.js ? which node-red ? Which OS ?

[dave....@gmail.com]

which node.js ?     v0.10.29

which node-red ?   v0.13.4 pre-installed

Which OS ?         Jessie - Linux 4.1.19-v7+ arm LE

[Dave C-J]

Ah.. .right - yes it was a "regression" I accidentally caused in that release - see this thread...

https://groups.google.com/forum/#!searchin/node-red/queue/node-red/-B9ZHGgEwtY/DuOHKUecAQAJ

so you can either pull that one file  - or upgrade to 0.14.

Apologies for the head-scratching this must have caused.

[dave....@gmail.com]

ok need to wait for    sudo apt-get install nodered     to catch up with latest release 14 then thanks.

[Dave C-J]

:-)... not long now... but as usual we have found a few small wrinkles in 0.14.3 that we would rather fix before dropping onto unsuspecting Pi users.