stream/promises.pipeline

stream/promises.pipeline conecta Readable → Transform → Writable com propagação de erros correta, destruição de stream em caso de falha e uma API de Promise adequada para manipuladores async/await.

Receita

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'),
);

Quando usar isso:

• Qualquer cadeia de processamento de arquivos ou HTTP com múltiplos estágios
• Substituindo o "espaguete" de .on('error') em pipes manuais
• Endpoints de download do Express/Fastify que transmitem arquivos
• Scripts ETL executados via cron ou processos worker

Exemplo de Trabalho

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('linhas processadas', (lineCount as Transform & { lines?: number }).lines);
  } catch (err) {
    if (!res.headersSent) res.writeHead(500);
    res.end();
    console.error('pipeline falhou', err);
  }
});
 
server.listen(3000);

O que isso demonstra:

• pipeline aceita res HTTP Writable como destino final
• Erros após headersSent exigem o encerramento da resposta sem uma segunda linha de status
• Transformações intermediárias podem acumular métricas enquanto passam bytes
• await integra-se com try/catch como qualquer I/O assíncrono

Mergulho Profundo

Como Funciona

• Em caso de erro, pipeline destrói os streams participantes com o erro.
• Ao concluir, os streams são fechados em ordem - Writable recebe end.
• Suporta AbortSignal no Node 24 para cancelamento em andamento.
• Existe uma versão de callback em node:stream - a variante Promise é preferida em código async.

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 });
}

Armadilhas

• Não capturar a rejeição do pipeline - rejeição não tratada causa falha no processo. Correção: use try/catch em manipuladores de rota.
• Cabeçalhos após o início do pipe - é preciso writeHead antes de pipeline para o destino da resposta. Correção: defina os cabeçalhos primeiro.
• Abandono do cliente no meio do stream - trate como não fatal; ouça req.on('aborted') opcionalmente. Correção: passe AbortSignal vinculado à requisição.
• Misturar pipeline e escrita manual no mesmo Writable - saída corrompida. Correção: um pipeline por resposta.
• Transform não descarrega estado parcial - implemente _flush para dados em buffer pendentes.

Alternativas

FAQs

Relacionados

• Fundamentos de Streams - introdução ao pipe
• Backpressure - controle de fluxo
• Streaming de Respostas HTTP - proxies e cabeçalhos
• Melhores Práticas de Streams - regra de sempre usar pipeline

Versões da Pilha: Esta página foi escrita para Node.js 24.18.0 (Active LTS), npm 10+, TypeScript 5.6+, Express 5, Fastify 5 e NestJS 11.