Can the Delay node rate limit, but pass first message immediately, and always pass on the latest message?

Colin · 31 August 2022 08:34

I have a scenario with messages coming in at random intervals. Often a long way apart but occasionally close together. These are the goals

When a message comes in and there has been more than five seconds since the last message then the new one is passed on immediately.
If a second message comes in less than five seconds after the first then it will be queued and released at the end of the five seconds unless ...
If a third (and subsequent) message comes in when there is already one queued then it replaces the one already in the queue and will be released at the end of the period.

So the purpose is to rate limit, but immediately to pass on the first message, and to guarantee that the latest message is always delivered eventually. I can't work out how to do it with the Delay node. It is nearly achievable in All Topics mode, but in that mode it doesn't send the first message immediately.

Trying_to_learn · 31 August 2022 09:28

You may need to expand on that.

What happens if there is a message arrives and then nothing else?

Colin · 31 August 2022 09:42

That is covered by case 1

Trying_to_learn · 31 August 2022 09:43

(Sorry, I didn't parse it like that.)

Got it now.

Trying_to_learn · 31 August 2022 09:43

Doesn't this do what you want then?

[
    {
        "id": "791af1ddf9a1328d",
        "type": "inject",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "props": [
            {
                "p": "payload"
            },
            {
                "p": "topic",
                "vt": "str"
            }
        ],
        "repeat": "",
        "crontab": "",
        "once": false,
        "onceDelay": 0.1,
        "topic": "",
        "payload": "blah",
        "payloadType": "str",
        "x": 360,
        "y": 2170,
        "wires": [
            [
                "9b79800d4b74a795"
            ]
        ]
    },
    {
        "id": "9b79800d4b74a795",
        "type": "delay",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "pauseType": "rate",
        "timeout": "5",
        "timeoutUnits": "seconds",
        "rate": "1",
        "nbRateUnits": "5",
        "rateUnits": "second",
        "randomFirst": "1",
        "randomLast": "5",
        "randomUnits": "seconds",
        "drop": false,
        "allowrate": false,
        "outputs": 1,
        "x": 540,
        "y": 2170,
        "wires": [
            [
                "585d6add0477b5c6"
            ]
        ]
    },
    {
        "id": "585d6add0477b5c6",
        "type": "debug",
        "z": "f7d630bfc1b2f259",
        "name": "OUT",
        "active": true,
        "tosidebar": true,
        "console": false,
        "tostatus": false,
        "complete": "payload",
        "targetType": "msg",
        "statusVal": "",
        "statusType": "auto",
        "x": 700,
        "y": 2170,
        "wires": []
    }
]

Colin · 31 August 2022 09:57

Does it work for case 3?

Colin · 31 August 2022 10:32

It is a pity that sending {flush: 1,reset: true} to the Delay node does not flush and then reset. It appears to reset first, so there is nothing to flush. Sending them as two messages is not reliable as there is a window between them where a message could arrive and then get lost because of the reset.

Trying_to_learn · 31 August 2022 10:35

You may need to look at other nodes in the library.
https://flows.nodered.org/search?type=node

Colin · 31 August 2022 10:36

I have been looking. Any suggestions?

Trying_to_learn · 31 August 2022 10:38

q-gate may work.

It allows you to queue messages and spit out selected ones.

Trying_to_learn · 31 August 2022 10:41

(Sorry)

Could you try with point 3 again.

There are messages in the queue (as per 1 and 2) and a third message arrives.

Which messages are then on-sent?
The older or newer ones?

Trying_to_learn · 31 August 2022 10:54

Ok, not exact, but how about this:

[
    {
        "id": "8ac365e46b76938d",
        "type": "switch",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "property": "count",
        "propertyType": "msg",
        "rules": [
            {
                "t": "eq",
                "v": "2",
                "vt": "num"
            },
            {
                "t": "eq",
                "v": "1",
                "vt": "num"
            },
            {
                "t": "gt",
                "v": "2",
                "vt": "num"
            }
        ],
        "checkall": "true",
        "repair": false,
        "outputs": 3,
        "x": 530,
        "y": 1550,
        "wires": [
            [
                "b0826138af4b91c4"
            ],
            [
                "06721c1d68f7bfc3"
            ],
            [
                "d15070a114653524",
                "06721c1d68f7bfc3"
            ]
        ]
    },
    {
        "id": "5bd3658d5a3a11a7",
        "type": "counter",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "init": "0",
        "step": "1",
        "lower": null,
        "upper": null,
        "mode": "increment",
        "outputs": "1",
        "x": 360,
        "y": 1550,
        "wires": [
            [
                "8ac365e46b76938d"
            ]
        ]
    },
    {
        "id": "b0826138af4b91c4",
        "type": "delay",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "pauseType": "delay",
        "timeout": "5",
        "timeoutUnits": "seconds",
        "rate": "1",
        "nbRateUnits": "1",
        "rateUnits": "second",
        "randomFirst": "1",
        "randomLast": "5",
        "randomUnits": "seconds",
        "drop": false,
        "allowrate": false,
        "outputs": 1,
        "x": 700,
        "y": 1550,
        "wires": [
            [
                "06721c1d68f7bfc3"
            ]
        ]
    },
    {
        "id": "06721c1d68f7bfc3",
        "type": "debug",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "active": true,
        "tosidebar": true,
        "console": false,
        "tostatus": false,
        "complete": "true",
        "targetType": "full",
        "statusVal": "",
        "statusType": "auto",
        "x": 680,
        "y": 1670,
        "wires": []
    },
    {
        "id": "d15070a114653524",
        "type": "change",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "rules": [
            {
                "t": "set",
                "p": "reset",
                "pt": "msg",
                "to": "reset",
                "tot": "str"
            },
            {
                "t": "delete",
                "p": "payload",
                "pt": "msg"
            }
        ],
        "action": "",
        "property": "",
        "from": "",
        "to": "",
        "reg": false,
        "x": 700,
        "y": 1510,
        "wires": [
            [
                "b0826138af4b91c4"
            ]
        ]
    },
    {
        "id": "03284863bd452713",
        "type": "inject",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "props": [
            {
                "p": "payload"
            },
            {
                "p": "topic",
                "vt": "str"
            }
        ],
        "repeat": "",
        "crontab": "",
        "once": false,
        "onceDelay": 0.1,
        "topic": "",
        "payload": "",
        "payloadType": "date",
        "x": 100,
        "y": 1640,
        "wires": [
            [
                "5bd3658d5a3a11a7",
                "5406068f12e1eeac"
            ]
        ]
    },
    {
        "id": "acbdf11c584a1633",
        "type": "change",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "rules": [
            {
                "t": "set",
                "p": "reset",
                "pt": "msg",
                "to": "reset",
                "tot": "str"
            },
            {
                "t": "delete",
                "p": "payload",
                "pt": "msg"
            }
        ],
        "action": "",
        "property": "",
        "from": "",
        "to": "",
        "reg": false,
        "x": 370,
        "y": 1510,
        "wires": [
            [
                "5bd3658d5a3a11a7"
            ]
        ]
    },
    {
        "id": "5406068f12e1eeac",
        "type": "delay",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "pauseType": "delay",
        "timeout": "5",
        "timeoutUnits": "seconds",
        "rate": "1",
        "nbRateUnits": "1",
        "rateUnits": "second",
        "randomFirst": "1",
        "randomLast": "5",
        "randomUnits": "seconds",
        "drop": false,
        "allowrate": false,
        "outputs": 1,
        "x": 140,
        "y": 1550,
        "wires": [
            [
                "e0da96193f54affc"
            ]
        ]
    },
    {
        "id": "e0da96193f54affc",
        "type": "delay",
        "z": "f7d630bfc1b2f259",
        "name": "",
        "pauseType": "rate",
        "timeout": "5",
        "timeoutUnits": "seconds",
        "rate": "1",
        "nbRateUnits": "5",
        "rateUnits": "second",
        "randomFirst": "1",
        "randomLast": "5",
        "randomUnits": "seconds",
        "drop": true,
        "allowrate": false,
        "outputs": 1,
        "x": 160,
        "y": 1510,
        "wires": [
            [
                "acbdf11c584a1633"
            ]
        ]
    }
]

Foreign nodes.
node-red-contrib-counter

dceejay · 31 August 2022 10:57

errrr flush then reset doesn't make sense (to me) - as once flushed it is in the same state as reset (IE nothing in Q) - so the reset is superfluous.

Trying_to_learn · 31 August 2022 11:00

This may be better.

[{"id":"8ac365e46b76938d","type":"switch","z":"f7d630bfc1b2f259","name":"","property":"count","propertyType":"msg","rules":[{"t":"eq","v":"2","vt":"num"},{"t":"eq","v":"1","vt":"num"},{"t":"gt","v":"2","vt":"num"}],"checkall":"true","repair":false,"outputs":3,"x":370,"y":1600,"wires":[["b0826138af4b91c4"],["06721c1d68f7bfc3"],["d15070a114653524","6cc4a6c19d7efbc8"]]},{"id":"5bd3658d5a3a11a7","type":"counter","z":"f7d630bfc1b2f259","name":"","init":"0","step":"1","lower":null,"upper":null,"mode":"increment","outputs":"1","x":360,"y":1550,"wires":[["8ac365e46b76938d"]]},{"id":"b0826138af4b91c4","type":"delay","z":"f7d630bfc1b2f259","name":"","pauseType":"delay","timeout":"5","timeoutUnits":"seconds","rate":"1","nbRateUnits":"1","rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"allowrate":false,"outputs":1,"x":700,"y":1550,"wires":[["06721c1d68f7bfc3"]]},{"id":"06721c1d68f7bfc3","type":"debug","z":"f7d630bfc1b2f259","name":"","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"true","targetType":"full","statusVal":"","statusType":"auto","x":720,"y":1670,"wires":[]},{"id":"d15070a114653524","type":"change","z":"f7d630bfc1b2f259","name":"","rules":[{"t":"set","p":"reset","pt":"msg","to":"reset","tot":"str"},{"t":"delete","p":"payload","pt":"msg"}],"action":"","property":"","from":"","to":"","reg":false,"x":700,"y":1510,"wires":[["b0826138af4b91c4"]]},{"id":"6cc4a6c19d7efbc8","type":"delay","z":"f7d630bfc1b2f259","name":"","pauseType":"rate","timeout":"5","timeoutUnits":"seconds","rate":"1","nbRateUnits":"5","rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"allowrate":false,"outputs":1,"x":380,"y":1670,"wires":[["ac8840e7f8c36742"]]},{"id":"03284863bd452713","type":"inject","z":"f7d630bfc1b2f259","name":"","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","payload":"","payloadType":"date","x":100,"y":1640,"wires":[["5bd3658d5a3a11a7","d64f36e617cfaa63"]]},{"id":"acbdf11c584a1633","type":"change","z":"f7d630bfc1b2f259","name":"","rules":[{"t":"set","p":"reset","pt":"msg","to":"reset","tot":"str"},{"t":"delete","p":"payload","pt":"msg"}],"action":"","property":"","from":"","to":"","reg":false,"x":370,"y":1510,"wires":[["5bd3658d5a3a11a7"]]},{"id":"ac8840e7f8c36742","type":"delay","z":"f7d630bfc1b2f259","name":"","pauseType":"delay","timeout":"5","timeoutUnits":"seconds","rate":"1","nbRateUnits":"1","rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"allowrate":false,"outputs":1,"x":560,"y":1670,"wires":[["06721c1d68f7bfc3"]]},{"id":"d64f36e617cfaa63","type":"trigger","z":"f7d630bfc1b2f259","name":"","op1":"","op2":"0","op1type":"nul","op2type":"str","duration":"5","extend":true,"overrideDelay":false,"units":"s","reset":"","bytopic":"all","topic":"topic","outputs":1,"x":140,"y":1550,"wires":[["e0da96193f54affc"]]},{"id":"e0da96193f54affc","type":"delay","z":"f7d630bfc1b2f259","name":"","pauseType":"rate","timeout":"5","timeoutUnits":"seconds","rate":"1","nbRateUnits":"5","rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":true,"allowrate":false,"outputs":1,"x":160,"y":1510,"wires":[["acbdf11c584a1633"]]}]

But still not perfect.

jbudd · 31 August 2022 11:04

My best attempt:

[{"id":"864772ef56f82916","type":"tab","label":"Flow 2","disabled":false,"info":"","env":[]},{"id":"6c7f6a05311c3f9d","type":"group","z":"864772ef56f82916","style":{"stroke":"#999999","stroke-opacity":"1","fill":"none","fill-opacity":"1","label":true,"label-position":"nw","color":"#a4a4a4"},"nodes":["169b911cfc6e0ee4","508272893fdea567","3c7b3c50819be497","b4b5e64c6a8d1a89"],"x":314,"y":79,"w":452,"h":142},{"id":"b4b5e64c6a8d1a89","type":"junction","z":"864772ef56f82916","g":"6c7f6a05311c3f9d","x":340,"y":160,"wires":[["508272893fdea567","169b911cfc6e0ee4"]]},{"id":"8416ee8d414b6c99","type":"inject","z":"864772ef56f82916","name":"","props":[{"p":"payload"}],"repeat":"","crontab":"","once":false,"onceDelay":"2","topic":"","payload":"Wales","payloadType":"str","x":210,"y":120,"wires":[["b4b5e64c6a8d1a89"]]},{"id":"bc2001d5d56c5276","type":"inject","z":"864772ef56f82916","name":"","props":[{"p":"payload"}],"repeat":"","crontab":"","once":false,"onceDelay":"3","topic":"","payload":"Scotland","payloadType":"str","x":220,"y":160,"wires":[["b4b5e64c6a8d1a89"]]},{"id":"c5d6473c292a3bde","type":"inject","z":"864772ef56f82916","name":"","props":[{"p":"payload"}],"repeat":"","crontab":"","once":false,"onceDelay":"4","topic":"","payload":"England","payloadType":"str","x":220,"y":200,"wires":[["b4b5e64c6a8d1a89"]]},{"id":"d5c6363f1343d657","type":"debug","z":"864772ef56f82916","name":"debug 4","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"false","statusVal":"","statusType":"auto","x":880,"y":180,"wires":[]},{"id":"169b911cfc6e0ee4","type":"q-gate","z":"864772ef56f82916","g":"6c7f6a05311c3f9d","name":"","controlTopic":"control","defaultState":"open","openCmd":"open","closeCmd":"close","toggleCmd":"toggle","queueCmd":"queue","defaultCmd":"default","triggerCmd":"trigger","flushCmd":"flush","resetCmd":"reset","peekCmd":"peek","dropCmd":"drop","statusCmd":"status","maxQueueLength":"1","keepNewest":true,"qToggle":false,"persist":false,"storeName":"memory","x":490,"y":180,"wires":[["d5c6363f1343d657"]]},{"id":"508272893fdea567","type":"trigger","z":"864772ef56f82916","g":"6c7f6a05311c3f9d","name":"Set mode Queue","op1":"queue","op2":"open","op1type":"str","op2type":"str","duration":"5","extend":false,"overrideDelay":false,"units":"s","reset":"","bytopic":"all","topic":"topic","outputs":1,"x":450,"y":120,"wires":[["3c7b3c50819be497"]]},{"id":"3c7b3c50819be497","type":"change","z":"864772ef56f82916","g":"6c7f6a05311c3f9d","name":"topic = control","rules":[{"t":"set","p":"topic","pt":"msg","to":"control","tot":"str"}],"action":"","property":"","from":"","to":"","reg":false,"x":660,"y":120,"wires":[["169b911cfc6e0ee4"]]}]

The node-red-contrib-queue-gate defaults to Open but a message (after it has passed through) sets it to Queue mode for 5 sec and only the latest 1 message is queued.

I thought it might need a millisecond or two delay before switching modes but so far I have not seen a problem.

Oh, and I originally set up the grouped bit as a subflow, but since it discards some messages you get link node timeout errors. Edit - not a subflow, link in and link out.

Edit 2:
This works too, maybe better since there is less possibility of a race between msg1 passing the gate and setting the gate to Queue.
But feedback loops ring mental alarm bells. No actual problem here I think.

Trying_to_learn · 31 August 2022 11:06

Not bad.

But isn't there a problem with scenario 3. Message 2 would still get through.

I read it as: If a third message arrives (and there is a queued message) the queued message is dropped and 3 is sent after the timeout time.
(But I am not sure)

jbudd · 31 August 2022 11:14

These two settings allow only the most recently queued message to emerge after 5 seconds.

Trying_to_learn · 31 August 2022 11:15

Sorry...

I use that node myself, and overlooked that part.

(I've got to learn to read)

I was working on doing just that and made my convoluted mess.
Glad you saw that trick.

Yeah, your way looks better.

But what happens if there are other (more) messages coming through?
Like 4, 5 or 6?

dceejay · 31 August 2022 11:25

As long as the actual messages are different then actually a simple trigger node and filter may work...

[{"id":"ce4b2f27d2d181d8","type":"inject","z":"a7fc399c358cfac0","name":"","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"A","payload":"1","payloadType":"num","x":215,"y":360,"wires":[["8c004dc7912dc3ac"]]},{"id":"8c004dc7912dc3ac","type":"trigger","z":"a7fc399c358cfac0","name":"","op1":"","op2":"","op1type":"pay","op2type":"payl","duration":"5","extend":false,"overrideDelay":false,"units":"s","reset":"","bytopic":"all","topic":"topic","outputs":1,"x":440,"y":405,"wires":[["f2d6fb1495637cd9"]]},{"id":"f2d6fb1495637cd9","type":"rbe","z":"a7fc399c358cfac0","name":"","func":"rbe","gap":"","start":"","inout":"out","septopics":true,"property":"payload","topi":"topic","x":625,"y":405,"wires":[["8317fdb6d1bae677"]]},{"id":"d88cc3cace4cda48","type":"inject","z":"a7fc399c358cfac0","name":"","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"A","payload":"2","payloadType":"num","x":215,"y":405,"wires":[["8c004dc7912dc3ac"]]},{"id":"87acfb08b3f28e66","type":"inject","z":"a7fc399c358cfac0","name":"","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"A","payload":"3","payloadType":"num","x":215,"y":450,"wires":[["8c004dc7912dc3ac"]]},{"id":"8317fdb6d1bae677","type":"debug","z":"a7fc399c358cfac0","name":"debug 3","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"false","statusVal":"","statusType":"auto","x":795,"y":405,"wires":[]}]

Trying_to_learn · 31 August 2022 11:29

Ok, so can messages be on-going so you get the first message then a second, third and the more?

To what limit?

Are they all to be kept?

Is there an upper limit?

In all of these it would seem Message2 gets dropped and all you see are message 1, 3 and subsequent message at 5 second intervals.

Topic		Replies	Views
Delay node rate limit General	1	354	31 May 2021
Delay node rate limit by topic starts limiting General delay	8	123	25 January 2024
Delay node, rate limit don't send first message General	10	3119	26 February 2019
Rate Limit per topic, releasing first message General	5	1335	16 March 2022
Delay node with rate limit by x msg / x time unit Feature Requests delay , function-node	12	277	8 April 2023

Can the Delay node rate limit, but pass first message immediately, and always pass on the latest message?

Related Topics