Node.js Streams

Por qué los streams son importantes

Cargar un archivo grande en memoria funciona hasta que el archivo es más grande que la memoria disponible. Los streams solucionan esto procesando los datos en fragmentos (chunks): lees una parte, la procesas y continúas, de modo que el uso de memoria se mantiene prácticamente constante sin importar cuán grande sea la entrada.

Los streams están en todas partes en Node.js. Los cuerpos de las solicitudes y respuestas HTTP, la lectura y escritura de archivos, la compresión, el cifrado y muchos parsers son streams. Comprenderlos es lo que te permite construir servidores y herramientas capaces de manejar datos ilimitados sin que el sistema colapse.

Los cuatro tipos de streams

Cada stream pertenece a una de estas cuatro categorías:

• Readable — una fuente de la cual se lee. Archivos, cuerpos de peticiones HTTP, sockets y process.stdin.
• Writable — un destino donde se escribe. Archivos, respuestas HTTP, sockets y process.stdout.
• Duplex — tanto readable como writable, como un socket TCP.
• Transform — un stream duplex que modifica los datos a medida que pasan, como zlib.createGzip() o un parser de CSV.

Los streams readable y writable se conectan mediante pipelines, y los transforms se ubican en medio de ellos.

Lectura y escritura

La forma más sencilla de consumir un readable stream es mediante la iteración asíncrona.

// read.js
import { createReadStream } from "node:fs";

const stream = createReadStream("access.log", { encoding: "utf8" });

for await (const chunk of stream) {
  process.stdout.write(chunk);
}

Para los writable streams, llama a write() y señala el final con end().

// write.js
import { createWriteStream } from "node:fs";

const out = createWriteStream("out.txt");

out.write("first line\n");
out.write("second line\n");
out.end();

El stream almacena las escrituras en un buffer interno y las vuelca de manera eficiente, por lo que no tienes que gestionar los chunks tú mismo.

Piping y pipeline

Un pipeline conecta un readable, cero o más transforms y un writable.

// compress.js
import { createReadStream, createWriteStream } from "node:fs";
import { pipeline } from "node:stream/promises";
import { createGzip } from "node:zlib";

await pipeline(
  createReadStream("access.log"),
  createGzip(),
  createWriteStream("access.log.gz"),
);

Prefiere siempre pipeline sobre el encadenamiento manual de .pipe(). pipeline reenvía los errores de cada etapa, destruye los streams en caso de fallo y devuelve una promesa, lo que simplifica el manejo de errores. El .pipe() manual no propaga los errores, por lo que un fallo upstream puede dejar el pipeline colgado y el destino abierto.

Backpressure

Los streams tienen buffers internos. Cuando escribes más rápido de lo que el destino puede consumir, el buffer se llena y write() devuelve false.

// backpressure.js
function writeAll(stream, chunks) {
  return new Promise((resolve, reject) => {
    let i = 0;
    const next = () => {
      while (i < chunks.length) {
        const ok = stream.write(chunks[i++]);
        if (!ok) {
          stream.once("drain", next);
          return;
        }
      }
      stream.end(resolve);
    };
    stream.on("error", reject);
    next();
  });
}

Esperar a drain antes de escribir más es la forma en que un productor respeta a un consumidor lento. pipeline y pipe gestionan esto automáticamente, lo cual es otra razón más para utilizarlos.

Transform streams

Un transform stream aplica una función a cada chunk. Puedes construir los tuyos propios.

// upper.js
import { Transform } from "node:stream";

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

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

Node incluye transforms útiles en node:zlib (gzip, deflate), node:crypto (ciphers y hashes) y en muchas librerías, como parsers de CSV y JSON.

Modo objeto

Por defecto, los streams transportan bytes. El modo objeto permite que transporten objetos JavaScript en su lugar, lo cual es ideal para pipelines estructurados.

// object-mode.js
import { Readable, Transform } from "node:stream";

Readable.from([{ id: 1 }, { id: 2 }])
  .pipe(
    new Transform({
      objectMode: true,
      transform(record, _encoding, callback) {
        callback(null, { ...record, seen: true });
      },
    }),
  )
  .on("data", console.log);

En modo objeto, highWaterMark cuenta objetos en lugar de bytes. Así es como se construyen los streams de filas de bases de datos, los procesadores de logs y los pipelines de ETL.

Patrones comunes

• Copiar un archivo: pipeline(createReadStream(src), createWriteStream(dest)).
• Comprimir o cifrar: insertar createGzip() o un transform de cifrado.
• Transmitir una respuesta HTTP: redirigir (pipe) un archivo o el resultado de una consulta directamente a res.
• Parsear datos delimitados por líneas: usar un transform que divida el contenido por saltos de línea.
• Progreso de subida: contar los bytes en un transform a medida que pasan.

Mejores prácticas

• Prefiere pipeline sobre .pipe() para el manejo de errores y la limpieza.
• Transmite archivos y respuestas grandes mediante streams en lugar de almacenarlos en búfer.
• Respeta la contrapresión (backpressure); nunca ignores cuando write() devuelva false.
• Ajusta highWaterMark solo después de realizar mediciones.
• Usa el modo de objeto para datos estructurados.
• Gestiona error en cada stream que crees.
• Destruye los streams en caso de fallo para que se liberen los descriptores de archivo.

Errores comunes

• Usar readFile para datos extensos o sin límite.
• Encadenar .pipe() y perder los errores.
• Ignorar el backpressure y almacenar en búfer datos sin límite.
• Olvidar que un callback de transform debe llamarse exactamente una vez.
• Mezclar codificaciones y generar una salida corrupta.
• Dejar streams abiertos después de un error.

Próximos pasos

Los streams son la forma en que Node maneja datos a escala. Ponlos en práctica con la guía del File System y comprende la programación subyacente en la guía del Event Loop. Después, reescribe una llamada a readFile como un stream y observa cómo se estabiliza el uso de memoria.