mirror of
https://github.com/node-red/node-red.git
synced 2023-10-10 13:36:53 +02:00
Add support for Messaging API to delay node
This commit is contained in:
Kunihiko Toumura
parent
15a600c763
commit
dc1ab7e331
@ -80,10 +80,10 @@ module.exports = function(RED) {
|
||||
this.drop = n.drop;
|
||||
var node = this;
|
||||
|
||||
function ourTimeout(handler, delay) {
|
||||
function ourTimeout(handler, delay, clearHandler) {
|
||||
var toutID = setTimeout(handler, delay);
|
||||
return {
|
||||
clear: function() { clearTimeout(toutID); },
|
||||
clear: function() { clearTimeout(toutID); clearHandler(); },
|
||||
trigger: function() { clearTimeout(toutID); return handler(); }
|
||||
};
|
||||
}
|
||||
@ -113,14 +113,15 @@ module.exports = function(RED) {
|
||||
}
|
||||
|
||||
if (node.pauseType === "delay") {
|
||||
node.on("input", function(msg) {
|
||||
if (msg.hasOwnProperty("flush")) { flushDelayList(); }
|
||||
node.on("input", function(msg, send, done) {
|
||||
if (msg.hasOwnProperty("flush")) { flushDelayList(); done(); }
|
||||
else {
|
||||
var id = ourTimeout(function() {
|
||||
node.idList.splice(node.idList.indexOf(id),1);
|
||||
if (node.idList.length === 0) { node.status({}); }
|
||||
node.send(msg);
|
||||
}, node.timeout);
|
||||
send(msg);
|
||||
done();
|
||||
}, node.timeout, () => done());
|
||||
node.idList.push(id);
|
||||
if ((node.timeout > 1000) && (node.idList.length !== 0)) {
|
||||
node.status({fill:"blue",shape:"dot",text:" "});
|
||||
@ -131,7 +132,7 @@ module.exports = function(RED) {
|
||||
node.on("close", function() { clearDelayList(); });
|
||||
}
|
||||
else if (node.pauseType === "delayv") {
|
||||
node.on("input", function(msg) {
|
||||
node.on("input", function(msg, send, done) {
|
||||
var delayvar = Number(node.timeout);
|
||||
if (msg.hasOwnProperty("delay") && !isNaN(parseFloat(msg.delay))) {
|
||||
delayvar = parseFloat(msg.delay);
|
||||
@ -140,8 +141,9 @@ module.exports = function(RED) {
|
||||
var id = ourTimeout(function() {
|
||||
node.idList.splice(node.idList.indexOf(id),1);
|
||||
if (node.idList.length === 0) { node.status({}); }
|
||||
node.send(msg);
|
||||
}, delayvar);
|
||||
send(msg);
|
||||
done();
|
||||
}, delayvar, () => done());
|
||||
node.idList.push(id);
|
||||
if ((delayvar >= 0) && (node.idList.length !== 0)) {
|
||||
node.status({fill:"blue",shape:"dot",text:delayvar/1000+"s"});
|
||||
@ -152,7 +154,7 @@ module.exports = function(RED) {
|
||||
node.on("close", function() { clearDelayList(); });
|
||||
}
|
||||
else if (node.pauseType === "rate") {
|
||||
node.on("input", function(msg) {
|
||||
node.on("input", function(msg, send, done) {
|
||||
if (msg.hasOwnProperty("reset")) {
|
||||
if (node.intervalID !== -1 ) {
|
||||
clearInterval(node.intervalID);
|
||||
@ -161,17 +163,18 @@ module.exports = function(RED) {
|
||||
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) {
|
||||
node.buffer.push(m);
|
||||
node.buffer.push({msg: m, send: send, done: done});
|
||||
node.reportDepth();
|
||||
}
|
||||
else {
|
||||
node.send(m);
|
||||
send(m);
|
||||
node.reportDepth();
|
||||
node.intervalID = setInterval(function() {
|
||||
if (node.buffer.length === 0) {
|
||||
@ -179,16 +182,22 @@ module.exports = function(RED) {
|
||||
node.intervalID = -1;
|
||||
}
|
||||
if (node.buffer.length > 0) {
|
||||
node.send(node.buffer.shift());
|
||||
const msgInfo = node.buffer.shift();
|
||||
msgInfo.send(msgInfo.msg);
|
||||
msgInfo.done();
|
||||
}
|
||||
node.reportDepth();
|
||||
}, node.rate);
|
||||
done();
|
||||
}
|
||||
if (msg.hasOwnProperty("flush")) {
|
||||
while (node.buffer.length > 0) {
|
||||
node.send(node.buffer.shift());
|
||||
const msgInfo = node.buffer.shift();
|
||||
msgInfo.send(msgInfo.msg);
|
||||
msgInfo.done();
|
||||
}
|
||||
node.status({});
|
||||
done();
|
||||
}
|
||||
}
|
||||
else {
|
||||
@ -198,17 +207,19 @@ module.exports = function(RED) {
|
||||
}
|
||||
if (!node.lastSent) { // ensuring that we always send the first message
|
||||
node.lastSent = process.hrtime();
|
||||
node.send(msg);
|
||||
send(msg);
|
||||
}
|
||||
else if ( ( (timeSinceLast[0] * SECONDS_TO_NANOS) + timeSinceLast[1] ) > (node.rate * MILLIS_TO_NANOS) ) {
|
||||
node.lastSent = process.hrtime();
|
||||
node.send(msg);
|
||||
send(msg);
|
||||
}
|
||||
done();
|
||||
}
|
||||
});
|
||||
node.on("close", function() {
|
||||
clearInterval(node.intervalID);
|
||||
clearTimeout(node.busy);
|
||||
node.buffer.forEach((msgInfo) => msgInfo.done());
|
||||
node.buffer = [];
|
||||
node.status({});
|
||||
});
|
||||
@ -217,57 +228,75 @@ module.exports = function(RED) {
|
||||
node.intervalID = setInterval(function() {
|
||||
if (node.pauseType === "queue") {
|
||||
if (node.buffer.length > 0) {
|
||||
node.send(node.buffer.shift()); // send the first on the queue
|
||||
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
|
||||
node.send(node.buffer.shift());
|
||||
const msgInfo = node.buffer.shift();
|
||||
msgInfo.send(msgInfo.msg);
|
||||
msgInfo.done();
|
||||
}
|
||||
}
|
||||
node.reportDepth();
|
||||
},node.rate);
|
||||
|
||||
var hit;
|
||||
node.on("input", function(msg) {
|
||||
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].topic) {
|
||||
node.buffer[b] = msg; // if so - replace existing entry
|
||||
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); // if not add to end of queue
|
||||
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) {
|
||||
node.send(node.buffer.shift());
|
||||
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) {
|
||||
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);
|
||||
node.send(msg);
|
||||
send(msg);
|
||||
node.status({});
|
||||
}, wait);
|
||||
done();
|
||||
}, wait, () => done());
|
||||
node.idList.push(id);
|
||||
if ((node.timeout >= 1000) && (node.idList.length !== 0)) {
|
||||
node.status({fill:"blue",shape:"dot",text:parseInt(wait/10)/100+"s"});
|
||||
|
||||
@ -719,4 +719,95 @@ describe('delay Node', function() {
|
||||
setImmediate( function() { delayNode1.receive({reset:true}); }); // reset the queue
|
||||
});
|
||||
});
|
||||
|
||||
/* Messaging API support */
|
||||
function mapiDoneTestHelper(done, pauseType, drop, msgAndTimings) {
|
||||
const completeNode = require("nr-test-utils").require("@node-red/nodes/core/common/24-complete.js");
|
||||
const flow = [{id:"delayNode1",type:"delay",name:"delayNode", pauseType:pauseType, timeout:"1", timeoutUnits: "seconds",
|
||||
rate: "1", nbRateUnits: "1", rateUnits: "second", randomFirst:"950", randomLast:"1050",randomUnits:"milliseconds",
|
||||
drop: drop, wires: [[]]},
|
||||
{id:"completeNode1",type:"complete",scope: ["delayNode1"],uncaught:false,wires:[["helperNode1"]]},
|
||||
{id:"helperNode1",type:"helper", wires:[[]]}];
|
||||
const numMsgs = msgAndTimings.length;
|
||||
helper.load([delayNode, completeNode], flow, function () {
|
||||
const delayNode1 = helper.getNode("delayNode1");
|
||||
const helperNode1 = helper.getNode("helperNode1");
|
||||
const t = Date.now();
|
||||
let c = 0;
|
||||
helperNode1.on("input", function (msg) {
|
||||
msg.should.have.a.property('payload', msgAndTimings[c].msg.payload);
|
||||
(Date.now() - t).should.be.approximately(msgAndTimings[c].avr, msgAndTimings[c].var);
|
||||
c += 1;
|
||||
if ( c === numMsgs) {
|
||||
done();
|
||||
}
|
||||
});
|
||||
for (let i = 0; i < numMsgs; i++) {
|
||||
setImmediate( function() { delayNode1.receive(msgAndTimings[i].msg); } );
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
it('calls done when queued message is emitted (type: delay)', function(done) {
|
||||
mapiDoneTestHelper(done, "delay", false, [{msg:{payload:1}, avr:1000, var:100}]);
|
||||
});
|
||||
it('calls done when queued message is emitted (type: delayv)', function(done) {
|
||||
mapiDoneTestHelper(done, "delayv", false, [{msg:{payload:1, delay:1000}, avr:1000, var:100}]);
|
||||
});
|
||||
it('calls done when queued message is emitted (type: delay)', function(done) {
|
||||
mapiDoneTestHelper(done, "random", false, [{msg:{payload:1}, avr:1000, var:100}]);
|
||||
});
|
||||
it('calls done when queued message is cleared (type: delay)', function(done) {
|
||||
mapiDoneTestHelper(done, "delay", false, [{msg:{payload:1}, avr:100, var:100},
|
||||
{msg:{payload:2,reset:true}, avr:100, var:100}]);
|
||||
});
|
||||
it('calls done when queued message is cleared (type: delayv)', function(done) {
|
||||
mapiDoneTestHelper(done, "delayv", false, [{msg:{payload:1, delay:1000}, avr:100, var:100},
|
||||
{msg:{payload:2, reset:true}, avr:100, var:100}]);
|
||||
});
|
||||
it('calls done when queued message is cleared (type: random)', function(done) {
|
||||
mapiDoneTestHelper(done, "random", false, [{msg:{payload:1}, avr:100, var:100},
|
||||
{msg:{payload:2,reset:true}, avr:100, var:100}]);
|
||||
});
|
||||
it('calls done when queued message is flushed (type: delay)', function(done) {
|
||||
mapiDoneTestHelper(done, "delay", false, [{msg:{payload:1}, avr:100, var:100},
|
||||
{msg:{payload:2,flush:true}, avr:100, var:100}]);
|
||||
});
|
||||
it('calls done when queued message is flushed (type: delayv)', function(done) {
|
||||
mapiDoneTestHelper(done, "delayv", false, [{msg:{payload:1, delay:1000}, avr:100, var:100},
|
||||
{msg:{payload:2, flush:true}, avr:100, var:100}]);
|
||||
});
|
||||
it('calls done when queued message is flushed (type: random)', function(done) {
|
||||
mapiDoneTestHelper(done, "random", false, [{msg:{payload:1}, avr:100, var:100},
|
||||
{msg:{payload:2,flush:true}, avr:100, var:100}]);
|
||||
});
|
||||
it('calls done when rated message is emitted (drop: false)', function(done) {
|
||||
mapiDoneTestHelper(done, "rate", false, [{msg:{payload:1}, avr:0, var:100},
|
||||
{msg:{payload:2}, avr:1000, var:100}]);
|
||||
});
|
||||
it('calls done when rated message is emitted (drop: true)', function(done) {
|
||||
mapiDoneTestHelper(done, "rate", true, [{msg:{payload:1}, avr:0, var:100},
|
||||
{msg:{payload:2}, avr:0, var:100}]);
|
||||
});
|
||||
it('calls done when rated message is flushed', function(done) {
|
||||
mapiDoneTestHelper(done, "rate", false, [{msg:{payload:1}, avr:0, var:100},
|
||||
{msg:{payload:2}, avr:0, var:100},
|
||||
{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}]);
|
||||
});
|
||||
it('calls done when queued messages are sent (timed)', function(done) {
|
||||
mapiDoneTestHelper(done, "timed", false, [{msg:{payload:1,topic:"a"}, avr:500, var:700},
|
||||
{msg:{payload:2,topic:"b"}, avr:500, var:700}]);
|
||||
});
|
||||
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}]);
|
||||
});
|
||||
});
|
||||
|
||||
Loading…
Reference in New Issue
Block a user