bare-stream

import { Readable, Writable, Transform } from 'bare-stream'; class MyGenerator extends Readable { constructor(limit) { super(); this.index = 0; this.limit = limit; } _read(size) { if (this.index < this.limit) { const chunk = `Data chunk ${this.index++}\n`; this.push(Buffer.from(chunk)); } else { this.push(null); // Signal end of stream } } } class MyProcessor extends Transform { _transform(chunk, encoding, callback) { // Convert to uppercase and add a prefix this.push(Buffer.from(`PROCESSED: ${chunk.toString().toUpperCase()}`)); callback(); } } class MyConsumer extends Writable { _write(chunk, encoding, callback) { console.log(chunk.toString().trim()); callback(); } } const generator = new MyGenerator(3); const processor = new MyProcessor(); const consumer = new MyConsumer(); generator.pipe(processor).pipe(consumer); // Error handling is crucial for streams generator.on('error', (err) => console.error('Generator error:', err)); processor.on('error', (err) => console.error('Processor error:', err)); consumer.on('error', (err) => console.error('Consumer error:', err)); consumer.on('finish', () => console.log('All data processed.'));