From dc1ab7e3319cc9fc747c62caeb2d7fbde6031c52 Mon Sep 17 00:00:00 2001 From: Kunihiko Toumura Date: Mon, 26 Oct 2020 16:52:18 +0900 Subject: [PATCH] Add support for Messaging API to delay node --- .../@node-red/nodes/core/function/89-delay.js | 81 +++++++++++------ test/nodes/core/function/89-delay_spec.js | 91 +++++++++++++++++++ 2 files changed, 146 insertions(+), 26 deletions(-) diff --git a/packages/node_modules/@node-red/nodes/core/function/89-delay.js b/packages/node_modules/@node-red/nodes/core/function/89-delay.js index b700913cc..b9f76ef9e 100644 --- a/packages/node_modules/@node-red/nodes/core/function/89-delay.js +++ b/packages/node_modules/@node-red/nodes/core/function/89-delay.js @@ -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"}); diff --git a/test/nodes/core/function/89-delay_spec.js b/test/nodes/core/function/89-delay_spec.js index 51583d8a4..6a5a3c25f 100644 --- a/test/nodes/core/function/89-delay_spec.js +++ b/test/nodes/core/function/89-delay_spec.js @@ -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}]); + }); });