Pruebas de contrapresión de Transform. De alguna manera, con este código simple, el drenaje no funciona. Simplemente colgó. Cualquier ayuda es apreciada.

El código está debajo, si ejecuta el código, notará que se colgó.

flujo const = require('flujo'); const pipelineAsync = require('util').promisify(stream.pipeline);

 class MyTransform extends stream.Transform { constructor() { super({ objectMode: true, writableHighWaterMark: 10, }); } _transform(chunk, enc, cb) { console.log(`transform`, chunk) const nextData = `T${chunk}`; if (this.push(nextData)) { return cb(); } console.log('!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! back pressured') this.once('drain', () => { console.log('drained======================================> This never fires on time') cb(); }) /* //this is bad as we can't gurranty wait is long enough (or too long and cause performance) setTimeout(() => { console.log('delayed cb !!!!!!!!!!!!!!!!!!!!') cb(); }, 1000); */ } } function getConsumer() { return new stream.Writable({ objectMode: true, write(chunk, enc, cb) { console.log(`======> Finish chunk`, chunk); setTimeout(() => { cb(); }, 100); } }) } function getAry() { const res = []; for (let i = 0; i < 100; i++) res.push(i); return res; } async function test() { await pipelineAsync( stream.Readable.from(getAry()) , new MyTransform() , getConsumer()); console.log('all produced'); } test();