frodovdr/node-red
1
0
Fork 0
mirror of https://github.com/node-red/node-red.git synced 2025-03-01 10:36:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

let delay node handle both flush then reset

Browse Source 
and add tests
This commit is contained in:
Dave Conway-Jones
2022-09-22 10:51:48 +01:00
parent 30ea300f65
commit fc657ecc71
2 changed files with 138 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -201,6 +201,7 @@ module.exports = function(RED) {
});
node.on("close", function() { clearDelayList(); });
}
else if (node.pauseType === "random") {
node.on("input", function(msg, send, done) {
var wait = node.randomFirst + (node.diff * Math.random());
@@ -226,34 +227,19 @@ module.exports = function(RED) {
// The rate limit/queue type modes
else if (node.pauseType === "rate") {
node.on("input", function(msg, send, done) {
if (msg.hasOwnProperty("reset")) {
if (node.intervalID !== -1 ) {
clearInterval(node.intervalID);
node.intervalID = -1;
}
delete node.lastSent;
node.buffer = [];
node.rate = node.fixedrate;
node.status({fill:"blue",shape:"ring",text:0});
done();
return;
}
if (!node.drop) {
var m = RED.util.cloneMessage(msg);
delete m.flush;
delete m.lifo;
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;
if (node.allowrate && m.hasOwnProperty("rate") && !isNaN(parseFloat(m.rate)) && node.rate !== m.rate) {
node.rate = m.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);
node.error(RED._("delay.errors.too-many"), m);
} else if (msg.toFront === true) {
node.buffer.unshift({msg: m, send: send, done: done});
node.reportDepth();
@@ -263,8 +249,8 @@ module.exports = function(RED) {
}
}
else {
if (node.allowrate && msg.hasOwnProperty("rate") && !isNaN(parseFloat(msg.rate))) {
node.rate = msg.rate;
if (node.allowrate && m.hasOwnProperty("rate") && !isNaN(parseFloat(m.rate))) {
node.rate = m.rate;
}
send(m);
node.reportDepth();
@@ -282,6 +268,8 @@ module.exports = function(RED) {
else {
while (len > 0) {
const msgInfo = node.buffer.shift();
delete msgInfo.msg.flush;
delete msgInfo.msg.reset;
if (Object.keys(msgInfo.msg).length > 1) {
node.send(msgInfo.msg);
msgInfo.done();
@@ -335,6 +323,21 @@ module.exports = function(RED) {
}
done();
}
if (msg.hasOwnProperty("reset")) {
if (msg.flush === undefined) {
if (node.intervalID !== -1 ) {
clearInterval(node.intervalID);
node.intervalID = -1;
}
delete node.lastSent;
}
node.buffer = [];
node.rate = node.fixedrate;
node.status({fill:"blue",shape:"ring",text:0});
done();
return;
}
});
node.on("close", function() {
clearInterval(node.intervalID);
@@ -387,6 +390,22 @@ module.exports = function(RED) {
node.buffer.push({msg, send, done}); // if not add to end of queue
node.reportDepth();
}
if (msg.hasOwnProperty("flush")) {
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();
delete msgInfo.msg.flush;
delete msgInfo.msg.reset;
if (Object.keys(msgInfo.msg).length > 2) {
node.send(msgInfo.msg);
msgInfo.done();
}
len = len - 1;
}
node.status({});
done();
}
if (msg.hasOwnProperty("reset")) {
while (node.buffer.length > 0) {
const msgInfo = node.buffer.shift();
@@ -397,21 +416,6 @@ module.exports = function(RED) {
node.status({text:"reset"});
done();
}
if (msg.hasOwnProperty("flush")) {
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();
delete msgInfo.msg.flush;
if (Object.keys(msgInfo.msg).length > 2) {
node.send(msgInfo.msg);
msgInfo.done();
}
len = len - 1;
}
node.status({});
done();
}
});
node.on("close", function() {
clearInterval(node.intervalID);