smartstream/test/backpressure.ts
2023-11-12 22:34:55 +01:00

57 lines
1.6 KiB
TypeScript

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