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

Add "topic based fair queue" option to delay node

Browse Source
This commit is contained in:
Dave C-J
2014-10-24 20:00:08 +01:00
parent cf81de415a
commit b54e9edfa6
2 changed files with 61 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
nodes/core/core/89-delay.js
View File

@@ -17,10 +17,10 @@
//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;
function random(n) {
var wait = n.randomFirst + (n.diff * Math.random());
if (n.buffer.length > 0) {
@@ -136,7 +136,7 @@ module.exports = function(RED) {
if (node.lastSent) {
timeSinceLast = process.hrtime(node.lastSent);
}
if (!node.lastSent) { // ensuring that we always send the first message
if (!node.lastSent) { // ensuring that we always send the first message
node.lastSent = process.hrtime();
node.send(msg);
} else if ( ( (timeSinceLast[0] * SECONDS_TO_NANOS) + timeSinceLast[1] ) > (node.rate * MILLIS_TO_NANOS) ) {
@@ -151,6 +151,34 @@ module.exports = function(RED) {
this.buffer = [];
});
} else if (this.pauseType === "queue") {
this.intervalID = setInterval(function() {
if (node.buffer.length > 0) {
node.send(node.buffer.shift()); // send the first on the queue
}
node.status({text:node.buffer.length});
//console.log(node.buffer);
},node.rate);
this.on("input", function(msg) {
if (!msg.hasOwnProperty("topic")) { msg.topic = "_none_"; }
var hit = false;
for (var b in node.buffer) { // check if already in queue
if (msg.topic === node.buffer[b].topic) {
node.buffer[b] = msg; // if so - replace existing entry
hit = true;
}
}
if (!hit) { node.buffer.push(msg); } // if not add to end of queue
node.status({text:node.buffer.length});
});
this.on("close", function() {
clearInterval(this.intervalID);
this.buffer = [];
node.status({text:node.buffer.length});
});
} else if (this.pauseType === "random") {
this.on("input",function(msg){
node.buffer.push(msg);