mirror of
https://github.com/node-red/node-red.git
synced 2025-03-01 10:36:34 +00:00
0f45b1da48
behaviour is different. May cause minor breakage. if flush arrives with payload that is now included with the flushed data so no data is lost. Previously any payload was dropped.
375 lines
15 KiB
JavaScript
375 lines
15 KiB
JavaScript
/**
|
|
* Copyright JS Foundation and other contributors, http://js.foundation
|
|
*
|
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
|
* you may not use this file except in compliance with the License.
|
|
* You may obtain a copy of the License at
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
* See the License for the specific language governing permissions and
|
|
* limitations under the License.
|
|
**/
|
|
|
|
//Simple node to introduce a pause into a flow
|
|
module.exports = function(RED) {
|
|
"use strict";
|
|
|
|
var MILLIS_TO_NANOS = 1000000;
|
|
var SECONDS_TO_NANOS = 1000000000;
|
|
var _maxKeptMsgsCount;
|
|
|
|
function maxKeptMsgsCount(node) {
|
|
if (_maxKeptMsgsCount === undefined) {
|
|
var name = "nodeMessageBufferMaxLength";
|
|
if (RED.settings.hasOwnProperty(name)) {
|
|
_maxKeptMsgsCount = RED.settings[name];
|
|
}
|
|
else {
|
|
_maxKeptMsgsCount = 0;
|
|
}
|
|
}
|
|
return _maxKeptMsgsCount;
|
|
}
|
|
|
|
function DelayNode(n) {
|
|
RED.nodes.createNode(this,n);
|
|
|
|
this.pauseType = n.pauseType;
|
|
this.timeoutUnits = n.timeoutUnits;
|
|
this.randomUnits = n.randomUnits;
|
|
this.rateUnits = n.rateUnits;
|
|
|
|
if (n.timeoutUnits === "milliseconds") {
|
|
this.timeout = n.timeout;
|
|
} else if (n.timeoutUnits === "minutes") {
|
|
this.timeout = n.timeout * (60 * 1000);
|
|
} else if (n.timeoutUnits === "hours") {
|
|
this.timeout = n.timeout * (60 * 60 * 1000);
|
|
} else if (n.timeoutUnits === "days") {
|
|
this.timeout = n.timeout * (24 * 60 * 60 * 1000);
|
|
} else { // Default to seconds
|
|
this.timeout = n.timeout * 1000;
|
|
}
|
|
|
|
if (n.rateUnits === "minute") {
|
|
this.rate = (60 * 1000)/n.rate;
|
|
} else if (n.rateUnits === "hour") {
|
|
this.rate = (60 * 60 * 1000)/n.rate;
|
|
} else if (n.rateUnits === "day") {
|
|
this.rate = (24 * 60 * 60 * 1000)/n.rate;
|
|
} else { // Default to seconds
|
|
this.rate = 1000/n.rate;
|
|
}
|
|
|
|
this.rate *= (n.nbRateUnits > 0 ? n.nbRateUnits : 1);
|
|
|
|
if (n.randomUnits === "milliseconds") {
|
|
this.randomFirst = n.randomFirst * 1;
|
|
this.randomLast = n.randomLast * 1;
|
|
} else if (n.randomUnits === "minutes") {
|
|
this.randomFirst = n.randomFirst * (60 * 1000);
|
|
this.randomLast = n.randomLast * (60 * 1000);
|
|
} else if (n.randomUnits === "hours") {
|
|
this.randomFirst = n.randomFirst * (60 * 60 * 1000);
|
|
this.randomLast = n.randomLast * (60 * 60 * 1000);
|
|
} else if (n.randomUnits === "days") {
|
|
this.randomFirst = n.randomFirst * (24 * 60 * 60 * 1000);
|
|
this.randomLast = n.randomLast * (24 * 60 * 60 * 1000);
|
|
} else { // Default to seconds
|
|
this.randomFirst = n.randomFirst * 1000;
|
|
this.randomLast = n.randomLast * 1000;
|
|
}
|
|
|
|
this.diff = this.randomLast - this.randomFirst;
|
|
this.name = n.name;
|
|
this.idList = [];
|
|
this.buffer = [];
|
|
this.intervalID = -1;
|
|
this.randomID = -1;
|
|
this.lastSent = null;
|
|
this.drop = n.drop;
|
|
this.droppedMsgs = 0;
|
|
var node = this;
|
|
|
|
function ourTimeout(handler, delay, clearHandler) {
|
|
var toutID = setTimeout(handler, delay);
|
|
return {
|
|
clear: function() { clearTimeout(toutID); clearHandler(); },
|
|
trigger: function() { clearTimeout(toutID); return handler(); }
|
|
};
|
|
}
|
|
|
|
var sendMsgFromBuffer = function() {
|
|
if (node.buffer.length === 0) {
|
|
clearInterval(node.intervalID);
|
|
node.intervalID = -1;
|
|
}
|
|
if (node.buffer.length > 0) {
|
|
const msgInfo = node.buffer.shift();
|
|
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(); }
|
|
node.idList = [];
|
|
if (s) { node.status({text:"reset"}); }
|
|
else { node.status({}); }
|
|
}
|
|
|
|
var flushDelayList = function() {
|
|
var len = node.idList.length;
|
|
for (var i=0; i<len; i++ ) { node.idList[0].trigger(); }
|
|
node.idList = [];
|
|
node.status({text:"flushed"});
|
|
}
|
|
|
|
node.reportDepth = function() {
|
|
if (!node.busy) {
|
|
node.busy = setTimeout(function() {
|
|
if (node.buffer.length > 0) { node.status({text:node.buffer.length}); }
|
|
else { node.status({}); }
|
|
node.busy = null;
|
|
}, 500);
|
|
}
|
|
}
|
|
|
|
var loggerId = setInterval(function () {
|
|
if (node.droppedMsgs !== 0) {
|
|
node.debug("node.droppedMsgs = " + node.droppedMsgs);
|
|
node.droppedMsgs = 0;
|
|
}
|
|
}, 15 * 1000);
|
|
node.on("close", function() { clearInterval(loggerId); });
|
|
|
|
if (node.pauseType === "delay") {
|
|
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({}); }
|
|
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:" "});
|
|
}
|
|
});
|
|
node.on("close", function() { clearDelayList(); });
|
|
}
|
|
else if (node.pauseType === "delayv") {
|
|
node.on("input", function(msg, send, done) {
|
|
var delayvar = Number(node.timeout);
|
|
if (msg.hasOwnProperty("delay") && !isNaN(parseFloat(msg.delay))) {
|
|
delayvar = parseFloat(msg.delay);
|
|
}
|
|
if (delayvar < 0) { delayvar = 0; }
|
|
var id = ourTimeout(function() {
|
|
node.idList.splice(node.idList.indexOf(id),1);
|
|
if (node.idList.length === 0) { node.status({}); }
|
|
send(msg);
|
|
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"});
|
|
}
|
|
});
|
|
node.on("close", function() { clearDelayList(); });
|
|
}
|
|
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.status({text:"reset"});
|
|
done();
|
|
return;
|
|
}
|
|
|
|
if (!node.drop) {
|
|
var m = RED.util.cloneMessage(msg);
|
|
delete m.flush;
|
|
if (node.intervalID !== -1) {
|
|
if (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();
|
|
}
|
|
}
|
|
else {
|
|
if (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) {
|
|
const msgInfo = node.buffer.shift();
|
|
msgInfo.send(msgInfo.msg);
|
|
msgInfo.done();
|
|
}
|
|
node.status({});
|
|
done();
|
|
}
|
|
}
|
|
else {
|
|
if (maxKeptMsgsCount(node) > 0) {
|
|
if (node.intervalID === -1) {
|
|
node.send(msg);
|
|
node.intervalID = setInterval(sendMsgFromBuffer, node.rate);
|
|
} else {
|
|
if (msg.hasOwnProperty("rate") && node.rate !== msg.rate) {
|
|
node.rate = msg.rate;
|
|
clearInterval(node.intervalID);
|
|
node.intervalID = setInterval(sendMsgFromBuffer, node.rate);
|
|
}
|
|
if (node.buffer.length < _maxKeptMsgsCount) {
|
|
var m = RED.util.cloneMessage(msg);
|
|
node.buffer.push({msg: m, send: send, done: done});
|
|
} else {
|
|
node.trace("dropped due to buffer overflow. msg._msgid = " + msg._msgid);
|
|
node.droppedMsgs++;
|
|
}
|
|
}
|
|
} else {
|
|
if (msg.hasOwnProperty("rate")) {
|
|
node.rate = msg.rate;
|
|
}
|
|
var timeSinceLast;
|
|
if (node.lastSent) {
|
|
timeSinceLast = process.hrtime(node.lastSent);
|
|
}
|
|
if (!node.lastSent) { // ensuring that we always send the first message
|
|
node.lastSent = process.hrtime();
|
|
send(msg);
|
|
}
|
|
else if ( ( (timeSinceLast[0] * SECONDS_TO_NANOS) + timeSinceLast[1] ) > (node.rate * MILLIS_TO_NANOS) ) {
|
|
node.lastSent = process.hrtime();
|
|
send(msg);
|
|
}
|
|
}
|
|
done();
|
|
}
|
|
});
|
|
node.on("close", function() {
|
|
clearInterval(node.intervalID);
|
|
clearTimeout(node.busy);
|
|
node.buffer.forEach((msgInfo) => msgInfo.done());
|
|
node.buffer = [];
|
|
node.status({});
|
|
});
|
|
}
|
|
else if ((node.pauseType === "queue") || (node.pauseType === "timed")) {
|
|
node.intervalID = setInterval(function() {
|
|
if (node.pauseType === "queue") {
|
|
if (node.buffer.length > 0) {
|
|
const msgInfo = node.buffer.shift();
|
|
msgInfo.send(msgInfo.msg); // send the first on the queue
|
|
msgInfo.done();
|
|
}
|
|
}
|
|
else {
|
|
while (node.buffer.length > 0) { // send the whole queue
|
|
const msgInfo = node.buffer.shift();
|
|
msgInfo.send(msgInfo.msg);
|
|
msgInfo.done();
|
|
}
|
|
}
|
|
node.reportDepth();
|
|
},node.rate);
|
|
|
|
var hit;
|
|
node.on("input", function(msg, send, done) {
|
|
if (!msg.hasOwnProperty("topic")) { msg.topic = "_none_"; }
|
|
hit = false;
|
|
for (var b in node.buffer) { // check if already in queue
|
|
if (msg.topic === node.buffer[b].msg.topic) {
|
|
node.buffer[b].done();
|
|
node.buffer[b] = {msg, send, done}; // if so - replace existing entry
|
|
hit = true;
|
|
break;
|
|
}
|
|
}
|
|
if (!hit) {
|
|
node.buffer.push({msg, send, done}); // if not add to end of queue
|
|
node.reportDepth();
|
|
}
|
|
if (msg.hasOwnProperty("reset")) {
|
|
while (node.buffer.length > 0) {
|
|
const msgInfo = node.buffer.shift();
|
|
msgInfo.done();
|
|
}
|
|
node.buffer = [];
|
|
node.status({text:"reset"});
|
|
done();
|
|
}
|
|
if (msg.hasOwnProperty("flush")) {
|
|
while (node.buffer.length > 0) {
|
|
const msgInfo = node.buffer.shift();
|
|
msgInfo.send(msgInfo.msg);
|
|
msgInfo.done();
|
|
}
|
|
node.status({});
|
|
done();
|
|
}
|
|
});
|
|
node.on("close", function() {
|
|
clearInterval(node.intervalID);
|
|
while (node.buffer.length > 0) {
|
|
const msgInfo = node.buffer.shift();
|
|
msgInfo.done();
|
|
}
|
|
node.buffer = [];
|
|
node.status({});
|
|
});
|
|
}
|
|
else if (node.pauseType === "random") {
|
|
node.on("input", function(msg, send, done) {
|
|
var wait = node.randomFirst + (node.diff * Math.random());
|
|
var id = ourTimeout(function() {
|
|
node.idList.splice(node.idList.indexOf(id),1);
|
|
send(msg);
|
|
node.status({});
|
|
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"});
|
|
}
|
|
});
|
|
node.on("close", function() { clearDelayList(); });
|
|
}
|
|
}
|
|
RED.nodes.registerType("delay",DelayNode);
|
|
}
|