mirror of
https://github.com/node-red/node-red.git
synced 2023-10-10 13:36:53 +02:00
Delay node updates to added flush control for queuing
This commit is contained in:
parent
b2f5a259ab
commit
41c8ca8ab4
@ -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":"";
|
||||
|
@ -112,31 +112,35 @@ module.exports = function(RED) {
|
||||
}
|
||||
if (node.buffer.length > 0) {
|
||||
const msgInfo = node.buffer.shift();
|
||||
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,6 +217,7 @@ module.exports = function(RED) {
|
||||
if (!node.drop) {
|
||||
var m = RED.util.cloneMessage(msg);
|
||||
delete m.flush;
|
||||
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;
|
||||
@ -232,13 +242,23 @@ module.exports = function(RED) {
|
||||
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);
|
||||
if (Object.keys(msgInfo.msg).length > 1) {
|
||||
node.send(msgInfo.msg);
|
||||
msgInfo.done();
|
||||
}
|
||||
node.status({});
|
||||
len = len - 1;
|
||||
}
|
||||
if (node.buffer.length === 0) {
|
||||
clearInterval(node.intervalID);
|
||||
node.intervalID = -1;
|
||||
}
|
||||
node.status({fill:"blue",shape:"dot",text:node.buffer.length});
|
||||
done();
|
||||
}
|
||||
}
|
||||
@ -340,11 +360,17 @@ 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);
|
||||
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(); });
|
||||
|
@ -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>
|
||||
|
@ -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},
|
||||
|
Loading…
Reference in New Issue
Block a user