frodovdr/node-red
1
0
Fork 0
mirror of https://github.com/node-red/node-red.git synced 2023-10-10 13:36:53 +02:00
Code Issues Releases Wiki Activity

Add nodeDone to File nodes

Browse Source
This commit is contained in:
Nick O'Leary 2019-08-14 22:28:10 +01:00
parent 84232f25f0
commit 4ce0e39760
No known key found for this signature in database
GPG Key ID: 4F2157149161A6C9
1 changed files with 31 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
packages/node_modules/@node-red/nodes/core/storage/50-file.js vendored
View File

@ -36,6 +36,7 @@ module.exports = function(RED) {
} }
function FileNode(n) { function FileNode(n) {
// Write/delete a file
RED.nodes.createNode(this,n); RED.nodes.createNode(this,n);
this.filename = n.filename; this.filename = n.filename;
this.appendNewline = n.appendNewline; this.appendNewline = n.appendNewline;
@ -48,7 +49,7 @@ module.exports = function(RED) {
node.closing = false; node.closing = false;
node.closeCallback = null; node.closeCallback = null;
function processMsg(msg, done) { function processMsg(msg,nodeSend, done) {
var filename = node.filename || msg.filename || ""; var filename = node.filename || msg.filename || "";
if ((!node.filename) && (!node.tout)) { if ((!node.filename) && (!node.tout)) {
node.tout = setTimeout(function() { node.tout = setTimeout(function() {
@ -68,7 +69,7 @@ module.exports = function(RED) {
if (RED.settings.verbose) { if (RED.settings.verbose) {
node.log(RED._("file.status.deletedfile",{file:filename})); node.log(RED._("file.status.deletedfile",{file:filename}));
} }
node.send(msg); nodeSend(msg);
} }
done(); done();
}); });
@ -101,7 +102,7 @@ module.exports = function(RED) {
}); });
wstream.on("open", function() { wstream.on("open", function() {
wstream.end(buf, function() { wstream.end(buf, function() {
node.send(msg); nodeSend(msg);
done(); done();
}); });
}) })
@ -150,13 +151,13 @@ module.exports = function(RED) {
if (node.filename) { if (node.filename) {
// Static filename - write and reuse the stream next time // Static filename - write and reuse the stream next time
node.wstream.write(buf, function() { node.wstream.write(buf, function() {
node.send(msg); nodeSend(msg);
done(); done();
}); });
} else { } else {
// Dynamic filename - write and close the stream // Dynamic filename - write and close the stream
node.wstream.end(buf, function() { node.wstream.end(buf, function() {
node.send(msg); nodeSend(msg);
delete node.wstream; delete node.wstream;
delete node.wstreamIno; delete node.wstreamIno;
done(); done();
@ -169,12 +170,13 @@ module.exports = function(RED) {
} }
} }
function processQ(queue) { function processQueue(queue) {
var msg = queue[0]; var event = queue[0];
processMsg(msg, function() { processMsg(event.msg, event.send, function() {
event.done();
queue.shift(); queue.shift();
if (queue.length > 0) { if (queue.length > 0) {
processQ(queue); processQueue(queue);
} }
else if (node.closing) { else if (node.closing) {
closeNode(); closeNode();
@ -182,14 +184,19 @@ module.exports = function(RED) {
}); });
} }
this.on("input", function(msg) { this.on("input", function(msg,nodeSend,nodeDone) {
var msgQueue = node.msgQueue; var msgQueue = node.msgQueue;
if (msgQueue.push(msg) > 1) { msgQueue.push({
msg: msg,
send: nodeSend,
done: nodeDone
})
if (msgQueue.length > 1) {
// pending write exists // pending write exists
return; return;
} }
try { try {
processQ(msgQueue); processQueue(msgQueue);
} }
catch (e) { catch (e) {
node.msgQueue = []; node.msgQueue = [];
@ -234,6 +241,7 @@ module.exports = function(RED) {
function FileInNode(n) { function FileInNode(n) {
// Read a file
RED.nodes.createNode(this,n); RED.nodes.createNode(this,n);
this.filename = n.filename; this.filename = n.filename;
this.format = n.format; this.format = n.format;
@ -248,13 +256,14 @@ module.exports = function(RED) {
if (this.format === "stream") { this.chunk = true; } if (this.format === "stream") { this.chunk = true; }
var node = this; var node = this;
this.on("input",function(msg) { this.on("input",function(msg, nodeSend, nodeDone) {
var filename = (node.filename || msg.filename || "").replace(/\t|\r|\n/g,''); var filename = (node.filename || msg.filename || "").replace(/\t|\r|\n/g,'');
if (!node.filename) { if (!node.filename) {
node.status({fill:"grey",shape:"dot",text:filename}); node.status({fill:"grey",shape:"dot",text:filename});
} }
if (filename === "") { if (filename === "") {
node.warn(RED._("file.errors.nofilename")); node.warn(RED._("file.errors.nofilename"));
nodeDone();
} }
else { else {
msg.filename = filename; msg.filename = filename;
@ -288,7 +297,7 @@ module.exports = function(RED) {
parts:{index:count, ch:ch, type:type, id:msg._msgid} parts:{index:count, ch:ch, type:type, id:msg._msgid}
} }
count += 1; count += 1;
node.send(m); nodeSend(m);
} }
spare = bits[i]; spare = bits[i];
} }
@ -304,7 +313,7 @@ module.exports = function(RED) {
getout = false; getout = false;
m.parts.count = count; m.parts.count = count;
} }
node.send(m); nodeSend(m);
} }
} }
else { else {
@ -318,8 +327,9 @@ module.exports = function(RED) {
var sendMessage = RED.util.cloneMessage(msg); var sendMessage = RED.util.cloneMessage(msg);
delete sendMessage.payload; delete sendMessage.payload;
sendMessage.error = err; sendMessage.error = err;
node.send(sendMessage); nodeSend(sendMessage);
} }
nodeDone();
}) })
.on('end', function() { .on('end', function() {
if (node.chunk === false) { if (node.chunk === false) {
@ -327,7 +337,7 @@ module.exports = function(RED) {
msg.payload = decode(lines, node.encoding); msg.payload = decode(lines, node.encoding);
} }
else { msg.payload = lines; } else { msg.payload = lines; }
node.send(msg); nodeSend(msg);
} }
else if (node.format === "lines") { else if (node.format === "lines") {
var m = { payload: spare, var m = { payload: spare,
@ -339,12 +349,13 @@ module.exports = function(RED) {
id: msg._msgid id: msg._msgid
} }
}; };
node.send(m); nodeSend(m);
} }
else if (getout) { // last chunk same size as high water mark - have to send empty extra packet. else if (getout) { // last chunk same size as high water mark - have to send empty extra packet.
var m = { parts:{index:count, count:count, ch:ch, type:type, id:msg._msgid} }; var m = { parts:{index:count, count:count, ch:ch, type:type, id:msg._msgid} };
node.send(m); nodeSend(m);
} }
nodeDone();
}); });
} }
}); });