Streams y Buffers

Por qué importan los streams

Imagina leer un archivo de 2GB. Sin streams tendrías que cargarlo completo en memoria antes de procesarlo. Con streams, procesas datos por partes mientras llegan, usando poca memoria.

Los streams son fundamentales en Node.js: requests/responses HTTP, lectura de archivos, compresión, cifrado y mucho más.

Buffers: datos binarios

Buffers son bloques de memoria fija usados para datos binarios. Son esenciales para streams, archivos y redes.

const buf1 = Buffer.alloc(10);
const buf2 = Buffer.from('Hola');
const buf3 = Buffer.from([72, 101, 108]);

console.log(buf2.toString()); // Hola
console.log(buf2.length);     // 4 bytes
console.log(buf2[0]);         // 72 (ASCII de H)

const combined = Buffer.concat([buf2, Buffer.from(' Mundo')]);
console.log(combined.toString()); // Hola Mundo

const base64 = buf2.toString('base64');
const hex = buf2.toString('hex');

const buf = Buffer.alloc(10);
buf.write('Hi');
console.log(buf.toString());

Readable streams

const fs = require('fs');

const readStream = fs.createReadStream('archivo-grande.txt', {
  encoding: 'utf8',
  highWaterMark: 64 * 1024
});

readStream.on('data', (chunk) => {
  console.log('Chunk:', chunk.length);
});

readStream.on('end', () => console.log('Listo'));
readStream.on('error', (err) => console.error(err));

readStream.on('data', (chunk) => {
  readStream.pause();
  processChunk(chunk).then(() => readStream.resume());
});

Writable streams

const fs = require('fs');
const writeStream = fs.createWriteStream('salida.txt');

writeStream.write('Hola ');
writeStream.write('Mundo\n');
writeStream.end('Fin');

writeStream.on('finish', () => console.log('Escritura completa'));
writeStream.on('error', (err) => console.error(err));

function writeData(stream, data) {
  const ok = stream.write(data);
  if (!ok) {
    stream.once('drain', () => console.log('Reanudando...'));
  }
}

Piping de streams

pipe conecta streams y maneja backpressure automáticamente.

const fs = require('fs');
const zlib = require('zlib');

fs.createReadStream('source.txt')
  .pipe(fs.createWriteStream('dest.txt'));

fs.createReadStream('file.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('file.txt.gz'));

Pipeline moderno

const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');

async function compressFile(input, output) {
  try {
    await pipeline(
      fs.createReadStream(input),
      zlib.createGzip(),
      fs.createWriteStream(output)
    );
    console.log('Compresión completa');
  } catch (err) {
    console.error('Pipeline falló:', err);
  }
}

Transform streams

const { Transform } = require('stream');

const upperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  }
});

process.stdin.pipe(upperCaseTransform).pipe(process.stdout);

Ejemplo práctico: CSV a JSON

const { Transform, pipeline } = require('stream');
const fs = require('fs');

class CSVParser extends Transform {
  constructor() {
    super({ objectMode: true });
    this.headers = null;
    this.buffer = '';
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();
    const lines = this.buffer.split('\n');
    this.buffer = lines.pop();

    for (const line of lines) {
      if (!line.trim()) continue;
      const values = line.split(',');
      if (!this.headers) { this.headers = values; continue; }
      const obj = {};
      this.headers.forEach((h, i) => { obj[h.trim()] = values[i]?.trim(); });
      this.push(obj);
    }
    callback();
  }
}

pipeline(
  fs.createReadStream('users.csv'),
  new CSVParser(),
  new Transform({
    objectMode: true,
    transform(obj, enc, cb) {
      cb(null, JSON.stringify(obj) + '\n');
    }
  }),
  fs.createWriteStream('users.json')
);

Comparación de rendimiento