form - Escribiendo archivos grandes con Node.js

node js upload file

Question

Estoy escribiendo un archivo grande con node.js usando una secuencia de escritura :

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] );
    }
}

Me pregunto si este esquema es seguro sin usar drain evento de drain . Si no lo es (lo que creo que es el caso), ¿cuál es el patrón para escribir datos grandes arbitrarios en un archivo?

Answer 1

Así es como finalmente lo hice. La idea subyacente es crear un flujo legible implementando la interfaz ReadStream y luego usar el método pipe() para canalizar los datos a un flujo grabable.

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!'');
});

El ejemplo de la clase MyReadStream se puede tomar de QueryStream mangosta.

Answer 2

Descubrí que las transmisiones son una forma de rendimiento deficiente para tratar archivos grandes, porque no se puede establecer un tamaño de búfer de entrada adecuado (al menos no tengo conocimiento de una buena forma de hacerlo). Esto es lo que hago:

var fs = require(''fs'');

var i = fs.openSync(''input.txt'', ''r'');
var o = fs.openSync(''output.txt'', ''w'');

var buf = new Buffer(1024 * 1024), len, prev = '''';

while(len = fs.readSync(i, buf, 0, buf.length)) {

    var a = (prev + buf.toString(''ascii'', 0, len)).split(''/n'');
    prev = len === buf.length ? ''/n'' + a.splice(a.length - 1)[0] : '''';

    var out = '''';
    a.forEach(function(line) {

        if(!line)
            return;

        // do something with your line here

        out += line + ''/n'';
    });

    var bout = new Buffer(out, ''ascii'');
    fs.writeSync(o, bout, 0, bout.length);
}

fs.closeSync(o);
fs.closeSync(i);

Answer 3

La forma más limpia de manejar esto es hacer que su generador de línea sea un flujo legible , llamémoslo lineReader . Luego, lo siguiente manejaría automáticamente los búferes y el drenaje de forma agradable para usted:

lineReader.pipe(fs.createWriteStream(''someFile.txt''));

Si no desea crear un flujo legible, puede escuchar la salida de write para completar el búfer y responder así:

var i = 0, n = lines.length;
function write () {
  if (i === n) return;  // A callback could go here to know when it''s done.
  while (stream.write(lines[i++]) && i < n);
  stream.once(''drain'', write);
}
write();  // Initial call.

Un ejemplo más largo de esta situación se puede encontrar here .

Answer 4

La idea detrás del drenaje es que lo usarías para probar aquí:

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
    }
}

que no eres Así que necesitarías un rearchitect para hacerlo "reentrante".

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
        }
    }
}

Sin embargo, ¿significa esto que debe mantener el búfer getLines también mientras espera?

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);
      }
    }
  }
});

Answer 5

Si no tiene un flujo de entrada, no puede usar tuberías fácilmente. Nada de lo anterior funcionó para mí, el evento de drenaje no se dispara. Resuelto de la siguiente manera (basado en la respuesta de Tylers):

var lines[]; // some very large array
var i = 0;

function write() {
    if (i < lines.length)  {
        wstream.write(lines[i]), function(err){
            if (err) {
                console.log(err);
            } else {
                i++;
                write();
            }
        });
    } else {
        wstream.end();
        console.log("done");
    }
};
write();

Answer 6

Varias respuestas sugeridas a esta pregunta han perdido el punto sobre las transmisiones por completo.

Este módulo puede ayudar a https://www.npmjs.org/package/JSONStream

Sin embargo, supongamos la situación como se describe y escribamos el código nosotros mismos. Estás leyendo de un MongoDB como una secuencia, con ObjectMode = true de forma predeterminada.

Esto dará lugar a problemas si intenta transmitir directamente al archivo, algo así como el error "Invalid non-string / buffer chunk".

La solución a este tipo de problema es muy simple.

Simplemente coloque otra Transformación entre lo legible y lo que se puede escribir para adaptar el Objeto legible a una Cadena que se pueda escribir adecuadamente.

Solución de código de muestra:

var fs = require(''fs''),
    writeStream = fs.createWriteStream(''./out'' + process.pid, {flags: ''w'', encoding: ''utf-8'' }),
    stream = require(''stream''),
    stringifier = new stream.Transform();
stringifier._writableState.objectMode = true;
stringifier._transform = function (data, encoding, done) {
    this.push(JSON.stringify(data));
    this.push(''/n'');
    done();
}
rowFeedDao.getRowFeedsStream(merchantId, jobId)
.pipe(stringifier)
.pipe(writeStream).on(''error'', function (err) {
   // handle error condition
}

Answer 7

[Editar] La documentación actualizada de Node.js writable.write(...) API dice:

[El] valor de retorno es estrictamente de asesoramiento. PUEDES continuar escribiendo, incluso si devuelve falso. Sin embargo, las escrituras se almacenarán en la memoria intermedia, por lo que es mejor no hacer esto excesivamente. En su lugar, espere el evento de drenaje antes de escribir más datos.

[Original] De la documentación de stream.write(...) (el énfasis es mío):

Devuelve true si la cadena se ha vaciado al búfer del núcleo. Devuelve false para indicar que el búfer del kernel está lleno y que los datos se enviarán en el futuro .

Interpreto que esto significa que la función de "escritura" devuelve true si la cadena dada se escribió inmediatamente en el búfer del sistema operativo subyacente o false si aún no se ha escrito, pero la función de escritura la escribirá (por ejemplo, probablemente el búfer para usted WriteStream) para que no tenga que volver a llamar "escribir".