Estoy trabajando en mi flujo de transformación personalizado. Dado que manejo el almacenamiento en búfer y el almacenamiento en caché manualmente, debo asegurarme de que no se envíen fragmentos automáticamente al volver a canalizarlo. Para lograrlo, debo borrar/restablecer el búfer interno (la cola de escritura) antes de intentar llamar a .pipe() .
Dado que mi lógica de transmisión es bastante complicada, hice este ejemplo simplificado para ayudarlo a comprender mejor lo que estoy tratando de resolver:
const { Readable, Writable } = require('stream') const wait = ms => new Promise(res => setTimeout(res, ms)) ;(async () => { const readable = new Readable() readable._read = () => {} const writable = new Writable() writable._write = (chunk, _encoding, callback) => { console.log(chunk.toString()) callback() } readable.pipe(writable) await wait(100) // This small delay replaces listening to the pipe/unpipe events readable.push('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa') console.log('unpipe') readable.unpipe(writable) await wait(100) readable.push('bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb') // readable.clearInternalBufferOrWhatever() // ??? console.log('pipe') readable.pipe(writable) await wait(100) readable.push('ccccccccccccccccccccccccccccccc') })() // => aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa (Expected ✔) // unpipe // pipe // bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb (Unexpected ❌) // ccccccccccccccccccccccccccccccc (Expected ✔) Como puede ver, la secuencia grabable recibió con éxito el segundo fragmento ( 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb' ) y eso es lo que estoy tratando de evitar. ¿Algunas ideas?
¡Me lo imaginé! Los flujos tienen una propiedad interna llamada _readableState que contiene el búfer interno. Es posible borrarlo con solo 2 líneas de código:
readable._readableState.buffer.clear() readable._readableState.length = 0 Tenga en cuenta que _readableState no es seguro ya que es interno y no está documentado y podría sufrir cambios importantes en cualquier momento. Sin embargo, se confirma que esta solución funciona en el nodo v8.10.0 y v14.17.2.
Aquí hay un ejemplo de trabajo completo:
const { Readable, Writable } = require('stream') const wait = ms => new Promise(res => setTimeout(res, ms)) ;(async () => { const readable = new Readable() readable._read = () => {} const writable = new Writable() writable._write = (chunk, _encoding, callback) => { console.log(chunk.toString()) callback() } readable.pipe(writable) await wait(100) readable.push('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa') console.log('unpipe') readable.unpipe(writable) await wait(100) readable.push('bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb') readable._readableState.buffer.clear() readable._readableState.length = 0 console.log('pipe') readable.pipe(writable) await wait(100) readable.push('ccccccccccccccccccccccccccccccc') })() // => aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa // unpipe // pipe // cccccccccccccccccccccccccccccccDe la documentación de flujo de Node en readable.read() :
Si no hay datos disponibles para leer, se devuelve
null.
Entonces, si seguimos llamando a readable.read() hasta que se devuelva null , entonces el búfer interno debe vaciarse.
function clearReadBuffer(readable) { while (true) { const chunk = readable.read(); if ( chunk === null ) return; } }