node.js - framework - ¿Cuál es una forma correcta de pausar la secuencia legible por canales de uno grabable en nodejs?

node.js ejemplos

Question

Estoy escribiendo un módulo, que es una secuencia de escritura. Quiero implementar la interfaz de tuberías para mis usuarios.

Si ocurre algún error, necesito pausar el flujo de lectura y emitir un evento de error. Entonces, el usuario decidirá: si está bien con el error, debería poder reanudar el procesamiento de datos.

var writeable = new BackPressureStream();
writeable.on(''error'', function(error){
    console.log(error);
    writeable.resume();
});

var readable = require(''fs'').createReadStream(''somefile.txt'');
readable.pipe.(writeable);

Veo que el nodo nos proporciona el método readable.pause() , que se puede usar para pausar la transmisión legible. Pero no puedo entender cómo puedo llamarlo desde mi módulo de flujo de escritura:

var Writable = require(''stream'').Writable;

function BackPressureStream(options) {
    Writable.call(this, options);
}
require(''util'').inherits(BackPressureStream, Writable);

BackPressureStream.prototype._write = function(chunk, encoding, done) {
    done();
};

BackPressureStream.prototype.resume = function() {
    this.emit(''drain'');
}

¿Cómo se puede implementar la contrapresión en una secuencia de escritura?

PS: es posible usar eventos pipe/unpipe , que proporcionan flujo legible como parámetro. Pero también se dice que, para las transmisiones por cañería, la única posibilidad de pausar es anular la transmisión legible de la escritura.

¿Lo entendí bien? Tengo que borrar mi secuencia de escritura hasta que reanuden las llamadas de los usuarios? Y después de reanudar las llamadas de los usuarios, ¿debo volver a conectar la transmisión legible?

Answer 1

Básicamente, según tengo entendido, estás buscando poner presión en la transmisión en el caso de un evento de error. Tienes unas cuantas opciones.

En primer lugar, como ya lo ha identificado, use la pipe para tomar una instancia de la secuencia de lectura y realice un trabajo de pies elegante.

Otra opción es crear una secuencia de escritura modificable que proporcione esta funcionalidad (es decir, toma WritableStream como entrada, y al implementar las funciones de transmisión, pasa los datos a la secuencia suministrada.

Básicamente terminas con algo así como

source stream -> wrapping writable -> writable

https://nodejs.org/api/stream.html#stream_implementing_a_writable_stream trata sobre la implementación de una secuencia de escritura.

La clave para usted es que si se produce un error en la escritura subyacente, establecería un indicador en la transmisión, y la siguiente llamada para write ocurriría, almacenaría el fragmento, almacenará la devolución de llamada y solo llamará. Algo como

// ...
constructor(wrappedWritableStream) {
    wrappedWritableStream.on(''error'', this.errorHandler);
    this.wrappedWritableStream = wrappedWritableStream;
}
// ...
write(chunk, encoding, callback) {
    if (this.hadError) {
        // Note: until callback is called, this function won''t be called again, so we will have maximum one stored
        //  chunk.
        this.bufferedChunk = [chunk, encoding, callback];
    } else {
        wrappedWritableStream.write(chunk, encoding, callback);
    }
}
// ...
errorHandler(err) {
    console.error(err);
    this.hadError = err;
    this.emit(err);
}
// ...
recoverFromError() {
    if (this.bufferedChunk) {
        wrappedWritableStream.write(...this.bufferedChunk);
        this.bufferedChunk = undefined;
    }
    this.hadError = false;
}

Nota: Solo debe implementar la función de write , pero lo aliento a que explore y juegue con las otras funciones de implementación.

También vale la pena señalar que puede tener algunos problemas para escribir en transmisiones que han emitido un evento de error, pero lo dejo como un problema separado para resolver.

Aquí hay otro buen recurso sobre contrapresión https://nodejs.org/en/docs/guides/backpressuring-in-streams/