Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add reset feature to batch node #2553

Merged
merged 2 commits into from May 11, 2020
Merged

Add reset feature to batch node #2553

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8750c4b
add reset feature to batch node
HiroyasuNishiyama May 5, 2020
ee13cd1
add checks for interval value
HiroyasuNishiyama May 6, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
33 changes: 30 additions & 3 deletions packages/node_modules/@node-red/nodes/core/sequence/19-batch.js
View file
Open in desktop
Expand Up @@ -179,6 +179,11 @@ module.exports = function(RED) {
}
node.pending = [];
this.on("input", function(msg) {
if (msg.hasOwnProperty("reset")) {
node.pending = [];
node.pending_count = 0;
return;
}
var queue = node.pending;
queue.push(msg);
node.pending_count++;
Expand All @@ -204,11 +209,26 @@ module.exports = function(RED) {
var interval = Number(n.interval || "0") *1000;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I realise this isn't related to the PR, but I noticed this line. If n.interval is set to 0 (or undefined) then this will setup an interval of 0ms - which sounds like a bad thing and unlikely to be something a user would actually want.

Should there be a safety check for a minimum interval value and default to a non-0 value to prevent accidentally performing a denial of service of the event loop?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added the check of the value of interval.
As additional information, current implementation limits the input value of the interval to be greater than 1, as shown below.

interval: {value:10,validate:function(v) { return RED.validators.number(v) && (v >= 1); }},

var allow_empty_seq = n.allowEmptySequence;
node.pending = []
var timer = setInterval(function() {
function msgHandler() {
send_interval(node, allow_empty_seq);
node.pending_count = 0;
}, interval);
}
var timer = undefined;
if (interval > 0) {
timer = setInterval(msgHandler, interval);
}
this.on("input", function(msg) {
if (msg.hasOwnProperty("reset")) {
if (timer !== undefined) {
clearInterval(timer);
}
node.pending = [];
node.pending_count = 0;
if (interval > 0) {
timer = setInterval(msgHandler, interval);
}
return;
}
node.pending.push(msg);
node.pending_count++;
var max_msgs = max_kept_msgs_count(node);
Expand All @@ -219,7 +239,9 @@ module.exports = function(RED) {
}
});
this.on("close", function() {
clearInterval(timer);
if (timer !== undefined) {
clearInterval(timer);
}
node.pending = [];
node.pending_count = 0;
});
Expand All @@ -230,6 +252,11 @@ module.exports = function(RED) {
});
node.pending = {};
this.on("input", function(msg) {
if (msg.hasOwnProperty("reset")) {
node.pending = {};
node.pending_count = 0;
return;
}
concat_msg(node, msg);
});
this.on("close", function() {
Expand Down
1 change: 1 addition & 0 deletions packages/node_modules/@node-red/nodes/locales/en-US/sequence/19-batch.html
View file
Open in desktop
Expand Up @@ -39,4 +39,5 @@ <h4>Storing messages</h4>
<p>This node will buffer messages internally in order to work across sequences. The
runtime setting <code>nodeMessageBufferMaxLength</code> can be used to limit how many messages nodes
will buffer.</p>
<p>If a message is received with the <b>msg.reset</b> property set, the buffered messages are deleted and not sent.</p>
</script>
2 changes: 2 additions & 0 deletions packages/node_modules/@node-red/nodes/locales/ja/sequence/19-batch.html
View file
Open in desktop
Expand Up @@ -31,4 +31,6 @@ <h3>詳細</h3>
</dl>
<h4>メッセージの蓄積</h4>
<p>このノードの処理ではメッセージ列の処理のためメッセージを内部に蓄積します。<b>settings.js</b>の<code>nodeMessageBufferMaxLength</code>を指定することで蓄積するメッセージの最大値を制限することができます。</p>
<p>メッセージが<b>msg.reset</b>プロパティを持つ場合、蓄積したメッセージを削除し送信を行いません。</p>

</script>
102 changes: 98 additions & 4 deletions test/nodes/core/sequence/19-batch_spec.js
View file
Open in desktop
Expand Up @@ -107,13 +107,18 @@ describe('BATCH node', function() {
}
}

function delayed_send(receiver, index, count, delay) {
function delayed_send(receiver, index, count, delay, done) {
if (index < count) {
setTimeout(function() {
receiver.receive({payload: index});
delayed_send(receiver, index+1, count, delay);
delayed_send(receiver, index+1, count, delay, done);
}, delay);
}
else if(index === count) {
if (done) {
done();
}
}
}

function check_interval(flow, results, delay, done) {
Expand Down Expand Up @@ -198,10 +203,28 @@ describe('BATCH node', function() {
});
});

it('should handle reset', function(done) {
var flow = [{id:"n1", type:"batch", name: "BatchNode", mode: "count", count: 2, overlap: 0, interval: 0, allowEmptySequence: false, topics: [], wires:[["n2"]]},
{id:"n2", type:"helper"}];
helper.load(batchNode, flow, function() {
var n1 = helper.getNode("n1");
var n2 = helper.getNode("n2");
var results = [
[0, 1],
[4, 5]
];
check_data(n1, n2, results, done);
n1.receive({payload:0});
n1.receive({payload:1});
n1.receive({payload:2});
n1.receive({payload:3, reset: true});
n1.receive({payload:4});
n1.receive({payload:5});
});
});
});

describe('mode: interval', function() {

it('should create seq. with interval', function(done) {
var flow = [{id:"n1", type:"batch", name: "BatchNode", mode: "interval", count: 0, overlap: 0, interval: 1, allowEmptySequence: false, topics: [], wires:[["n2"]]},
{id:"n2", type:"helper"}];
Expand Down Expand Up @@ -265,10 +288,29 @@ describe('BATCH node', function() {
});
});

it('should handle reset', function(done) {
var flow = [{id:"n1", type:"batch", name: "BatchNode", mode: "interval", count: 0, overlap: 0, interval: 1, allowEmptySequence: false, topics: [], wires:[["n2"]]},
{id:"n2", type:"helper"}];
helper.load(batchNode, flow, function() {
var n1 = helper.getNode("n1");
var n2 = helper.getNode("n2");
var results = [
[0, 1],
[4, 5]
];
check_data(n1, n2, results, done);
delayed_send(n1, 0, 3, 400, function () {
setTimeout(function () {
n1.receive({payload: "3", reset: true});
delayed_send(n1, 4, 7, 400);
}, 10);
});
});
});

});

describe('mode: concat', function() {

it('should concat two seq. (series)', function(done) {
var flow = [{id:"n1", type:"batch", name: "BatchNode", mode: "concat", count: 0, overlap: 0, interval: 1, allowEmptySequence: false, topics: [{topic: "TA"}, {topic: "TB"}], wires:[["n2"]]},
{id:"n2", type:"helper"}];
Expand Down Expand Up @@ -355,6 +397,58 @@ describe('BATCH node', function() {
});
});

it('should handle reset', function(done) {
var flow = [{id:"n1", type:"batch", name: "BatchNode", mode: "concat", count: 0, overlap: 0, interval: 1, allowEmptySequence: false, topics: [{topic: "TA"}, {topic: "TB"}], wires:[["n2"]]},
{id:"n2", type:"helper"}];
try {
helper.load(batchNode, flow, function() {
var n1 = helper.getNode("n1");
var n2 = helper.getNode("n2");
var results = [
[2, 3, 0, 1]
];
check_data(n1, n2, results, done);
var inputs0 = [
["TB", 0, 0, 2],
["TA", 1, 0, 2],
];
for(var data of inputs0) {
var msg = {
topic: data[0],
payload: data[1],
parts: {
id: data[0],
index: data[2],
count: data[3]
}
};
n1.receive(msg);
}
n1.receive({payload: undefined, reset: true});
var inputs1 = [
["TB", 0, 0, 2],
["TB", 1, 1, 2],
["TA", 2, 0, 2],
["TA", 3, 1, 2]
];
for(var data of inputs1) {
var msg = {
topic: data[0],
payload: data[1],
parts: {
id: data[0],
index: data[2],
count: data[3]
}
};
n1.receive(msg);
}
});
}
catch (e) {
done(e);
}
});
});

});