I'm writing a large file with node.js using a writable stream:
var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', { flags : 'w' });
var lines;
while (lines = getLines()) {
for (var i = 0; i < lines.length; i++) {
stream.write( lines[i] );
}
}
I'm wondering if this scheme is safe without using drain event? If it is not (which I think is the case), what is the pattern for writing an arbitrary large data to a file?
That's how I finally did it. The idea behind is to create readable stream implementing ReadStream interface and then use pipe() method to pipe data to writable stream.
var fs = require('fs');
var writeStream = fs.createWriteStream('someFile.txt', { flags : 'w' });
var readStream = new MyReadStream();
readStream.pipe(writeStream);
writeStream.on('close', function () {
console.log('All done!');
});
The example of MyReadStream class can be taken from mongoose QueryStream.
The idea behind drain is that you would use it to test here:
var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', {flags: 'w'});
var lines;
while (lines = getLines()) {
for (var i = 0; i < lines.length; i++) {
stream.write(lines[i]); //<-- the place to test
}
}
which you're not. So you would need to rearchitect to make it "reentrant".
var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', {flags: 'w'});
var lines;
while (lines = getLines()) {
for (var i = 0; i < lines.length; i++) {
var written = stream.write(lines[i]); //<-- the place to test
if (!written){
//do something here to wait till you can safely write again
//this means prepare a buffer and wait till you can come back to finish
// lines[i] -> remainder
}
}
}
However, does this mean that you need to keep buffering getLines as well while you wait?
var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', {flags: 'w'});
var lines,
buffer = {
remainingLines = []
};
while (lines = getLines()) {
for (var i = 0; i < lines.length; i++) {
var written = stream.write(lines[i]); //<-- the place to test
if (!written){
//do something here to wait till you can safely write again
//this means prepare a buffer and wait till you can come back to finish
// lines[i] -> remainder
buffer.remainingLines = lines.slice(i);
break;
//notice there's no way to re-run this once we leave here.
}
}
}
stream.on('drain',function(){
if (buffer.remainingLines.length){
for (var i = 0; i < buffer.remainingLines.length; i++) {
var written = stream.write(buffer.remainingLines[i]); //<-- the place to test
if (!written){
//do something here to wait till you can safely write again
//this means prepare a buffer and wait till you can come back to finish
// lines[i] -> remainder
buffer.remainingLines = lines.slice(i);
}
}
}
});