stream/promises.pipeline

stream/promises.pipeline conecta Readable → Transform → Writable con la propagación correcta de errores, la destrucción del stream en caso de fallo y una API de Promesas adecuada para manejadores async/await.

Receta

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

Cuándo usarlo:

• Cualquier cadena de procesamiento de archivos o HTTP de varias etapas
• Reemplazar el "spaghetti" de .on('error') en pipes manuales
• Endpoints de descarga de Express/Fastify que transmiten archivos
• Scripts ETL ejecutados mediante cron o procesos de worker

Ejemplo de trabajo

import { pipeline } from 'node:stream/promises';
import { createReadStream } from 'node:fs';
import { createServer } from 'node:http';
import { Transform } from 'node:stream';
import { createGzip } from 'node:zlib';
 
const lineCount = new Transform({
  transform(chunk, _enc, cb) {
    const lines = String(chunk).split('\n').length - 1;
    (this as Transform & { lines?: number }).lines =
      ((this as Transform & { lines?: number }).lines ?? 0) + lines;
    cb(null, chunk);
  },
});
 
const server = createServer(async (req, res) => {
  if (req.url !== '/export') {
    res.writeHead(404).end();
    return;
  }
 
  try {
    res.writeHead(200, {
      'content-type': 'application/gzip',
      'content-disposition': 'attachment; filename="export.log.gz"',
    });
 
    await pipeline(
      createReadStream('app.log'),
      lineCount,
      createGzip(),
      res,
    );
    console.log('lines processed', (lineCount as Transform & { lines?: number }).lines);
  } catch (err) {
    if (!res.headersSent) res.writeHead(500);
    res.end();
    console.error('pipeline failed', err);
  }
});
 
server.listen(3000);

Lo que esto demuestra:

• pipeline acepta el res Writable de HTTP como sumidero final
• Los errores después de headersSent requieren finalizar la respuesta sin una segunda línea de estado
• Un Transform intermedio puede acumular métricas mientras pasa bytes
• await se integra con try/catch como cualquier E/S asíncrona

Análisis profundo

Cómo funciona

• En caso de error, pipeline destruye los streams participantes con el error.
• Al finalizar, los streams se cierran en orden: Writable recibe end.
• Soporta AbortSignal en Node 24 para la cancelación en pleno vuelo.
• Existe una versión con callback en node:stream; la variante de Promesa es preferida en código asíncrono.

pipeline vs pipe

Notas de TypeScript

import { pipeline } from 'node:stream/promises';
import type { Readable, Writable } from 'node:stream';
 
export async function safePump(
  source: Readable,
  sink: Writable,
  signal?: AbortSignal,
): Promise<void> {
  await pipeline(source, sink, { signal });
}

Errores comunes

• No capturar el rechazo de pipeline - el rechazo no manejado bloquea el proceso. Solución: try/catch en los manejadores de ruta.
• Cabeceras después de que el piping ha comenzado - debes writeHead antes de pipeline para enviar la respuesta. Solución: establece las cabeceras primero.
• Aborto del cliente a mitad del stream - trátalo como no fatal; escucha req.on('aborted') opcionalmente. Solución: pasa AbortSignal vinculado a la solicitud.
• Mezclar pipeline y escritura manual en el mismo Writable - salida corrupta. Solución: un pipeline por respuesta.
• Transform no vacía el estado parcial - implementa _flush para los datos almacenados en búfer restantes.

Alternativas

Preguntas frecuentes

Relacionado

• Conceptos básicos de Streams - introducción a pipe
• Contrapresión - control de flujo
• Respuestas HTTP en streaming - proxies y cabeceras
• Mejores prácticas de Streams - la regla de siempre usar pipeline

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