Conceptos Básicos de Streams

8 ejemplos para que empieces con los Streams - 6 básicos y 2 intermedios.

Prerrequisitos

• Node.js 24.18.0, import { Readable, Writable } from 'node:stream'.
• Lee Conceptos Básicos de Buffers para el manejo de fragmentos.

Ejemplos Básicos

1. Readable desde un Iterador

Genera fragmentos sin cargar todo en memoria.

import { Readable } from 'node:stream';
 
const readable = Readable.from(['chunk-a', 'chunk-b', 'chunk-c']);
 
for await (const chunk of readable) {
  console.log(chunk);
}

• Readable.from envuelve iterables e iterables asíncronos.
• for await...of consume hasta el final null.
• El modo predeterminado es byte/string a menos que objectMode: true.

Relacionado: Patrones Readable y Writable - modos

2. Conectar (Pipe) a Writable

Conecta el productor con el consumidor - el clásico encadenamiento de tuberías al estilo Unix.

import { createWriteStream } from 'node:fs';
import { Readable } from 'node:stream';
 
Readable.from(['line1\n', 'line2\n']).pipe(createWriteStream('out.txt'));

• pipe conecta data a write y maneja end.
• Los errores en cualquiera de los lados necesitan manejadores explícitos - prefiere pipeline en producción.
• pipe no reenvía errores automáticamente en patrones antiguos.

Relacionado: stream/promises.pipeline - encadenamiento seguro

3. fs.createReadStream

Transmite archivos grandes desde el disco.

import { createReadStream } from 'node:fs';
 
const stream = createReadStream('package.json', { encoding: 'utf8', highWaterMark: 64 * 1024 });
 
stream.on('data', (chunk) => console.log('chunk', chunk.length));
stream.on('end', () => console.log('done'));

• highWaterMark controla el tamaño del búfer interno por stream.
• La codificación produce cadenas; omítela para Buffers sin procesar.
• Prefiere pipeline a la respuesta HTTP para descargas.

4. Respuesta HTTP como Writable

ServerResponse es un stream Writable.

import { createServer } from 'node:http';
import { createReadStream } from 'node:fs';
 
createServer((req, res) => {
  createReadStream('package.json').pipe(res);
}).listen(3000);

• Los encabezados deben establecerse antes de conectar (piping) o mediante writeHead primero.
• La desconexión del cliente debe destruir el stream de lectura - pipeline maneja la limpieza.
• Consulta Streaming de Respuestas HTTP para proxies.

5. Transform con map (simple)

Cambia los datos a medida que pasan.

import { Transform } from 'node:stream';
 
const upper = new Transform({
  transform(chunk, _enc, cb) {
    cb(null, String(chunk).toUpperCase());
  },
});
 
Readable.from(['hello']).pipe(upper).pipe(process.stdout);

• Transform es tanto readable como writable - se sitúa en el medio de la tubería.
• La función de callback cb(err, data) señala que el fragmento ha sido procesado o que ha ocurrido un error.
• _flush se ejecuta al final para la salida final.

Relacionado: Transform y Duplex - tuberías de análisis

6. Stream en Modo Objeto

Transmite objetos JavaScript en lugar de bytes.

import { Transform } from 'node:stream';
 
const parseLines = new Transform({
  objectMode: true,
  transform(chunk, _enc, cb) {
    const lines = String(chunk).split('\n').filter(Boolean);
    for (const line of lines) this.push({ line });
    cb();
  },
});

• highWaterMark cuenta objetos en modo objeto, no bytes.
• Los consumidores deben esperar objetos en los eventos data.
• Útil para el análisis de filas CSV y tuberías de eventos de registro.

Ejemplos Intermedios

7. pipeline con Promesas

Propagación de errores y limpieza en una sola llamada.

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';
 
await pipeline(
  createReadStream('access.log'),
  createGzip(),
  createWriteStream('access.log.gz'),
);

• pipeline destruye los streams en caso de error o cierre anticipado.
• Devuelve una Promesa - usa try/catch en manejadores asíncronos.
• Preferido sobre las cadenas manuales .pipe().pipe().

Relacionado: stream/promises.pipeline - análisis profundo

8. Contrapresión con pausa/reanudación

Writable señala cuándo no puede aceptar más datos.

import { Readable, Writable } from 'node:stream';
 
const slow = new Writable({
  write(chunk, _enc, cb) {
    setTimeout(() => cb(), 100);
  },
});
 
const fast = Readable.from(['a', 'b', 'c', 'd']);
fast.pipe(slow); // pipe maneja la pausa/reanudación automáticamente

• Cuando write devuelve false, pausa el readable hasta drain.
• Los bucles de read manuales deben verificar el valor de retorno de writable.write.
• Consulta Contrapresión para la optimización en producción.

Versiones de la pila: Esta página fue escrita para Node.js 24.18.0 (LTS Activo), npm 10+, TypeScript 5.6+, Express 5, Fastify 5 y NestJS 11.