import { SmartDuplex, type ISmartDuplexOptions, StreamWrapper } from '../ts/index.js'; import * as smartdelay from '@push.rocks/smartdelay'; async function testBackpressure() { const stream1 = new SmartDuplex({ objectMode: true, handleBackpressure: true, writeFunction: async (chunk, tools) => { console.log(`processed chunk ${chunk} in stream 1`); return chunk; // Fast processing } }); const stream2 = new SmartDuplex({ objectMode: true, handleBackpressure: true, writeFunction: async (chunk, tools) => { await new Promise(resolve => setTimeout(resolve, 100)); // Slow processing console.log(`processed chunk ${chunk} in stream 2`); return chunk; } }); // This stream processes data more slowly const stream3 = new SmartDuplex({ handleBackpressure: false, writeFunction: async (chunk, tools) => { console.log(`finished chunk ${chunk} in stream 3`); } }); stream1.pipe(stream2).pipe(stream3); let backpressured = false; for (let i = 1; i < 100; i++) { const canContinue = stream1.write(`Chunk ${i}`, 'utf8'); if (!canContinue) { backpressured = true; console.log(`Backpressure at chunk ${i}`); } } stream1.end(); stream1.on('finish', () => { console.log('Stream 1 finished processing.'); }); stream2.on('finish', () => { console.log('Stream 2 finished processing.'); }); stream3.on('finish', () => { console.log('Stream 3 finished processing.'); if (!backpressured) { console.log('No backpressure was observed.'); } }); } testBackpressure();