The filename should be an absolute path, otherwise it will be relative to the working directory of the Node-RED process.
On Windows, path separators may need to be escaped, for example: \\Users\\myUser.
Optionally, a text file can be split into lines, outputting one message per line, or a binary file + file into smaller buffer chunks, the chunk size is operating system dependant, but typically 64k (linux/mac) or 41k (Windows).
diff --git a/nodes/core/storage/50-file.js b/nodes/core/storage/50-file.js index 926965c63..10d5b10a6 100644 --- a/nodes/core/storage/50-file.js +++ b/nodes/core/storage/50-file.js @@ -18,6 +18,7 @@ module.exports = function(RED) { "use strict"; var fs = require("fs-extra"); var os = require("os"); + var path = require("path"); function FileNode(n) { RED.nodes.createNode(this,n); @@ -26,15 +27,27 @@ module.exports = function(RED) { this.overwriteFile = n.overwriteFile.toString(); this.createDir = n.createDir || false; var node = this; + node.wstream = null; + node.data = []; this.on("input",function(msg) { var filename = node.filename || msg.filename || ""; - if (!node.filename) { - node.status({fill:"grey",shape:"dot",text:filename}); + if (!node.filename) { node.status({fill:"grey",shape:"dot",text:filename}); } + if (filename === "") { node.warn(RED._("file.errors.nofilename")); } + else if (node.overwriteFile === "delete") { + fs.unlink(filename, function (err) { + if (err) { node.error(RED._("file.errors.deletefail",{error:err.toString()}),msg); } + else if (RED.settings.verbose) { node.log(RED._("file.status.deletedfile",{file:filename})); } + }); } - if (filename === "") { - node.warn(RED._("file.errors.nofilename")); - } else if (msg.hasOwnProperty("payload") && (typeof msg.payload !== "undefined")) { + else if (msg.hasOwnProperty("payload") && (typeof msg.payload !== "undefined")) { + var dir = path.dirname(filename); + if (node.createDir) { + fs.ensureDir(dir, function(err) { + if (err) { node.error(RED._("file.errors.createfail",{error:err.toString()}),msg); } + }); + } + var data = msg.payload; if ((typeof data === "object") && (!Buffer.isBuffer(data))) { data = JSON.stringify(data); @@ -42,52 +55,26 @@ module.exports = function(RED) { if (typeof data === "boolean") { data = data.toString(); } if (typeof data === "number") { data = data.toString(); } if ((this.appendNewline) && (!Buffer.isBuffer(data))) { data += os.EOL; } - data = new Buffer(data); - if (this.overwriteFile === "true") { - // using "binary" not {encoding:"binary"} to be 0.8 compatible for a while - //fs.writeFile(filename, data, "binary", function (err) { - fs.writeFile(filename, data, {encoding:"binary"}, function (err) { - if (err) { - if ((err.code === "ENOENT") && node.createDir) { - fs.ensureFile(filename, function (err) { - if (err) { node.error(RED._("file.errors.createfail",{error:err.toString()}),msg); } - else { - fs.writeFile(filename, data, "binary", function (err) { - if (err) { node.error(RED._("file.errors.writefail",{error:err.toString()}),msg); } - }); - } - }); - } - else { node.error(RED._("file.errors.writefail",{error:err.toString()}),msg); } + node.data.push(new Buffer(data)); + + while (node.data.length > 0) { + if (this.overwriteFile === "true") { + if (!node.wstream) { + node.wstream = fs.createWriteStream(filename, { encoding:'binary', flags:'w' }); + node.wstream.on("error", function(err) { + node.error(RED._("file.errors.writefail",{error:err.toString()}),msg); + }); } - else if (RED.settings.verbose) { node.log(RED._("file.status.wrotefile",{file:filename})); } - }); - } - else if (this.overwriteFile === "delete") { - fs.unlink(filename, function (err) { - if (err) { node.error(RED._("file.errors.deletefail",{error:err.toString()}),msg); } - else if (RED.settings.verbose) { node.log(RED._("file.status.deletedfile",{file:filename})); } - }); - } - else { - // using "binary" not {encoding:"binary"} to be 0.8 compatible for a while longer - //fs.appendFile(filename, data, "binary", function (err) { - fs.appendFile(filename, data, {encoding:"binary"}, function (err) { - if (err) { - if ((err.code === "ENOENT") && node.createDir) { - fs.ensureFile(filename, function (err) { - if (err) { node.error(RED._("file.errors.createfail",{error:err.toString()}),msg); } - else { - fs.appendFile(filename, data, "binary", function (err) { - if (err) { node.error(RED._("file.errors.appendfail",{error:err.toString()}),msg); } - }); - } - }); - } - else { node.error(RED._("file.errors.appendfail",{error:err.toString()}),msg); } + } + else { + if (!node.wstream) { + node.wstream = fs.createWriteStream(filename, { encoding:'binary', flags:'a' }); + node.wstream.on("error", function(err) { + node.error(RED._("file.errors.appendfail",{error:err.toString()}),msg); + }); } - else if (RED.settings.verbose) { node.log(RED._("file.status.appendedfile",{file:filename})); } - }); + } + node.wstream.write(node.data.shift()); } } }); @@ -100,14 +87,13 @@ module.exports = function(RED) { function FileInNode(n) { RED.nodes.createNode(this,n); - this.filename = n.filename; this.format = n.format; + this.chunk = false; + if (this.format === "lines") { this.chunk = true; } + if (this.format === "stream") { this.chunk = true; } var node = this; - var options = {}; - if (this.format) { - options['encoding'] = this.format; - } + this.on("input",function(msg) { var filename = node.filename || msg.filename || ""; if (!node.filename) { @@ -115,19 +101,82 @@ module.exports = function(RED) { } if (filename === "") { node.warn(RED._("file.errors.nofilename")); - } else { + } + else { msg.filename = filename; - fs.readFile(filename,options,function(err,data) { - if (err) { - node.error(err,msg); - msg.error = err; - delete msg.payload; - } else { - msg.payload = data; - delete msg.error; - } - node.send(msg); - }); + var lines = new Buffer.from([]); + var spare = ""; + var count = 0; + var type = "buffer"; + var ch = ""; + if (node.format === "lines") { + ch = "\n"; + type = "string"; + } + var hwm; + var getout = false; + + var rs = fs.createReadStream(filename) + .on('readable', function () { + var chunk; + var hwm = rs._readableState.highWaterMark; + while (null !== (chunk = rs.read())) { + if (node.chunk === true) { + getout = true; + if (node.format === "lines") { + spare += chunk.toString(); + var bits = spare.split("\n"); + for (var i=0; i < bits.length - 1; i++) { + var m = { + payload:bits[i], + topic:msg.topic, + filename:msg.filename, + parts:{index:count, ch:ch, type:type, id:msg._msgid} + } + count += 1; + if ((chunk.length < hwm) && (bits[i+1].length === 0)) { + m.parts.count = count; + } + node.send(m); + } + spare = bits[i]; + if (chunk.length !== hwm) { getout = false; } + //console.log("LEFT",bits[i].length,bits[i]); + } + if (node.format === "stream") { + var m = { + payload:chunk, + topic:msg.topic, + filename:msg.filename, + parts:{index:count, ch:ch, type:type, id:msg._msgid} + } + count += 1; + if (chunk.length < hwm) { // last chunk is smaller that high water mark = eof + getout = false; + m.parts.count = count; + } + node.send(m); + } + } + else { + lines = Buffer.concat([lines,chunk]); + } + } + }) + .on('error', function(err) { + node.error('Error while reading file.', msg); + }) + .on('end', function() { + if (node.chunk === false) { + if (node.format === "utf8") { msg.payload = lines.toString(); } + else { msg.payload = lines; } + node.send(msg); + } + 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} }; + node.send(m); + } + }); } }); this.on('close', function() { diff --git a/test/nodes/core/storage/50-file_spec.js b/test/nodes/core/storage/50-file_spec.js index 543a4aaf7..ccc62d734 100644 --- a/test/nodes/core/storage/50-file_spec.js +++ b/test/nodes/core/storage/50-file_spec.js @@ -183,8 +183,11 @@ describe('file Nodes', function() { it('should fail to write to a ro file', function(done) { // Stub file write so we can make writes fail - var spy = sinon.stub(fs, 'writeFile', function(arg1,arg2,arg3,arg4) { - arg4(new Error("Stub error message")); + var spy = sinon.stub(fs, 'createWriteStream', function(arg1,arg2) { + var ws = {}; + ws.on = function(e,d) { throw("Stub error message"); } + ws.write = function(e,d) { } + return ws; }); var flow = [{id:"fileNode1", type:"file", name: "fileNode", "filename":fileToTest, "appendNewline":false, "overwriteFile":true}]; @@ -198,11 +201,11 @@ describe('file Nodes', function() { //console.log(logEvents); logEvents.should.have.length(1); logEvents[0][0].should.have.a.property('msg'); - logEvents[0][0].msg.toString().should.startWith("file.errors.writefail"); + logEvents[0][0].msg.toString().should.startWith("Stub error message"); done(); } catch(e) { done(e); } - finally { fs.writeFile.restore(); } + finally { fs.createWriteStream.restore(); } },wait); n1.receive({payload:"test"}); }); @@ -210,7 +213,12 @@ describe('file Nodes', function() { it('should fail to append to a ro file', function(done) { // Stub file write so we can make writes fail - var spy = sinon.stub(fs, 'appendFile', function(arg,arg2,arg3,arg4) { arg4(new Error("Stub error message")); }); + var spy = sinon.stub(fs, 'createWriteStream', function(arg1,arg2) { + var ws = {}; + ws.on = function(e,d) { throw("Stub error message"); } + ws.write = function(e,d) { } + return ws; + }); var flow = [{id:"fileNode1", type:"file", name: "fileNode", "filename":fileToTest, "appendNewline":true, "overwriteFile":false}]; helper.load(fileNode, flow, function() { @@ -223,11 +231,11 @@ describe('file Nodes', function() { //console.log(logEvents); logEvents.should.have.length(1); logEvents[0][0].should.have.a.property('msg'); - logEvents[0][0].msg.toString().should.startWith("file.errors.appendfail"); + logEvents[0][0].msg.toString().should.startWith("Stub error message"); done(); } catch(e) { done(e); } - finally { fs.appendFile.restore(); } + finally { fs.createWriteStream.restore(); } },wait); n1.receive({payload:"test2"}); }); @@ -287,7 +295,7 @@ describe('file Nodes', function() { it('should try to create a new directory if asked to do so (append)', function(done) { // Stub file write so we can make writes fail var fileToTest2 = path.join(resourcesDir,"a","50-file-test-file.txt"); - var spy = sinon.stub(fs, "ensureFile", function(arg1,arg2,arg3,arg4) { arg2(null); }); + var spy = sinon.stub(fs, "ensureDir", function(arg1,arg2,arg3,arg4) { arg2(null); }); var flow = [{id:"fileNode1", type:"file", name: "fileNode", "filename":fileToTest2, "appendNewline":true, "overwriteFile":false, "createDir":true}]; helper.load(fileNode, flow, function() { var n1 = helper.getNode("fileNode1"); @@ -303,7 +311,7 @@ describe('file Nodes', function() { done(); } catch(e) { done(e); } - finally { fs.ensureFile.restore(); } + finally { fs.ensureDir.restore(); } },wait); n1.receive({payload:"test2"}); }); @@ -338,7 +346,7 @@ describe('file Nodes', function() { it('should try to create a new directory if asked to do so (overwrite)', function(done) { // Stub file write so we can make writes fail var fileToTest2 = path.join(resourcesDir,"a","50-file-test-file.txt"); - var spy = sinon.stub(fs, "ensureFile", function(arg1,arg2,arg3,arg4) { arg2(null); }); + var spy = sinon.stub(fs, "ensureDir", function(arg1,arg2,arg3,arg4) { arg2(null); }); var flow = [{id:"fileNode1", type:"file", name: "fileNode", "filename":fileToTest2, "appendNewline":true, "overwriteFile":true, "createDir":true}]; helper.load(fileNode, flow, function() { @@ -355,7 +363,7 @@ describe('file Nodes', function() { done(); } catch(e) { done(e); } - finally { fs.ensureFile.restore(); } + finally { fs.ensureDir.restore(); } },wait); n1.receive({payload:"test2"}); }); @@ -371,7 +379,7 @@ describe('file Nodes', function() { var wait = 150; beforeEach(function(done) { - fs.writeFileSync(fileToTest, "File message line 1\File message line 2\n"); + fs.writeFileSync(fileToTest, "File message line 1\nFile message line 2\n"); helper.startServer(done); }); @@ -392,60 +400,82 @@ describe('file Nodes', function() { }); it('should read in a file and output a buffer', function(done) { - var flow = [{id:"fileInNode1", type:"file in", name: "fileInNode", "filename":fileToTest, "format":"", wires:[["n2"]]}, + var flow = [{id:"fileInNode1", type:"file in", name:"fileInNode", "filename":fileToTest, "format":"", wires:[["n2"]]}, {id:"n2", type:"helper"}]; helper.load(fileNode, flow, function() { var n1 = helper.getNode("fileInNode1"); var n2 = helper.getNode("n2"); n2.on("input", function(msg) { msg.should.have.property('payload'); - msg.payload.should.have.length(39).and.be.a.Buffer; - msg.payload.toString().should.equal("File message line 1\File message line 2\n"); + msg.payload.should.have.length(40).and.be.a.Buffer; + msg.payload.toString().should.equal('File message line 1\nFile message line 2\n'); done(); }); n1.receive({payload:""}); }); }); -// Commented out to make build pass on node v.0.8 - reinstate when we drop 0.8 support... - //it('should read in a file and output a utf8 string', function(done) { - //var flow = [{id:"fileInNode1", type:"file in", name: "fileInNode", "filename":fileToTest, "format":"utf8", wires:[["n2"]]}, - //{id:"n2", type:"helper"}]; - //helper.load(fileNode, flow, function() { - //var n1 = helper.getNode("fileInNode1"); - //var n2 = helper.getNode("n2"); - //n2.on("input", function(msg) { - //msg.should.have.property('payload'); - //msg.payload.should.have.length(39).and.be.a.string; - //msg.payload.should.equal("File message line 1\File message line 2\n"); - //done(); - //}); - //n1.receive({payload:""}); - //}); - //}); + it('should read in a file and output a utf8 string', function(done) { + var flow = [{id:"fileInNode1", type:"file in", name: "fileInNode", "filename":fileToTest, "format":"utf8", wires:[["n2"]]}, + {id:"n2", type:"helper"}]; + helper.load(fileNode, flow, function() { + var n1 = helper.getNode("fileInNode1"); + var n2 = helper.getNode("n2"); + n2.on("input", function(msg) { + msg.should.have.property('payload'); + msg.payload.should.have.length(40).and.be.a.string; + msg.payload.should.equal("File message line 1\nFile message line 2\n"); + done(); + }); + n1.receive({payload:""}); + }); + }); -// Commented out as we no longer need to warn of the very old deprecated behaviour - //it('should warn if msg.props try to overide', function(done) { - //var flow = [{id:"fileInNode1", type:"file in", name: "fileInNode", "filename":fileToTest, "format":"", wires:[["n2"]]}, - //{id:"n2", type:"helper"}]; - //helper.load(fileNode, flow, function() { - //var n1 = helper.getNode("fileInNode1"); - //var n2 = helper.getNode("n2"); - //n2.on("input", function(msg) { - //msg.should.have.property('payload'); - //msg.payload.should.have.length(39).and.be.a.Buffer; - //msg.payload.toString().should.equal("File message line 1\File message line 2\n"); - //var logEvents = helper.log().args.filter(function(evt) { - //return evt[0].type == "file in"; - //}); - //logEvents.should.have.length(1); - //logEvents[0][0].should.have.a.property('msg'); - //logEvents[0][0].msg.toString().should.startWith("file.errors.nooverride"); - //done(); - //}); - //n1.receive({payload:"",filename:"foo.txt"}); - //}); - //}); + it('should read in a file and output split lines with parts', function(done) { + var flow = [{id:"fileInNode1", type:"file in", name: "fileInNode", filename:fileToTest, format:"lines", wires:[["n2"]]}, + {id:"n2", type:"helper"}]; + helper.load(fileNode, flow, function() { + var n1 = helper.getNode("fileInNode1"); + var n2 = helper.getNode("n2"); + var c = 0; + n2.on("input", function(msg) { + msg.should.have.property('payload'); + msg.payload.should.have.length(19).and.be.a.string; + if (c === 0) { + msg.payload.should.equal("File message line 1"); + c++; + } else { + msg.payload.should.equal("File message line 2"); + msg.should.have.property('parts'); + msg.parts.should.have.property('index',1); + msg.parts.should.have.property('count',2); + msg.parts.should.have.property('type','string'); + msg.parts.should.have.property('ch','\n'); + done(); + } + }); + n1.receive({payload:""}); + }); + }); + + it('should read in a file and output a buffer with parts', function(done) { + var flow = [{id:"fileInNode1", type:"file in", name: "fileInNode", filename:fileToTest, format:"stream", wires:[["n2"]]}, + {id:"n2", type:"helper"}]; + helper.load(fileNode, flow, function() { + var n1 = helper.getNode("fileInNode1"); + var n2 = helper.getNode("n2"); + n2.on("input", function(msg) { + msg.should.have.property('payload'); + msg.payload.should.have.length(40).and.be.a.Buffer; + msg.should.have.property('parts'); + msg.parts.should.have.property('count',1); + msg.parts.should.have.property('type','buffer'); + msg.parts.should.have.property('ch',''); + done(); + }); + n1.receive({payload:""}); + }); + }); it('should warn if no filename set', function(done) { var flow = [{id:"fileInNode1", type:"file in", name: "fileInNode", "format":""}]; @@ -474,8 +504,7 @@ describe('file Nodes', function() { }); logEvents.should.have.length(1); logEvents[0][0].should.have.a.property('msg'); - //logEvents[0][0].msg.toString().should.equal("Error: ENOENT, open 'badfile'"); - logEvents[0][0].msg.toString().should.startWith("Error: ENOENT"); + logEvents[0][0].msg.toString().should.startWith("Error"); done(); },wait); n1.receive({payload:""}); @@ -483,5 +512,4 @@ describe('file Nodes', function() { }); }); - });