Node.js Streams

Por que streams são importantes

Carregar um arquivo grande na memória funciona até que o arquivo seja maior do que a memória disponível. As Streams resolvem isso processando os dados em pedaços (chunks): você lê uma parte, a processa e segue adiante, mantendo o uso de memória aproximadamente constante, independentemente do tamanho da entrada.

Streams estão em todo lugar no Node.js. Corpos de requisições e respostas HTTP, leitura e escrita de arquivos, compressão, criptografia e diversos parsers são todos streams. Compreendê-las é o que permite construir servidores e ferramentas que lidam com volumes de dados ilimitados sem travar.

Os quatro tipos de streams

Toda stream pertence a um destes quatro tipos:

• Readable — uma fonte de onde você lê dados. Arquivos, corpos de requisições HTTP, sockets e process.stdin.
• Writable — um destino para onde você escreve dados. Arquivos, respostas HTTP, sockets e process.stdout.
• Duplex — tanto readable quanto writable, como um socket TCP.
• Transform — uma stream duplex que modifica os dados enquanto eles passam, como zlib.createGzip() ou um parser de CSV.

Streams readable e writable se conectam em pipelines, e as transforms ficam posicionadas no meio.

Leitura e escrita

A maneira mais simples de consumir um readable stream é através de iteração assí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 writable streams, chame write() e sinalize o término com 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();

O stream faz o buffer das escritas internamente e as descarrega (flush) de forma eficiente, portanto, você não precisa gerenciar os chunks manualmente.

Piping e pipeline

Um pipeline conecta um readable, zero ou mais transforms e um 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"),
);

Sempre prefira pipeline em vez do encadeamento manual de .pipe(). O pipeline encaminha erros de cada etapa, destrói os streams em caso de falha e retorna uma promise, o que torna o tratamento de erros muito mais simples. O .pipe() manual não propaga erros, portanto, uma falha no início do fluxo pode deixar o pipeline travado e o destino aberto.

Backpressure

Streams possuem buffers internos. Quando você escreve dados em uma velocidade maior do que o destino consegue consumir, o buffer enche e write() retorna 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();
  });
}

Aguardar por drain antes de escrever mais dados é a forma como um produtor respeita um consumidor lento. pipeline e pipe lidam com isso automaticamente, o que é mais um motivo para utilizá-los.

Transform streams

Um transform stream aplica uma função a cada chunk. Você pode criar o seu próprio.

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

O Node fornece transforms úteis em node:zlib (gzip, deflate), node:crypto (ciphers e hashes) e em diversas bibliotecas, como parsers de CSV e JSON.

Object mode

Por padrão, as streams transportam bytes. O Object mode permite que elas transportem objetos JavaScript, o que é ideal para pipelines estruturados.

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

No object mode, highWaterMark conta objetos em vez de bytes. É assim que são construídas as streams de linhas de banco de dados, processadores de log e pipelines de ETL.

Padrões comuns

• Copiar um arquivo: pipeline(createReadStream(src), createWriteStream(dest)).
• Comprimir ou criptografar: insira createGzip() ou um transform de cifra.
• Transmitir uma resposta HTTP: direcione (pipe) um arquivo ou o resultado de uma query diretamente para res.
• Analisar dados delimitados por linha: use um transform que realize a divisão por quebras de linha.
• Progresso de upload: conte os bytes em um transform conforme eles passam.

Melhores práticas

• Prefira pipeline em vez de .pipe() para tratamento de erros e limpeza.
• Faça o stream de arquivos e respostas grandes em vez de carregá-los em buffer.
• Respeite o backpressure; nunca ignore quando write() retornar false.
• Ajuste o highWaterMark apenas após realizar medições.
• Use o object mode para dados estruturados.
• Trate error em todo stream que você criar.
• Destrua os streams em caso de falha para que os file descriptors sejam liberados.

Erros comuns

• Usar readFile para dados volumosos ou sem limite (unbounded).
• Encadear .pipe() e perder erros.
• Ignorar o backpressure e fazer o buffering de dados sem limite.
• Esquecer que um callback de transform deve ser chamado exatamente uma vez.
• Misturar encodings e gerar saídas corrompidas.
• Deixar streams abertas após um erro.

Próximos passos

Streams são a forma como o Node.js lida com dados em escala. Coloque-os em prática com o guia de File System e entenda o agendamento por trás disso no guia de Event Loop. Depois, reescreva uma chamada readFile como um stream e observe o consumo de memória estabilizar.