frodovdr/node-red
1
0
Fork 0
mirror of https://github.com/node-red/node-red.git synced 2023-10-10 13:36:53 +02:00
Code Issues Releases Wiki Activity

Merge pull request #3059 from node-red/delay-queue

Browse Source 
Delay node updates to added numeric flush control for releasing things from queue
This commit is contained in:
Nick O'Leary 2021-07-13 11:37:13 +01:00 committed by GitHub
commit 5b24e8b69c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 174 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
packages/node_modules/@node-red/nodes/core/function/89-delay.html vendored
View File

@ -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":"";

124
packages/node_modules/@node-red/nodes/core/function/89-delay.js vendored
View File

@ -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<node.idList.length; i++ ) { node.idList[i].clear(); }
var len = node.idList.length;
for (var i=0; i<len; i++ ) { node.idList[i].clear(); }
node.idList = [];
if (s) { node.status({text:"reset"}); }
if (s) { node.status({fill:"blue",shape:"ring",text:0}); }
else { node.status({}); }
}
var flushDelayList = function() {
var flushDelayList = function(n) {
var len = node.idList.length;
if (typeof(n) == 'number') { len = Math.min(Math.floor(n),len); }
for (var i=0; i<len; i++ ) { node.idList[0].trigger(); }
node.idList = [];
node.status({text:"flushed"});
node.status({fill:"blue",shape:"dot",text:node.idList.length});
}
node.reportDepth = function() {
if (!node.busy) {
node.busy = setTimeout(function() {
if (node.buffer.length > 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(); });

3
packages/node_modules/@node-red/nodes/locales/en-US/function/89-delay.html vendored
View File

@ -32,7 +32,8 @@
<dd>If the received message has this property set to any value, all
outstanding messages held by the node are cleared without being sent.</dd>
<dt class="optional">flush</dt>
<dd>If the received message has this property set to any value, all
<dd>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.</dd>
</dl>
<h3>Details</h3>

99
test/nodes/core/function/89-delay_spec.js
View File

@ -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"]]},
@ -809,8 +901,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 +912,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}]);