From 41c8ca8ab481dbc7c970f93de1ae03390a6587f1 Mon Sep 17 00:00:00 2001 From: Dave Conway-Jones Date: Thu, 8 Jul 2021 12:13:23 +0100 Subject: [PATCH 1/2] Delay node updates to added flush control for queuing --- .../nodes/core/function/89-delay.html | 11 -- .../@node-red/nodes/core/function/89-delay.js | 124 +++++++++++------- .../locales/en-US/function/89-delay.html | 3 +- test/nodes/core/function/89-delay_spec.js | 7 +- 4 files changed, 82 insertions(+), 63 deletions(-) diff --git a/packages/node_modules/@node-red/nodes/core/function/89-delay.html b/packages/node_modules/@node-red/nodes/core/function/89-delay.html index 97cf05e39..cd8fef360 100644 --- a/packages/node_modules/@node-red/nodes/core/function/89-delay.html +++ b/packages/node_modules/@node-red/nodes/core/function/89-delay.html @@ -141,17 +141,6 @@ return this._("delay.label.limitTopic")+" "+rate; } } - // var units = ''; - // if (this.nbRateUnits > 1) { - // units = this.nbRateUnits + ' ' + this._("delay.label.units." + this.rateUnits + ".plural"); - // } else { - // units = this._("delay.label.units." + this.rateUnits + ".singular"); - // } - // return this.name || this.rate + " " + this._("delay.label.timed") + ' ' + units; - // } else { - // var units = this.rateUnits ? (this.nbRateUnits > 1 ? this.nbRateUnits : '') + this.rateUnits.charAt(0) : "s"; - // return this.name || this._("delay.label.queue")+" "+this.rate+" msg/"+units; - // } }, labelStyle: function() { return this.name?"node_label_italic":""; diff --git a/packages/node_modules/@node-red/nodes/core/function/89-delay.js b/packages/node_modules/@node-red/nodes/core/function/89-delay.js index 748421862..10f30796b 100644 --- a/packages/node_modules/@node-red/nodes/core/function/89-delay.js +++ b/packages/node_modules/@node-red/nodes/core/function/89-delay.js @@ -112,31 +112,35 @@ module.exports = function(RED) { } if (node.buffer.length > 0) { const msgInfo = node.buffer.shift(); - msgInfo.send(msgInfo.msg); - msgInfo.done(); + if (Object.keys(msgInfo.msg).length > 1) { + msgInfo.send(msgInfo.msg); + msgInfo.done(); + } } node.reportDepth(); } var clearDelayList = function(s) { - for (var i=0; i 0) { node.status({text:node.buffer.length}); } - else { node.status({}); } + // if (node.buffer.length > 0) { node.status({text:node.buffer.length}); } + // else { node.status({}); } + node.status({fill:"blue",shape:"dot",text:node.buffer.length}); node.busy = null; }, 500); } @@ -154,16 +158,18 @@ module.exports = function(RED) { node.on("input", function(msg, send, done) { var id = ourTimeout(function() { node.idList.splice(node.idList.indexOf(id),1); - if (node.idList.length === 0) { node.status({}); } + if (node.timeout > 1000) { + node.status({fill:"blue",shape:"dot",text:node.idList.length}); + } send(msg); done(); }, node.timeout, () => done()); if (Object.keys(msg).length === 2 && msg.hasOwnProperty("flush")) { id.clear(); } else { node.idList.push(id); } if (msg.hasOwnProperty("reset")) { clearDelayList(true); } - if (msg.hasOwnProperty("flush")) { flushDelayList(); done(); } - if ((node.timeout > 1000) && (node.idList.length !== 0)) { - node.status({fill:"blue",shape:"dot",text:" "}); + else if (msg.hasOwnProperty("flush")) { flushDelayList(msg.flush); done(); } + else if (node.timeout > 1000) { + node.status({fill:"blue",shape:"dot",text:node.idList.length}); } }); node.on("close", function() { clearDelayList(); }); @@ -179,13 +185,16 @@ module.exports = function(RED) { node.idList.splice(node.idList.indexOf(id),1); if (node.idList.length === 0) { node.status({}); } send(msg); + if (delayvar >= 0) { + node.status({fill:"blue",shape:"dot",text:node.idList.length}); + } done(); }, delayvar, () => done()); node.idList.push(id); if (msg.hasOwnProperty("reset")) { clearDelayList(true); } - if (msg.hasOwnProperty("flush")) { flushDelayList(); done(); } - if ((delayvar >= 0) && (node.idList.length !== 0)) { - node.status({fill:"blue",shape:"dot",text:delayvar/1000+"s"}); + if (msg.hasOwnProperty("flush")) { flushDelayList(msg.flush); done(); } + if (delayvar >= 0) { + node.status({fill:"blue",shape:"dot",text:node.idList.length}); } }); node.on("close", function() { clearDelayList(); }); @@ -200,7 +209,7 @@ module.exports = function(RED) { delete node.lastSent; node.buffer = []; node.rate = node.fixedrate; - node.status({text:"reset"}); + node.status(node.status({fill:"blue",shape:"ring",text:0})); done(); return; } @@ -208,37 +217,48 @@ module.exports = function(RED) { if (!node.drop) { var m = RED.util.cloneMessage(msg); delete m.flush; - if (node.intervalID !== -1) { - if (node.allowrate && msg.hasOwnProperty("rate") && !isNaN(parseFloat(msg.rate)) && node.rate !== msg.rate) { - node.rate = msg.rate; - clearInterval(node.intervalID); - node.intervalID = setInterval(sendMsgFromBuffer, node.rate); + if (Object.keys(m).length > 1) { + if (node.intervalID !== -1) { + if (node.allowrate && msg.hasOwnProperty("rate") && !isNaN(parseFloat(msg.rate)) && node.rate !== msg.rate) { + node.rate = msg.rate; + clearInterval(node.intervalID); + node.intervalID = setInterval(sendMsgFromBuffer, node.rate); + } + var max_msgs = maxKeptMsgsCount(node); + if ((max_msgs > 0) && (node.buffer.length >= max_msgs)) { + node.buffer = []; + node.error(RED._("delay.errors.too-many"), msg); + } else { + node.buffer.push({msg: m, send: send, done: done}); + node.reportDepth(); + } } - var max_msgs = maxKeptMsgsCount(node); - if ((max_msgs > 0) && (node.buffer.length >= max_msgs)) { - node.buffer = []; - node.error(RED._("delay.errors.too-many"), msg); - } else { - node.buffer.push({msg: m, send: send, done: done}); + else { + if (node.allowrate && msg.hasOwnProperty("rate") && !isNaN(parseFloat(msg.rate))) { + node.rate = msg.rate; + } + send(m); node.reportDepth(); + node.intervalID = setInterval(sendMsgFromBuffer, node.rate); + done(); } } - else { - if (node.allowrate && msg.hasOwnProperty("rate") && !isNaN(parseFloat(msg.rate))) { - node.rate = msg.rate; - } - send(m); - node.reportDepth(); - node.intervalID = setInterval(sendMsgFromBuffer, node.rate); - done(); - } if (msg.hasOwnProperty("flush")) { - while (node.buffer.length > 0) { + var len = node.buffer.length; + if (typeof(msg.flush) == 'number') { len = Math.min(Math.floor(msg.flush),len); } + while (len > 0) { const msgInfo = node.buffer.shift(); - msgInfo.send(msgInfo.msg); - msgInfo.done(); + if (Object.keys(msgInfo.msg).length > 1) { + node.send(msgInfo.msg); + msgInfo.done(); + } + len = len - 1; } - node.status({}); + if (node.buffer.length === 0) { + clearInterval(node.intervalID); + node.intervalID = -1; + } + node.status({fill:"blue",shape:"dot",text:node.buffer.length}); done(); } } @@ -340,10 +360,16 @@ module.exports = function(RED) { done(); } if (msg.hasOwnProperty("flush")) { - while (node.buffer.length > 0) { + var len = node.buffer.length; + if (typeof(msg.flush) == 'number') { len = Math.min(Math.floor(msg.flush,len)); } + while (len > 0) { const msgInfo = node.buffer.shift(); - msgInfo.send(msgInfo.msg); - msgInfo.done(); + delete msgInfo.msg.flush; + if (Object.keys(msgInfo.msg).length > 2) { + node.send(msgInfo.msg); + msgInfo.done(); + } + len = len - 1; } node.status({}); done(); @@ -365,15 +391,17 @@ module.exports = function(RED) { var id = ourTimeout(function() { node.idList.splice(node.idList.indexOf(id),1); send(msg); - node.status({}); + if (node.timeout >= 1000) { + node.status({fill:"blue",shape:"dot",text:node.idList.length}); + } done(); }, wait, () => done()); if (Object.keys(msg).length === 2 && msg.hasOwnProperty("flush")) { id.clear(); } else { node.idList.push(id); } if (msg.hasOwnProperty("reset")) { clearDelayList(true); } - if (msg.hasOwnProperty("flush")) { flushDelayList(); done(); } - if ((node.timeout >= 1000) && (node.idList.length !== 0)) { - node.status({fill:"blue",shape:"dot",text:parseInt(wait/10)/100+"s"}); + if (msg.hasOwnProperty("flush")) { flushDelayList(msg.flush); done(); } + if (node.timeout >= 1000) { + node.status({fill:"blue",shape:"dot",text:node.idList.length}); } }); node.on("close", function() { clearDelayList(); }); diff --git a/packages/node_modules/@node-red/nodes/locales/en-US/function/89-delay.html b/packages/node_modules/@node-red/nodes/locales/en-US/function/89-delay.html index 632cd9429..1e801698a 100644 --- a/packages/node_modules/@node-red/nodes/locales/en-US/function/89-delay.html +++ b/packages/node_modules/@node-red/nodes/locales/en-US/function/89-delay.html @@ -32,7 +32,8 @@
If the received message has this property set to any value, all outstanding messages held by the node are cleared without being sent.
flush
-
If the received message has this property set to any value, all +
If the received message has this property set to a numeric value then that many messages + will be released immediately. If set to any other type (e.g. boolean), then all outstanding messages held by the node are sent immediately.

Details

diff --git a/test/nodes/core/function/89-delay_spec.js b/test/nodes/core/function/89-delay_spec.js index 3fa5f9de8..8a74901e0 100644 --- a/test/nodes/core/function/89-delay_spec.js +++ b/test/nodes/core/function/89-delay_spec.js @@ -809,8 +809,9 @@ describe('delay Node', function() { {msg:{payload:3,flush:true}, avr:0, var:100}]); }); it('calls done when queued messages are sent (queue)', function(done) { - mapiDoneTestHelper(done, "queue", false, [{msg:{payload:1,topic:"a"}, avr:500, var:700}, - {msg:{payload:2, topic:"b"}, avr:1500, var:700}]); + this.timeout(3000); + mapiDoneTestHelper(done, "queue", false, [{msg:{payload:1,topic:"A"}, avr:1000, var:700}, + {msg:{payload:2,topic:"B"}, avr:2000, var:700}]); }); it('calls done when queued messages are sent (timed)', function(done) { mapiDoneTestHelper(done, "timed", false, [{msg:{payload:1,topic:"a"}, avr:500, var:700}, @@ -819,7 +820,7 @@ describe('delay Node', function() { it('calls done when queue is reset (queue/timed)', function(done) { mapiDoneTestHelper(done, "timed", false, [{msg:{payload:1,topic:"a"}, avr:0, var:500}, {msg:{payload:2,reset:true}, avr:0, var:500}]); - }); + }); it('calls done when queue is flushed (queue/timed)', function(done) { mapiDoneTestHelper(done, "timed", false, [{msg:{payload:1,topic:"a"}, avr:0, var:500}, {msg:{payload:2,flush:true}, avr:0, var:500}]); From ea11aa7a0d39d180a5941390b79f07de2914fde9 Mon Sep 17 00:00:00 2001 From: Dave Conway-Jones Date: Thu, 8 Jul 2021 13:51:37 +0100 Subject: [PATCH 2/2] add some part flush tests --- test/nodes/core/function/89-delay_spec.js | 92 +++++++++++++++++++++++ 1 file changed, 92 insertions(+) diff --git a/test/nodes/core/function/89-delay_spec.js b/test/nodes/core/function/89-delay_spec.js index 8a74901e0..87ca9c434 100644 --- a/test/nodes/core/function/89-delay_spec.js +++ b/test/nodes/core/function/89-delay_spec.js @@ -629,6 +629,52 @@ describe('delay Node', function() { }); }); + it('can part flush delay queue', function(done) { + this.timeout(2000); + var flow = [{"id":"delayNode1","type":"delay","name":"delayNode","pauseType":"delay","timeout":1,"timeoutUnits":"seconds","rate":2,"rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"wires":[["helperNode1"]]}, + {id:"helperNode1", type:"helper", wires:[]}]; + helper.load(delayNode, flow, function() { + var delayNode1 = helper.getNode("delayNode1"); + var helperNode1 = helper.getNode("helperNode1"); + var t = Date.now(); + var c = 0; + helperNode1.on("input", function(msg) { + msg.should.have.a.property('payload'); + msg.should.have.a.property('topic'); + try { + if (msg.topic === "foo") { + msg.payload.should.equal(1); + (Date.now() - t).should.be.approximately(0,100); + c = c + 1; + } + else if (msg.topic === "bar") { + msg.payload.should.equal(1); + (Date.now() - t).should.be.approximately(200,100); + c = c + 1; + } + else if (msg.topic === "boo") { + msg.payload.should.equal(1); + (Date.now() - t).should.be.approximately(400,100); + c = c + 1; + } + if (c === 5) { done(); } + } catch(e) { + done(e); + } + }); + + // send test messages + delayNode1.receive({payload:1,topic:"foo"}); + setImmediate( function() { delayNode1.receive({payload:1,topic:"foo"}); } ); + setImmediate( function() { delayNode1.receive({payload:1,topic:"bar"}); } ); + setImmediate( function() { delayNode1.receive({payload:1,topic:"boo"}); } ); + setImmediate( function() { delayNode1.receive({payload:1,topic:"boo"}); } ); + setImmediate( function() { delayNode1.receive({flush:2}); }); + setTimeout( function() { delayNode1.receive({flush:1}); }, 200); + setTimeout( function() { delayNode1.receive({flush:4}); }, 400); + }); + }); + it('can reset delay queue', function(done) { this.timeout(2000); var flow = [{"id":"delayNode1","type":"delay","name":"delayNode","pauseType":"delay","timeout":1,"timeoutUnits":"seconds","rate":2,"rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"wires":[["helperNode1"]]}, @@ -697,6 +743,52 @@ describe('delay Node', function() { }); }); + it('can part flush rate limit queue', function(done) { + this.timeout(2000); + var flow = [{"id":"delayNode1","type":"delay","name":"delayNode","pauseType":"rate","timeout":1,"timeoutUnits":"seconds","rate":2,"rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"wires":[["helperNode1"]]}, + {id:"helperNode1", type:"helper", wires:[]}]; + helper.load(delayNode, flow, function() { + var delayNode1 = helper.getNode("delayNode1"); + var helperNode1 = helper.getNode("helperNode1"); + var t = Date.now(); + var c = 0; + helperNode1.on("input", function(msg) { + msg.should.have.a.property('payload'); + msg.should.have.a.property('topic'); + try { + if (msg.topic === "foo") { + msg.payload.should.equal(1); + (Date.now() - t).should.be.approximately(0,100); + c = c + 1; + } + else if (msg.topic === "bar") { + msg.payload.should.equal(1); + (Date.now() - t).should.be.approximately(200,100); + c = c + 1; + } + else if (msg.topic === "boo") { + msg.payload.should.equal(1); + (Date.now() - t).should.be.approximately(400,100); + c = c + 1; + } + if (c === 5) { done(); } + } catch(e) { + done(e); + } + }); + + // send test messages + delayNode1.receive({payload:1,topic:"foo"}); + setImmediate( function() { delayNode1.receive({payload:1,topic:"foo"}); } ); + setImmediate( function() { delayNode1.receive({payload:1,topic:"foo"}); } ); + setImmediate( function() { delayNode1.receive({payload:1,topic:"bar"}); } ); + setImmediate( function() { delayNode1.receive({payload:1,topic:"boo"}); } ); + setImmediate( function() { delayNode1.receive({flush:2}); }); + setTimeout( function() { delayNode1.receive({flush:1}); }, 200); + setTimeout( function() { delayNode1.receive({flush:4}); }, 400); + }); + }); + it('can reset rate limit queue', function(done) { this.timeout(2000); var flow = [{"id":"delayNode1","type":"delay","name":"delayNode","pauseType":"rate","timeout":1,"timeoutUnits":"seconds","rate":2,"rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"wires":[["helperNode1"]]},